fix(network): align destination checks for direct connections (#5004)

Signed-off-by: John Myers <9696606+johntmyers@users.noreply.github.com>
2026-10-31 07:44:57 +08:00 · 2026-10-28 18:41:46 +00:00
parent 33138516ad
commit ff1f8ff55c
4 changed files with 461 additions and 42 deletions
@@ -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(&regorus::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::<IpAddr>().unwrap();
+                let public = public.parse::<IpAddr>().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 =
@@ -84,15 +84,26 @@ pub(crate) struct MappingLookup {
 }

 impl MappingLookup {
-    pub(crate) fn pinned_addresses(&self) -> Vec<IpAddr> {
+    /// Select only contracts authorized for this process and destination port.
+    pub(crate) fn pinned_addresses_for(
+        &self,
+        endpoint_ids: &[PolicyEndpointId],
+    ) -> Result<Vec<IpAddr>, 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::<Vec<_>>();
+        if addresses.is_empty() {
+            return Err(MappingLookupError::EndpointMismatch);
+        }
+        Ok(addresses)
    }

    pub(crate) fn endpoint_ids(&self) -> impl Iterator<Item = &PolicyEndpointId> {
@@ -112,18 +123,7 @@ impl MappingLookup {
        &self,
        endpoint_id: &PolicyEndpointId,
    ) -> Result<UpstreamConnector, MappingLookupError> {
-        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::<Vec<_>>();
-        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::<IpAddr>().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);
@@ -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::<Vec<_>>();
+    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::<IpAddr>().unwrap();
+            let public = public.parse::<IpAddr>().unwrap();
+            let second_public = second_public.parse::<IpAddr>().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<IpAddr>| {
+                        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);

@@ -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::<std::net::Ipv4Addr>()
+        .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") {