diff --git a/crates/openshell-supervisor-network/src/policy_dns/mod.rs b/crates/openshell-supervisor-network/src/policy_dns/mod.rs index 63fb8eeca2..bc2ea487f1 100644 --- a/crates/openshell-supervisor-network/src/policy_dns/mod.rs +++ b/crates/openshell-supervisor-network/src/policy_dns/mod.rs @@ -373,7 +373,10 @@ fn eligible_endpoints( continue; } let raw_allowed_ips = value_string_array(&endpoint.endpoint, "allowed_ips"); - let exact_declared_host = !pattern.contains('*') && pattern == name.as_str(); + let exact_declared_host = !pattern.contains('*') + && pattern == name.as_str() + && value_field(&endpoint.endpoint, "advisor_proposed") + != Some(®orus::Value::Bool(true)); let destination_plan = build_validation_plan( name.as_str(), name.as_str(), @@ -1126,6 +1129,68 @@ process: { run_as_user: sandbox, run_as_group: sandbox } assert!(matches!(result, Err(PolicyDnsError::NoValidAddress))); } + #[tokio::test] + async fn advisor_endpoints_filter_private_answers_without_losing_public_or_explicit_ips() { + for protocol in [ + "", + "tcp", + "rest, access: read-only", + "rest, access: read-only, tls: skip", + ] { + for (family, private, public) in [ + (AddressFamily::Ipv4, "10.2.3.4", "8.8.8.8"), + (AddressFamily::Ipv6, "fd12::4", "2001:4860:4860::8888"), + ] { + let private = private.parse::().unwrap(); + let public = public.parse::().unwrap(); + let yaml = BASE_POLICY.replace( + "protocol: tcp", + &format!("protocol: {protocol}, advisor_proposed: true"), + ); + let yaml = yaml.replace("protocol: , ", ""); + let private_only = service(&yaml, vec![private]); + assert!( + matches!( + private_only + .answer_query("db.example", family, Instant::now()) + .await, + Err(PolicyDnsError::NoValidAddress) + ), + "{protocol}: {private}" + ); + for addresses in [vec![public], vec![private, public]] { + let service = service(&yaml, addresses); + let now = Instant::now(); + let answer = service + .answer_query("db.example", family, now) + .await + .unwrap(); + let mapping = service + .store + .lookup(answer.address, 5432, answer.policy_generation, now) + .unwrap(); + assert_eq!(mapping.record.contracts[0].pinned_addresses, [public]); + } + // Explicit narrow address permission retains its usual precedence. + let yaml = yaml.replace( + "advisor_proposed: true", + &format!("advisor_proposed: true, allowed_ips: ['{private}']"), + ); + let service = service(&yaml, vec![private, public]); + let now = Instant::now(); + let answer = service + .answer_query("db.example", family, now) + .await + .unwrap(); + let mapping = service + .store + .lookup(answer.address, 5432, answer.policy_generation, now) + .unwrap(); + assert_eq!(mapping.record.contracts[0].pinned_addresses, [private]); + } + } + } + #[tokio::test] async fn allowed_ips_filters_each_answer_without_rejecting_usable_addresses() { let yaml = diff --git a/crates/openshell-supervisor-network/src/policy_dns/store.rs b/crates/openshell-supervisor-network/src/policy_dns/store.rs index eeb5d4b85f..9c76e89c5e 100644 --- a/crates/openshell-supervisor-network/src/policy_dns/store.rs +++ b/crates/openshell-supervisor-network/src/policy_dns/store.rs @@ -84,15 +84,26 @@ pub(crate) struct MappingLookup { } impl MappingLookup { - pub(crate) fn pinned_addresses(&self) -> Vec { + /// Select only contracts authorized for this process and destination port. + pub(crate) fn pinned_addresses_for( + &self, + endpoint_ids: &[PolicyEndpointId], + ) -> Result, MappingLookupError> { let mut seen = HashSet::new(); - self.record + let addresses = self + .record .contracts .iter() - .filter(|contract| contract.port == self.port) + .filter(|contract| { + contract.port == self.port && endpoint_ids.contains(&contract.endpoint_id) + }) .flat_map(|contract| contract.pinned_addresses.iter().copied()) .filter(|address| seen.insert(*address)) - .collect() + .collect::>(); + if addresses.is_empty() { + return Err(MappingLookupError::EndpointMismatch); + } + Ok(addresses) } pub(crate) fn endpoint_ids(&self) -> impl Iterator { @@ -112,18 +123,7 @@ impl MappingLookup { &self, endpoint_id: &PolicyEndpointId, ) -> Result { - let mut seen = HashSet::new(); - let addresses = self - .record - .contracts - .iter() - .filter(|contract| contract.port == self.port && &contract.endpoint_id == endpoint_id) - .flat_map(|contract| contract.pinned_addresses.iter().copied()) - .filter(|address| seen.insert(*address)) - .collect::>(); - if addresses.is_empty() { - return Err(MappingLookupError::EndpointMismatch); - } + let addresses = self.pinned_addresses_for(std::slice::from_ref(endpoint_id))?; let plan = build_pinned_validation_plan(addresses) .map_err(|_| MappingLookupError::InvalidMapping)?; validate_destination(DestinationRequest { @@ -735,7 +735,10 @@ mod tests { .lookup_intent(observed.synthetic_address, 443, 1, now) .unwrap(); assert_eq!(intent.record.normalized_name.as_str(), "unknown.example"); - assert!(intent.pinned_addresses().is_empty()); + assert!(matches!( + intent.pinned_addresses_for(&[]), + Err(MappingLookupError::EndpointMismatch) + )); assert!(matches!( store.lookup_intent(observed.synthetic_address, 443, 2, now), Err(MappingLookupError::StalePolicy) @@ -1021,6 +1024,54 @@ mod tests { assert_eq!(connector.addrs(), &["203.0.113.8:5432".parse().unwrap()]); } + #[tokio::test] + async fn endpoint_selection_unions_authorized_contracts_only_and_binds_connector() { + let store = store(1); + let now = Instant::now(); + let mut request = request("must-not-resolve.invalid", 1, Duration::from_secs(5)); + let first = request.contracts[0].clone(); + let mut second = first.clone(); + second.endpoint_id.endpoint_index = 1; + second.pinned_addresses = vec!["203.0.113.9".parse().unwrap(), first.pinned_addresses[0]]; + let mut unrelated = first.clone(); + unrelated.endpoint_id.policy_name = "other-binary".to_string(); + unrelated.pinned_addresses = vec!["10.2.3.4".parse().unwrap()]; + let mut other_port = second.clone(); + other_port.port = 443; + other_port.pinned_addresses = vec!["203.0.113.10".parse().unwrap()]; + request + .contracts + .extend([second.clone(), unrelated.clone(), other_port]); + let record = store.publish(request, 1, now).unwrap(); + let lookup = store + .lookup(record.synthetic_address, 5432, 1, now) + .unwrap(); + assert_eq!( + lookup + .pinned_addresses_for(&[first.endpoint_id.clone(), second.endpoint_id]) + .unwrap(), + [ + "203.0.113.8".parse::().unwrap(), + "203.0.113.9".parse().unwrap() + ] + ); + let connector = lookup.connector_for(&first.endpoint_id).await.unwrap(); + assert_eq!(connector.addrs(), &["203.0.113.8:5432".parse().unwrap()]); + assert!(matches!( + lookup + .connector_for(&PolicyEndpointId { + policy_name: "absent".to_string(), + endpoint_index: 0 + }) + .await, + Err(MappingLookupError::EndpointMismatch) + )); + assert!(matches!( + lookup.pinned_addresses_for(&[]), + Err(MappingLookupError::EndpointMismatch) + )); + } + #[tokio::test] async fn connector_preserves_resolver_address_order_while_deduplicating() { let store = store(1); diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 94e2a2947d..26ce2fbf98 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -15,9 +15,9 @@ use openshell_sandbox_backend::test_support as rpc_test; use crate::identity::{BinaryIdentityCache, SuppliedIdentityError}; use crate::l7::tls::ProxyTlsState; use crate::opa::{NetworkAction, OpaEngine, PolicyGenerationGuard}; -#[cfg(any(target_os = "linux", test))] -use crate::policy_dns::PolicyEndpointId; -use crate::policy_dns::{MappingLookup, MappingLookupError, ResolvedEndpointStore}; +use crate::policy_dns::{ + MappingLookup, MappingLookupError, PolicyEndpointId, ResolvedEndpointStore, +}; use crate::policy_local::{POLICY_LOCAL_HOST, PolicyLocalContext}; use crate::upstream_proxy::{self, UpstreamProxyConfig}; use futures::{FutureExt as _, StreamExt as _, stream::FuturesUnordered}; @@ -130,7 +130,8 @@ fn is_reserved_relay_destination(destination: SocketAddr) -> bool { use self::destination::{ DestinationDenial, DestinationDenialKind, DestinationRequest, DestinationValidationPlan, - build_pinned_validation_plan, build_validation_plan, validate_destination, + build_pinned_validation_plan, build_validation_plan, filter_resolved_addresses, + validate_destination, }; use self::egress::{ EgressDecision, EgressIntent, EndpointDecision, IdentityUnavailableReason, L7ConfigSnapshot, @@ -836,6 +837,20 @@ async fn preauthorize_transparent_open( let _ = completion.send(TcpOpenDecision::Denied(denial)); return None; } + if let Err(denial) = + hydrate_destination_plan(&mut decision, backend_host_gateway, trusted_host_gateway) + { + warn!(%destination, reason = %denial.reason, "Denied staged transparent destination"); + emit_staged_transparent_denial( + destination, + mapped_host, + &binary_identity, + &denial.reason, + "transparent_tcp_destination_denied", + ); + let _ = completion.send(TcpOpenDecision::Denied(TcpOpenDenial::InvalidDestination)); + return None; + } // A synthetic destination may dial only the addresses pinned for the // generation that produced this decision. Re-acquiring the mapping at that // generation keeps a policy reload between correlation and authorization @@ -876,20 +891,6 @@ async fn preauthorize_transparent_open( } }, }; - if let Err(denial) = - hydrate_destination_plan(&mut decision, backend_host_gateway, trusted_host_gateway) - { - warn!(%destination, reason = %denial.reason, "Denied staged transparent destination"); - emit_staged_transparent_denial( - destination, - mapped_host, - &binary_identity, - &denial.reason, - "transparent_tcp_destination_denied", - ); - let _ = completion.send(TcpOpenDecision::Denied(TcpOpenDenial::InvalidDestination)); - return None; - } if let Some(plan) = pinned_plan { decision.endpoint.destination = Some(plan); } @@ -1043,8 +1044,9 @@ fn resolve_transparent_target( } /// Re-acquire a policy DNS mapping at the generation that produced -/// `decision` and pin its addresses. Observation records, stale mappings, and -/// unmapped ports fail closed. +/// `decision`, select only process-authorized endpoint contracts, and apply +/// the hydrated destination restrictions before pinning. Observation records, +/// stale mappings, unmatched endpoints, and unmapped ports fail closed. fn pinned_transparent_plan( store: &ResolvedEndpointStore, destination: SocketAddr, @@ -1056,8 +1058,29 @@ fn pinned_transparent_plan( decision.policy_generation, std::time::Instant::now(), )?; - build_pinned_validation_plan(mapping.pinned_addresses()) - .map_err(|_| MappingLookupError::InvalidMapping) + let endpoint_ids = decision + .endpoint + .matched_endpoints + .iter() + .map(|endpoint| PolicyEndpointId { + policy_name: endpoint.policy_name.clone(), + endpoint_index: endpoint.endpoint_index, + }) + .collect::>(); + let addresses = mapping.pinned_addresses_for(&endpoint_ids)?; + let plan = decision + .endpoint + .destination + .as_ref() + .ok_or(MappingLookupError::InvalidMapping)?; + let addresses = filter_resolved_addresses( + plan, + mapping.record.normalized_name.as_str(), + destination.port(), + &addresses, + ) + .map_err(|_| MappingLookupError::InvalidMapping)?; + build_pinned_validation_plan(addresses).map_err(|_| MappingLookupError::InvalidMapping) } fn valid_policy_local_request(method: &str, target: &str, request_headers: &str) -> bool { @@ -7956,6 +7979,162 @@ process: { run_as_user: sandbox, run_as_group: sandbox } assert_eq!(connector.addrs(), &["203.0.113.8:5432".parse().unwrap()]); } + #[tokio::test] + async fn staged_transparent_pins_preserve_process_destination_authorization() { + for (family, private, public, second_public) in [ + ( + crate::policy_dns::AddressFamily::Ipv4, + "10.2.3.4", + "8.8.8.8", + "8.8.4.4", + ), + ( + crate::policy_dns::AddressFamily::Ipv6, + "fd12::4", + "2001:4860:4860::8888", + "2001:4860:4860::8844", + ), + ] { + let private = private.parse::().unwrap(); + let public = public.parse::().unwrap(); + let second_public = second_public.parse::().unwrap(); + for endpoint in [ + "host: '*.service.example'", + "host: db.service.example, advisor_proposed: true", + ] { + // No contract for the caller, mixed pins in its contract, a + // compatible union, explicit narrow permission, and no usable pin. + for case in 0..6 { + let allowed_ips = if case == 3 { + format!(", allowed_ips: ['{private}']") + } else { + String::new() + }; + let other_binary = if case == 5 { + "/usr/bin/curl" + } else { + "/usr/bin/psql" + }; + let policy = format!( + r" +network_policies: + other_binary: + name: other_binary + endpoints: [{{ host: db.service.example, port: 5432, protocol: tcp }}] + binaries: [{{ path: {other_binary} }}] + caller: + name: caller + endpoints: + - {{ {endpoint}, port: 5432, protocol: tcp{allowed_ips} }} + - {{ {endpoint}, port: 5432, protocol: tcp{allowed_ips} }} + binaries: [{{ path: /usr/bin/curl }}] +filesystem_policy: {{ include_workdir: true, read_only: [], read_write: [] }} +landlock: {{ compatibility: best_effort }} +process: {{ run_as_user: sandbox, run_as_group: sandbox }} +" + ); + let engine = native_identity_engine( + include_str!("../data/sandbox-policy.rego"), + &policy_with_native_executable_paths(&policy), + ) + .unwrap(); + let store = policy_dns_test_store(); + let contract = |policy_name: &str, endpoint_index, addresses: Vec| { + crate::policy_dns::ResolvedPortContract { + endpoint_id: PolicyEndpointId { + policy_name: policy_name.to_string(), + endpoint_index, + }, + port: 5432, + destination_plan: DestinationValidationPlan { + address_authorization: + destination::AddressAuthorization::ExactDeclaredHost, + }, + pinned_addresses: addresses, + } + }; + let mut contracts = vec![contract("other_binary", 0, vec![private])]; + if case != 0 { + contracts.push(contract( + "caller", + 0, + if case == 4 { + vec![private] + } else { + vec![private, public] + }, + )); + } + if case == 2 { + contracts.push(contract("caller", 1, vec![second_public])); + } + let record = store + .publish( + crate::policy_dns::PublishRequest { + normalized_name: crate::policy_dns::NormalizedName::parse( + "db.service.example", + ) + .unwrap(), + family, + allocation_identity: [1; 32], + policy_generation: engine.current_generation(), + ttl: std::time::Duration::from_secs(30), + contracts, + }, + engine.current_generation(), + std::time::Instant::now(), + ) + .unwrap(); + let (open, completion) = staged_curl_open( + SocketAddr::new(record.synthetic_address, 5432), + engine.current_generation(), + ); + let accepted = preauthorize_transparent_open( + open, + Some(&store), + &engine, + &BinaryIdentityCache::new(), + None, + None, + false, + None, + None, + ) + .await; + if case == 0 || case == 4 { + assert!(accepted.is_none(), "{endpoint}: case {case}, {family:?}"); + assert_eq!( + completion.await.unwrap(), + TcpOpenDecision::Denied(TcpOpenDenial::InvalidDestination) + ); + } else { + let (_, _, _, transparent) = accepted.expect("authorized endpoint pins"); + assert_eq!(completion.await.unwrap(), TcpOpenDecision::RelayReady); + let (_, connector) = + transparent.and_then(|open| open.authorization).unwrap(); + let expected = match case { + 2 => vec![ + SocketAddr::new(public, 5432), + SocketAddr::new(second_public, 5432), + ], + 3 => vec![SocketAddr::new(private, 5432)], + 5 => vec![ + SocketAddr::new(private, 5432), + SocketAddr::new(public, 5432), + ], + _ => vec![SocketAddr::new(public, 5432)], + }; + assert_eq!( + connector.addrs(), + expected, + "{endpoint}: case {case}, {family:?}" + ); + } + } + } + } + } + #[tokio::test] async fn staged_transparent_open_proposes_a_denied_observation_hostname() { let engine = native_identity_engine( @@ -8088,12 +8267,14 @@ process: { run_as_user: sandbox, run_as_group: sandbox } engine.current_generation(), ); let decide = |host: &str, port| { - authorize_supplied_identity( + let mut decision = authorize_supplied_identity( &engine, &BinaryIdentityCache::new(), EgressIntent::connect(host.to_string(), port), &open.binary_identity, - ) + ); + hydrate_destination_plan(&mut decision, None, None).unwrap(); + decision }; let mapped_destination = SocketAddr::new(mapped.synthetic_address, 5432); diff --git a/e2e/rust/tests/transparent_tcp.rs b/e2e/rust/tests/transparent_tcp.rs index 3c9192d31c..1046145118 100644 --- a/e2e/rust/tests/transparent_tcp.rs +++ b/e2e/rust/tests/transparent_tcp.rs @@ -250,6 +250,128 @@ async fn wait_for_sandbox_logs( } } +fn private_fixture_probe(host: &str, marker: &str, should_connect: bool) -> String { + format!( + r"import socket +answers = socket.getaddrinfo({host:?}, {FIXTURE_PORT}, type=socket.SOCK_STREAM) +assert any(item[4][0].startswith(('198.18.', '198.19.')) for item in answers), answers +reached = False +try: + with socket.create_connection(({host:?}, {FIXTURE_PORT}), timeout=3) as conn: + conn.sendall({marker:?}.encode('ascii')) + reached = conn.recv(1024) == b'private-fixture-ok' +except OSError: + pass +assert reached == {expected}, reached +print('private-pin-probe-ok') +", + expected = if should_connect { "True" } else { "False" } + ) +} + +#[tokio::test] +async fn rootless_podman_private_dns_pins_do_not_cross_binary_rules() { + if !is_e2e_driver("podman") { + return; + } + let host = "transparent-private.openshell.test"; + let fixture = SupportContainer::start_python( + host, + &format!( + r"import socket +s = socket.socket() +s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +s.bind(('0.0.0.0', {FIXTURE_PORT})) +s.listen() +while True: + c, _ = s.accept() + data = c.recv(1024) + if data: + print(data.decode('ascii'), flush=True) + c.sendall(b'private-fixture-ok') + c.close() +" + ), + FIXTURE_PORT, + ) + .await + .expect("start private-network fixture"); + let fixture_ip = fixture + .ip() + .expect("fixture IP") + .parse::() + .expect("IPv4 fixture"); + assert!( + fixture_ip.is_private(), + "fixture must exercise private address enforcement" + ); + let mut policy = NamedTempFile::new().expect("create policy"); + write!( + policy, + r"version: 1 +filesystem_policy: + include_workdir: true + read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log] + read_write: [/sandbox, /tmp, /dev/null] +landlock: {{ compatibility: best_effort }} +process: {{ run_as_user: sandbox, run_as_group: sandbox }} +network_policies: + declared: + name: declared + endpoints: + - {{ host: {host}, port: {FIXTURE_PORT}, protocol: tcp }} + binaries: [{{ path: /sandbox/declared-python }}] + wildcard: + name: wildcard + endpoints: + - {{ host: '*.openshell.test', port: {FIXTURE_PORT}, protocol: tcp }} + binaries: [{{ path: /sandbox/wildcard-python }}] +" + ) + .expect("write policy"); + policy.flush().expect("flush policy"); + let mut sandbox = SandboxGuard::create_keep_with_args( + &["--policy", &policy.path().to_string_lossy()], + &["sh", "-c", "set -e; cp $(command -v python3) /sandbox/declared-python; cp $(command -v python3) /sandbox/wildcard-python; echo Ready; sleep infinity"], + "Ready", + ).await.expect("create private DNS sandbox"); + // Copied uv-managed executables still need their original standard library. + let python_prefix = sandbox + .exec(&["python3", "-c", "import sys; print(sys.base_prefix)"]) + .await + .expect("find original Python installation prefix"); + let python_home = format!("PYTHONHOME={}", python_prefix.trim()); + let positive = sandbox + .exec(&[ + "env", + &python_home, + "/sandbox/declared-python", + "-c", + &private_fixture_probe(host, "authorized-control", true), + ]) + .await + .expect("exact-host binary reaches fixture"); + assert!(positive.contains("private-pin-probe-ok"), "{positive}"); + let negative = sandbox + .exec(&[ + "env", + &python_home, + "/sandbox/wildcard-python", + "-c", + &private_fixture_probe(host, "blocked-probe", false), + ]) + .await + .expect("wildcard binary cannot borrow private pins"); + assert!(negative.contains("private-pin-probe-ok"), "{negative}"); + let logs = fixture.logs().expect("private fixture logs"); + assert!(logs.contains("authorized-control"), "{logs}"); + assert!( + !logs.contains("blocked-probe"), + "denied probe reached private fixture: {logs}" + ); + sandbox.cleanup().await; +} + #[tokio::test] async fn local_container_native_tcp_uses_policy_dns_and_fails_closed() { if !is_e2e_driver("docker") && !is_e2e_driver("podman") {