diff --git a/Cargo.lock b/Cargo.lock index c0afff104b..5576ef2937 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -127,7 +127,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -138,7 +138,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1204,6 +1204,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -1728,7 +1734,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2187,6 +2193,25 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e712f64ec3850b98572bffac52e2c6f282b29fe6c5fa6d42334b30be438d95c1" +[[package]] +name = "hickory-proto" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bab31817bfb44672a252e97fe81cd0c18d1b2cf892108922f6818820df8c643" +dependencies = [ + "data-encoding", + "idna", + "ipnet", + "jni 0.22.4", + "once_cell", + "rand 0.10.2", + "ring", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "url", +] + [[package]] name = "hkdf" version = "0.12.4" @@ -2784,6 +2809,36 @@ dependencies = [ "windows-sys 0.45.0", ] +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys 0.4.1", + "log", + "simd_cesu8", + "thiserror 2.0.18", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "simd_cesu8", + "syn 2.0.117", +] + [[package]] name = "jni-sys" version = "0.3.1" @@ -3446,7 +3501,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3624,6 +3679,10 @@ name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +dependencies = [ + "critical-section", + "portable-atomic", +] [[package]] name = "once_cell_polyfill" @@ -4204,6 +4263,7 @@ dependencies = [ "futures", "glob", "hex", + "hickory-proto", "http 1.4.0", "ipnet", "libc", @@ -5622,7 +5682,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5692,7 +5752,7 @@ checksum = "1d99feebc72bae7ab76ba994bb5e121b8d83d910ca40b36e0921f53becc41784" dependencies = [ "core-foundation", "core-foundation-sys", - "jni", + "jni 0.21.1", "log", "once_cell", "rustls 0.23.38", @@ -5702,7 +5762,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6182,6 +6242,22 @@ version = "0.3.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" +[[package]] +name = "simd_cesu8" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520" +dependencies = [ + "rustc_version", + "simdutf8", +] + +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "simple_asn1" version = "0.6.4" @@ -6232,7 +6308,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -6729,7 +6805,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6765,7 +6841,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7762,7 +7838,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/architecture/sandbox.md b/architecture/sandbox.md index 71094cb07b..c92ad75479 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -72,9 +72,16 @@ its guarded single-request relay while sharing authorization, request context, policy-pinning, and destination boundaries. Adapter-specific response and OCSF event shapes remain at the protocol boundary. Policy authors may use `protocol: tcp` as an explicit spelling of the existing -L4 passthrough behavior. Omitting `protocol` remains equivalent. The egress -intent reserves a transparent TCP adapter and a policy-DNS-pinned address, but -DNS serving and transparent TCP capture are not active yet. +L4 passthrough behavior. Omitting `protocol` remains equivalent. The network +supervisor contains a dormant policy-DNS boundary for explicit TCP endpoints: +it snapshots eligible endpoint identities from one policy generation, resolves +eligible names only through an explicitly supplied trusted resolver, filters +answers through the shared destination controls, and publishes expiring +synthetic-address mappings with separate mapping generations. Refreshes retain +their synthetic identity, and policy reload, expiry, wrong ports, missing +mappings, or pool exhaustion fail closed. The pinned connector never resolves +the name again. No DNS listener is exposed to workloads, resolver configuration +is not injected, and transparent TCP capture is not active in this increment. Provider credential placeholders are resolved through the live provider state for each HTTP request, after destination and L7 policy admission. A static diff --git a/crates/openshell-supervisor-network/Cargo.toml b/crates/openshell-supervisor-network/Cargo.toml index 58360aa56c..7fddef0357 100644 --- a/crates/openshell-supervisor-network/Cargo.toml +++ b/crates/openshell-supervisor-network/Cargo.toml @@ -27,6 +27,7 @@ bytes = { workspace = true } flate2 = "1" glob = { workspace = true } hex = "0.4" +hickory-proto = "0.26.1" ipnet = "2" miette = { workspace = true } prost-types = { workspace = true } diff --git a/crates/openshell-supervisor-network/data/sandbox-policy.rego b/crates/openshell-supervisor-network/data/sandbox-policy.rego index 5db01373e6..c85115cc48 100644 --- a/crates/openshell-supervisor-network/data/sandbox-policy.rego +++ b/crates/openshell-supervisor-network/data/sandbox-policy.rego @@ -921,6 +921,31 @@ _matching_endpoint_records := [record | record := records[_] ] +# Endpoints eligible for policy DNS are a policy-data snapshot, not an +# authorization decision. In particular, they do not depend on input.exec or +# grant access to any process. Only endpoints that explicitly opt into raw TCP +# and provide a resolvable host plus concrete ports are materialized. +policy_dns_eligible_endpoint_records := [record | + some policy_name + policy := data.network_policies[policy_name] + some endpoint_index + ep := policy.endpoints[endpoint_index] + lower(object.get(ep, "protocol", "")) == "tcp" + object.get(ep, "host", "") != "" + ports := object.get(ep, "ports", []) + count(ports) > 0 + every port in ports { + is_number(port) + port >= 1 + port <= 65535 + } + record := { + "policy_name": policy_name, + "endpoint_index": endpoint_index, + "endpoint": ep, + } +] + matched_endpoint_config := _matching_endpoint_configs[0] if { count(_matching_endpoint_configs) > 0 } diff --git a/crates/openshell-supervisor-network/src/lib.rs b/crates/openshell-supervisor-network/src/lib.rs index f5d0205e3a..ccb0a4d166 100644 --- a/crates/openshell-supervisor-network/src/lib.rs +++ b/crates/openshell-supervisor-network/src/lib.rs @@ -12,6 +12,7 @@ pub mod identity; pub mod inference_routes; pub mod l7; pub mod opa; +pub(crate) mod policy_dns; pub mod policy_local; pub mod procfs; pub mod proxy; diff --git a/crates/openshell-supervisor-network/src/opa.rs b/crates/openshell-supervisor-network/src/opa.rs index e70b6f6dfe..b2d405ceb2 100644 --- a/crates/openshell-supervisor-network/src/opa.rs +++ b/crates/openshell-supervisor-network/src/opa.rs @@ -58,6 +58,16 @@ pub struct MatchedEndpoint { pub endpoint: regorus::Value, } +/// Policy-DNS eligible endpoint metadata captured from one policy generation. +/// +/// This is policy data only. It deliberately contains no process identity or +/// network authorization decision. +#[derive(Debug, Clone)] +pub struct PolicyDnsEligibilitySnapshot { + pub endpoints: Vec, + pub generation: u64, +} + /// Atomic policy result used to authorize and materialize one egress request. #[derive(Debug, Clone)] pub struct EgressAuthorization { @@ -583,6 +593,48 @@ impl OpaEngine { }) } + /// Return all explicit TCP endpoints eligible for policy DNS. + /// + /// The owned endpoint records and generation are captured while holding + /// the engine lock, so reloads cannot mix data from one generation with + /// the generation number of another. Fail-closed quarantine produces an + /// empty snapshot for its quarantine generation. + pub fn policy_dns_eligibility_snapshot(&self) -> Result { + let mut engine = self + .engine + .lock() + .map_err(|_| miette::miette!("OPA engine lock poisoned"))?; + let generation = self.current_generation(); + + if self + .fail_closed_reason + .read() + .map_err(|_| miette::miette!("OPA fail-closed state lock poisoned"))? + .is_some() + { + return Ok(PolicyDnsEligibilitySnapshot { + endpoints: Vec::new(), + generation, + }); + } + + let value = engine + .eval_rule("data.openshell.sandbox.policy_dns_eligible_endpoint_records".into()) + .map_err(|error| miette::miette!("{error}"))?; + let endpoints = match value { + regorus::Value::Array(values) => { + values.iter().filter_map(parse_matched_endpoint).collect() + } + regorus::Value::Undefined => Vec::new(), + other => parse_matched_endpoint(&other).into_iter().collect(), + }; + + Ok(PolicyDnsEligibilitySnapshot { + endpoints, + generation, + }) + } + /// Reload policy and data from strings (data is YAML). /// /// Designed for future gRPC hot-reload from the openshell gateway. @@ -738,9 +790,37 @@ impl OpaEngine { self.generation.load(Ordering::Acquire) } + /// Run a short operation only while `expected_generation` is current. + /// + /// The engine mutex is also the policy reload mutex. Holding it across the + /// generation comparison and callback linearizes state derived from an OPA + /// snapshot with every policy reload and fail-closed transition. Callers + /// must not perform I/O or other long-running work in `operation`. + pub(crate) fn with_current_generation( + &self, + expected_generation: u64, + operation: impl FnOnce(u64) -> T, + ) -> Result> { + let _engine = self + .engine + .lock() + .map_err(|_| miette::miette!("OPA engine lock poisoned"))?; + let current_generation = self.current_generation(); + if current_generation != expected_generation { + return Ok(None); + } + Ok(Some(operation(current_generation))) + } + /// Replace the complete middleware service registry and invalidate /// existing tunnels so subsequent requests use the new service set. pub fn replace_middleware_registry(&self, registry: MiddlewareRegistry) -> Result<()> { + // Generation changes serialize through the engine lock so guarded + // publication cannot overlap any runtime generation transition. + let _engine = self + .engine + .lock() + .map_err(|_| miette::miette!("OPA engine lock poisoned"))?; let mut runner = self .middleware_runner .write() @@ -2021,6 +2101,71 @@ mod tests { } } + const POLICY_DNS_SNAPSHOT_DATA: &str = r#" +network_policies: + dns_transport: + name: dns_transport + endpoints: + - { host: resolver.example, ports: [53, 853], protocol: tcp } + - { host: web.example, port: 443, protocol: rest, access: full } + - { host: implicit.example, port: 443 } + - { host: "", port: 53, protocol: tcp, allowed_ips: [8.8.8.8] } + - { host: secondary.example, port: 5353, protocol: tcp } + binaries: + - { path: /usr/bin/one-process } +filesystem_policy: + include_workdir: true + read_only: [] + read_write: [] +landlock: + compatibility: best_effort +process: + run_as_user: sandbox + run_as_group: sandbox +"#; + + #[test] + fn policy_dns_snapshot_is_tcp_only_stable_and_generation_consistent() { + let engine = OpaEngine::from_strings(TEST_POLICY, POLICY_DNS_SNAPSHOT_DATA).unwrap(); + + let snapshot = engine.policy_dns_eligibility_snapshot().unwrap(); + + assert_eq!(snapshot.generation, engine.current_generation()); + assert_eq!(snapshot.endpoints.len(), 2); + assert_eq!(snapshot.endpoints[0].policy_name, "dns_transport"); + assert_eq!(snapshot.endpoints[0].endpoint_index, 0); + assert_eq!( + get_str(&snapshot.endpoints[0].endpoint, "host").as_deref(), + Some("resolver.example") + ); + let Some(regorus::Value::Array(ports)) = + get_field(&snapshot.endpoints[0].endpoint, "ports") + else { + panic!("eligible endpoint must retain concrete ports"); + }; + assert_eq!(ports.as_ref(), &[53.into(), 853.into()]); + assert_eq!(snapshot.endpoints[1].endpoint_index, 4); + + engine + .reload(TEST_POLICY, POLICY_DNS_SNAPSHOT_DATA) + .unwrap(); + let reloaded = engine.policy_dns_eligibility_snapshot().unwrap(); + assert_eq!(reloaded.generation, snapshot.generation + 1); + assert_eq!(reloaded.endpoints.len(), 2); + assert_eq!(reloaded.endpoints[1].endpoint_index, 4); + } + + #[test] + fn policy_dns_snapshot_is_empty_during_fail_closed_quarantine() { + let engine = OpaEngine::from_strings(TEST_POLICY, POLICY_DNS_SNAPSHOT_DATA).unwrap(); + let generation = engine.enter_fail_closed("invalid candidate").unwrap(); + + let snapshot = engine.policy_dns_eligibility_snapshot().unwrap(); + + assert_eq!(snapshot.generation, generation); + assert!(snapshot.endpoints.is_empty()); + } + #[test] fn allowed_binary_and_endpoint() { let engine = test_engine(); diff --git a/crates/openshell-supervisor-network/src/policy_dns/mod.rs b/crates/openshell-supervisor-network/src/policy_dns/mod.rs new file mode 100644 index 0000000000..1c5bdffcdc --- /dev/null +++ b/crates/openshell-supervisor-network/src/policy_dns/mod.rs @@ -0,0 +1,687 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#![allow( + clippy::redundant_pub_crate, + reason = "the crate-private API is consumed by the runtime activation slice" +)] + +//! Dormant policy-gated DNS and synthetic resolved-endpoint correlation. +//! +//! This module implements the DNS security boundary and mapping state only. +//! Runtime listener startup, resolver injection, and transparent TCP capture +//! intentionally land in later stack entries. + +#![allow( + dead_code, + unused_imports, + reason = "PR2 exposes a dormant library boundary consumed by PR3 runtime wiring" +)] + +mod name; +mod resolver; +mod store; +mod wire; + +pub(crate) use name::NormalizedName; +pub(crate) use resolver::{AddressFamily, SocketTrustedResolver, TrustedAnswer, TrustedResolver}; +pub(crate) use store::{ + MappingLookup, MappingLookupError, PolicyDnsMetricsSnapshot, PolicyEndpointId, PublishError, + PublishRequest, ResolvedEndpointRecord, ResolvedEndpointStore, ResolvedPortContract, + StoreConfig, SyntheticPools, +}; + +use crate::opa::OpaEngine; +use crate::proxy::destination::{build_validation_plan, filter_resolved_addresses}; +use crate::proxy::is_host_gateway_alias; +use openshell_core::host_pattern::HostSelector; +use openshell_ocsf::{ + ActionId, ActivityId, ConfigStateChangeBuilder, DispositionId, Endpoint, + NetworkActivityBuilder, SeverityId, StateId, StatusId, ocsf_emit, +}; +use sha2::{Digest, Sha256}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +pub(crate) const MIN_MAPPING_TTL: Duration = Duration::from_secs(1); +pub(crate) const MAX_MAPPING_TTL: Duration = Duration::from_secs(30); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct SyntheticAnswer { + pub(crate) address: std::net::IpAddr, + pub(crate) ttl: Duration, + pub(crate) mapping_id: uuid::Uuid, + pub(crate) mapping_generation: u64, + pub(crate) policy_generation: u64, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum PolicyDnsError { + #[error("DNS query name is invalid")] + InvalidName, + #[error("DNS name is not eligible for policy DNS")] + Ineligible, + #[error("trusted host gateway is unavailable for the reserved alias")] + TrustedGatewayUnavailable, + #[error("trusted resolver failed: {0}")] + Resolver(#[from] resolver::ResolveError), + #[error("no trusted resolver address passed endpoint destination policy")] + NoValidAddress, + #[error("policy generation changed before DNS mapping publication")] + StalePolicy, + #[error("resolved endpoint mapping could not be published: {0}")] + Publish(#[from] PublishError), + #[error("policy DNS eligibility snapshot failed: {0}")] + Policy(String), +} + +/// Policy-gated DNS evaluator and synthetic mapping publisher. +/// +/// No socket is bound by this type. A later runtime adapter owns listener and +/// namespace lifecycle and calls the bounded wire helpers in this module. +pub(crate) struct PolicyDnsService { + policy: Arc, + resolver: R, + store: Arc, + trusted_host_gateway: Option, +} + +impl PolicyDnsService { + pub(crate) fn new( + policy: Arc, + resolver: R, + store: Arc, + trusted_host_gateway: Option, + ) -> Self { + Self { + policy, + resolver, + store, + trusted_host_gateway, + } + } + + pub(crate) async fn answer_query( + &self, + raw_name: &str, + family: AddressFamily, + now: Instant, + ) -> Result { + self.store.note_query(); + let normalized_name = + NormalizedName::parse(raw_name).map_err(|_| PolicyDnsError::InvalidName)?; + if is_host_gateway_alias(normalized_name.as_str()) && self.trusted_host_gateway.is_none() { + self.store.note_refused(); + emit_dns_denial( + &normalized_name, + "policy_dns_trusted_gateway_unavailable", + "Policy DNS refused a reserved host-gateway alias because no trusted gateway is configured", + ); + return Err(PolicyDnsError::TrustedGatewayUnavailable); + } + let snapshot = self + .policy + .policy_dns_eligibility_snapshot() + .map_err(|error| PolicyDnsError::Policy(error.to_string()))?; + let eligible = eligible_endpoints( + &snapshot.endpoints, + &normalized_name, + self.trusted_host_gateway, + )?; + if eligible.is_empty() { + self.store.note_refused(); + emit_dns_denial( + &normalized_name, + "policy_dns_ineligible", + "Policy DNS refused a name that is not eligible in the active policy", + ); + return Err(PolicyDnsError::Ineligible); + } + + // The trusted resolver is invoked only after the immutable snapshot + // proved policy eligibility. It never consults sandbox resolver state. + self.store.note_upstream_query(); + let trusted_answer = self.resolver.resolve(&normalized_name, family).await?; + let ttl = clamp_mapping_ttl(trusted_answer.ttl); + let allocation_identity = allocation_identity(&eligible); + let mut contracts = Vec::new(); + for endpoint in eligible { + for port in endpoint.ports { + let Ok(pinned_addresses) = filter_resolved_addresses( + &endpoint.destination_plan, + normalized_name.as_str(), + port, + &trusted_answer.addresses, + ) else { + continue; + }; + contracts.push(ResolvedPortContract { + endpoint_id: endpoint.endpoint_id.clone(), + port, + destination_plan: endpoint.destination_plan.clone(), + pinned_addresses, + }); + } + } + contracts.sort_by(|left, right| { + (&left.endpoint_id, left.port).cmp(&(&right.endpoint_id, right.port)) + }); + if contracts.is_empty() { + self.store.note_no_valid_address(); + emit_dns_denial( + &normalized_name, + "policy_dns_no_valid_address", + "Policy DNS rejected every trusted resolver address", + ); + return Err(PolicyDnsError::NoValidAddress); + } + + let request = PublishRequest { + normalized_name: normalized_name.clone(), + family, + allocation_identity, + policy_generation: snapshot.generation, + ttl, + contracts, + }; + let publication = self + .policy + .with_current_generation(snapshot.generation, |current_generation| { + self.store.publish(request, current_generation, now) + }) + .map_err(|error| PolicyDnsError::Policy(error.to_string()))? + .ok_or(PolicyDnsError::StalePolicy)?; + let record = publication?; + emit_mapping_publication(&record); + Ok(SyntheticAnswer { + address: record.synthetic_address, + ttl, + mapping_id: record.mapping_id, + mapping_generation: record.mapping_generation, + policy_generation: record.policy_generation, + }) + } + + pub(crate) fn store(&self) -> &Arc { + &self.store + } +} + +struct EligibleEndpoint { + endpoint_id: PolicyEndpointId, + ports: Vec, + destination_plan: crate::proxy::destination::DestinationValidationPlan, + contract_fingerprint: String, +} + +fn eligible_endpoints( + endpoints: &[crate::opa::MatchedEndpoint], + name: &NormalizedName, + trusted_host_gateway: Option, +) -> Result, PolicyDnsError> { + let mut eligible = Vec::new(); + for endpoint in endpoints { + let Some(pattern) = value_string(&endpoint.endpoint, "host") else { + continue; + }; + let pattern = pattern.trim_end_matches('.').to_ascii_lowercase(); + let selector = HostSelector::new(std::slice::from_ref(&pattern), &[]) + .map_err(PolicyDnsError::Policy)?; + if !selector.matches(name.as_str()) { + continue; + } + let ports = value_ports(&endpoint.endpoint); + if ports.is_empty() { + continue; + } + let raw_allowed_ips = value_string_array(&endpoint.endpoint, "allowed_ips"); + let exact_declared_host = !pattern.contains('*') && pattern == name.as_str(); + let destination_plan = build_validation_plan( + name.as_str(), + name.as_str(), + trusted_host_gateway, + &raw_allowed_ips, + exact_declared_host, + ) + .map_err(|error| PolicyDnsError::Policy(error.reason))?; + eligible.push(EligibleEndpoint { + endpoint_id: PolicyEndpointId { + policy_name: endpoint.policy_name.clone(), + endpoint_index: endpoint.endpoint_index, + }, + ports, + destination_plan, + contract_fingerprint: endpoint.endpoint.to_string(), + }); + } + Ok(eligible) +} + +fn allocation_identity(endpoints: &[EligibleEndpoint]) -> [u8; 32] { + let mut contracts = endpoints + .iter() + .map(|endpoint| { + format!( + "{}\0{}\0{}", + endpoint.endpoint_id.policy_name, + endpoint.endpoint_id.endpoint_index, + endpoint.contract_fingerprint + ) + }) + .collect::>(); + contracts.sort(); + let mut hasher = Sha256::new(); + for contract in contracts { + hasher.update(contract.as_bytes()); + hasher.update([0xff]); + } + hasher.finalize().into() +} + +fn value_field<'a>(value: &'a regorus::Value, key: &str) -> Option<&'a regorus::Value> { + let regorus::Value::Object(fields) = value else { + return None; + }; + fields.get(®orus::Value::String(key.into())) +} + +fn value_string(value: ®orus::Value, key: &str) -> Option { + match value_field(value, key) { + Some(regorus::Value::String(value)) => Some(value.to_string()), + _ => None, + } +} + +fn value_string_array(value: ®orus::Value, key: &str) -> Vec { + match value_field(value, key) { + Some(regorus::Value::Array(values)) => values + .iter() + .filter_map(|value| match value { + regorus::Value::String(value) => Some(value.to_string()), + _ => None, + }) + .collect(), + _ => Vec::new(), + } +} + +fn value_ports(value: ®orus::Value) -> Vec { + let mut ports = match value_field(value, "ports") { + Some(regorus::Value::Array(values)) => values + .iter() + .filter_map(|value| match value { + regorus::Value::Number(number) => number + .as_i64() + .and_then(|port| u16::try_from(port).ok()) + .filter(|port| *port != 0), + _ => None, + }) + .collect::>(), + _ => Vec::new(), + }; + ports.sort_unstable(); + ports.dedup(); + ports +} + +fn clamp_mapping_ttl(ttl: Duration) -> Duration { + ttl.max(MIN_MAPPING_TTL).min(MAX_MAPPING_TTL) +} + +fn emit_dns_denial(name: &NormalizedName, detail: &str, message: &str) { + ocsf_emit!( + NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Refuse) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain(name.as_str(), 53)) + .status_detail(detail) + .message(message) + .build() + ); +} + +fn emit_mapping_publication(record: &ResolvedEndpointRecord) { + ocsf_emit!( + ConfigStateChangeBuilder::new(openshell_ocsf::ctx::ctx()) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .state(StateId::Enabled, "published") + .unmapped("normalized_name", record.normalized_name.as_str()) + .unmapped("address_family", format!("{:?}", record.family)) + .unmapped("allowed_port_count", record.allowed_ports().len() as u64) + .unmapped("policy_generation", record.policy_generation) + .unmapped("mapping_generation", record.mapping_generation) + .unmapped("mapping_id", record.mapping_id.to_string()) + .message("Policy DNS resolved-endpoint mapping published") + .build() + ); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::Notify; + + struct FakeResolver { + calls: AtomicUsize, + answer: TrustedAnswer, + } + + impl TrustedResolver for FakeResolver { + async fn resolve( + &self, + _name: &NormalizedName, + _family: AddressFamily, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(self.answer.clone()) + } + } + + fn service(policy_yaml: &str, addresses: Vec) -> PolicyDnsService { + service_with_gateway(policy_yaml, addresses, None) + } + + fn service_with_gateway( + policy_yaml: &str, + addresses: Vec, + trusted_host_gateway: Option, + ) -> PolicyDnsService { + let policy = Arc::new( + OpaEngine::from_strings(include_str!("../../data/sandbox-policy.rego"), policy_yaml) + .unwrap(), + ); + let pools = SyntheticPools::new( + Ipv4Addr::new(198, 18, 0, 1)..=Ipv4Addr::new(198, 18, 0, 8), + "fd00:1::1".parse::().unwrap()..="fd00:1::8".parse::().unwrap(), + ) + .unwrap(); + PolicyDnsService::new( + policy, + FakeResolver { + calls: AtomicUsize::new(0), + answer: TrustedAnswer { + addresses, + ttl: Duration::from_secs(300), + }, + }, + Arc::new(ResolvedEndpointStore::new( + StoreConfig::new(pools, 16).unwrap(), + )), + trusted_host_gateway, + ) + } + + const BASE_POLICY: &str = r" +network_policies: + database: + name: database + endpoints: + - { host: db.example, port: 5432, protocol: tcp } + binaries: [{ path: /usr/bin/psql }] +filesystem_policy: { include_workdir: true, read_only: [], read_write: [] } +landlock: { compatibility: best_effort } +process: { run_as_user: sandbox, run_as_group: sandbox } +"; + + #[tokio::test] + async fn refuses_ineligible_name_before_upstream_resolution() { + let service = service(BASE_POLICY, vec!["8.8.8.8".parse().unwrap()]); + let result = service + .answer_query("other.example", AddressFamily::Ipv4, Instant::now()) + .await; + assert!(matches!(result, Err(PolicyDnsError::Ineligible))); + assert_eq!(service.resolver.calls.load(Ordering::SeqCst), 0); + assert_eq!(service.store.metrics(Instant::now()).refused, 1); + } + + #[tokio::test] + async fn eligible_name_filters_answers_and_publishes_bounded_mapping() { + let service = service( + BASE_POLICY, + vec!["127.0.0.1".parse().unwrap(), "10.2.3.4".parse().unwrap()], + ); + let now = Instant::now(); + let answer = service + .answer_query("DB.EXAMPLE.", AddressFamily::Ipv4, now) + .await + .unwrap(); + assert_eq!(answer.ttl, MAX_MAPPING_TTL); + let mapping = service + .store + .lookup(answer.address, 5432, answer.policy_generation, now) + .unwrap(); + assert_eq!(mapping.record.normalized_name.as_str(), "db.example"); + assert_eq!( + mapping.record.contracts[0].pinned_addresses, + ["10.2.3.4".parse::().unwrap()] + ); + } + + #[tokio::test] + async fn wildcard_is_eligible_but_uses_public_only_destination_rules() { + let yaml = BASE_POLICY.replace("db.example", "'*.example.com'"); + let service = service(&yaml, vec!["10.2.3.4".parse().unwrap()]); + let result = service + .answer_query("db.example.com", AddressFamily::Ipv4, Instant::now()) + .await; + assert!(matches!(result, Err(PolicyDnsError::NoValidAddress))); + } + + #[tokio::test] + async fn allowed_ips_filters_each_answer_without_rejecting_usable_addresses() { + let yaml = + BASE_POLICY.replace("protocol: tcp", "protocol: tcp, allowed_ips: [10.2.0.0/16]"); + let service = service( + &yaml, + vec!["10.3.4.5".parse().unwrap(), "10.2.3.4".parse().unwrap()], + ); + let now = Instant::now(); + let answer = service + .answer_query("db.example", AddressFamily::Ipv4, now) + .await + .unwrap(); + let mapping = service + .store + .lookup(answer.address, 5432, answer.policy_generation, now) + .unwrap(); + assert_eq!( + mapping.record.contracts[0].pinned_addresses, + ["10.2.3.4".parse::().unwrap()] + ); + } + + const HOST_GATEWAY_POLICY: &str = r" +network_policies: + gateway: + name: gateway + endpoints: + - { host: host.openshell.internal, port: 8080, protocol: tcp } + binaries: [{ path: /usr/bin/client }] +filesystem_policy: { include_workdir: true, read_only: [], read_write: [] } +landlock: { compatibility: best_effort } +process: { run_as_user: sandbox, run_as_group: sandbox } +"; + + fn gateway_service( + addresses: Vec, + trusted_host_gateway: Option, + ) -> PolicyDnsService { + service_with_gateway(HOST_GATEWAY_POLICY, addresses, trusted_host_gateway) + } + + #[tokio::test] + async fn reserved_gateway_alias_without_trusted_address_never_queries_resolver() { + for alias in [ + "host.openshell.internal", + "host.containers.internal", + "host.docker.internal", + ] { + let yaml = HOST_GATEWAY_POLICY.replace("host.openshell.internal", alias); + let service = service_with_gateway(&yaml, vec!["169.254.1.2".parse().unwrap()], None); + + let result = service + .answer_query(alias, AddressFamily::Ipv4, Instant::now()) + .await; + + assert!(matches!( + result, + Err(PolicyDnsError::TrustedGatewayUnavailable) + )); + assert_eq!(service.resolver.calls.load(Ordering::SeqCst), 0); + } + } + + #[tokio::test] + async fn reserved_gateway_alias_pins_only_the_exact_trusted_address() { + let trusted: IpAddr = "169.254.1.2".parse().unwrap(); + let service = gateway_service( + vec![ + "169.254.169.254".parse().unwrap(), + "169.254.1.3".parse().unwrap(), + "10.2.3.4".parse().unwrap(), + trusted, + ], + Some(trusted), + ); + let now = Instant::now(); + + let answer = service + .answer_query("host.openshell.internal", AddressFamily::Ipv4, now) + .await + .unwrap(); + let mapping = service + .store + .lookup(answer.address, 8080, answer.policy_generation, now) + .unwrap(); + + assert_eq!(mapping.record.contracts[0].pinned_addresses, [trusted]); + } + + #[tokio::test] + async fn reserved_gateway_alias_rejects_mismatch_metadata_private_and_wrong_family_answers() { + let trusted: IpAddr = "169.254.1.2".parse().unwrap(); + for (family, address) in [ + (AddressFamily::Ipv4, "169.254.1.3"), + (AddressFamily::Ipv4, "169.254.169.254"), + (AddressFamily::Ipv4, "10.2.3.4"), + (AddressFamily::Ipv6, "fe80::2"), + ] { + let service = gateway_service(vec![address.parse().unwrap()], Some(trusted)); + let result = service + .answer_query("host.openshell.internal", family, Instant::now()) + .await; + assert!( + matches!(result, Err(PolicyDnsError::NoValidAddress)), + "{address} must not satisfy the trusted gateway contract" + ); + } + } + + struct BlockingResolver { + started: Arc, + release: Arc, + } + + impl TrustedResolver for BlockingResolver { + async fn resolve( + &self, + _name: &NormalizedName, + _family: AddressFamily, + ) -> Result { + self.started.notify_one(); + self.release.notified().await; + Ok(TrustedAnswer { + addresses: vec!["8.8.8.8".parse().unwrap()], + ttl: Duration::from_secs(10), + }) + } + } + + #[tokio::test] + async fn delayed_stale_resolution_cannot_replace_newer_generation_mapping() { + let policy = Arc::new( + OpaEngine::from_strings(include_str!("../../data/sandbox-policy.rego"), BASE_POLICY) + .unwrap(), + ); + let pools = SyntheticPools::new( + Ipv4Addr::new(198, 18, 0, 1)..=Ipv4Addr::new(198, 18, 0, 2), + "fd00:1::1".parse::().unwrap()..="fd00:1::2".parse::().unwrap(), + ) + .unwrap(); + let store = Arc::new(ResolvedEndpointStore::new( + StoreConfig::new(pools, 4).unwrap(), + )); + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let service = Arc::new(PolicyDnsService::new( + policy.clone(), + BlockingResolver { + started: started.clone(), + release: release.clone(), + }, + store.clone(), + None, + )); + let query = tokio::spawn(async move { + service + .answer_query("db.example", AddressFamily::Ipv4, Instant::now()) + .await + }); + started.notified().await; + policy + .reload(include_str!("../../data/sandbox-policy.rego"), BASE_POLICY) + .unwrap(); + let current_service = PolicyDnsService::new( + policy.clone(), + FakeResolver { + calls: AtomicUsize::new(0), + answer: TrustedAnswer { + addresses: vec!["8.8.4.4".parse().unwrap()], + ttl: Duration::from_secs(10), + }, + }, + store.clone(), + None, + ); + let now = Instant::now(); + let current = current_service + .answer_query("db.example", AddressFamily::Ipv4, now) + .await + .unwrap(); + release.notify_one(); + assert!(matches!( + query.await.unwrap(), + Err(PolicyDnsError::StalePolicy) + )); + let mapping = store + .lookup(current.address, 5432, current.policy_generation, now) + .unwrap(); + assert_eq!( + mapping.record.policy_generation, + policy.current_generation() + ); + assert_eq!( + mapping.record.contracts[0].pinned_addresses, + ["8.8.4.4".parse::().unwrap()] + ); + let metrics = store.metrics(now); + assert_eq!(metrics.active_mappings, 1); + assert_eq!(metrics.allocated_identities, 1); + } + + #[test] + fn ttl_is_floored_and_capped() { + assert_eq!(clamp_mapping_ttl(Duration::ZERO), MIN_MAPPING_TTL); + assert_eq!( + clamp_mapping_ttl(Duration::from_secs(10)), + Duration::from_secs(10) + ); + assert_eq!(clamp_mapping_ttl(Duration::from_secs(300)), MAX_MAPPING_TTL); + } +} diff --git a/crates/openshell-supervisor-network/src/policy_dns/name.rs b/crates/openshell-supervisor-network/src/policy_dns/name.rs new file mode 100644 index 0000000000..a2c9acb66d --- /dev/null +++ b/crates/openshell-supervisor-network/src/policy_dns/name.rs @@ -0,0 +1,72 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Canonical DNS-name handling for policy lookup and correlation keys. + +use hickory_proto::rr::Name; +use std::fmt; + +/// A lower-case absolute DNS name without its presentation trailing dot. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub(crate) struct NormalizedName(String); + +impl NormalizedName { + pub(crate) fn parse(raw: &str) -> Result { + let trimmed = raw.trim(); + if trimmed.is_empty() || trimmed.parse::().is_ok() { + return Err(NameError); + } + + let absolute = if trimmed.ends_with('.') { + trimmed.to_string() + } else { + format!("{trimmed}.") + }; + let parsed = Name::from_ascii(&absolute).map_err(|_| NameError)?; + if parsed.is_root() { + return Err(NameError); + } + + let normalized = parsed.to_ascii().trim_end_matches('.').to_ascii_lowercase(); + if normalized.is_empty() { + return Err(NameError); + } + Ok(Self(normalized)) + } + + pub(crate) fn as_str(&self) -> &str { + &self.0 + } + + pub(crate) fn as_absolute_name(&self) -> Name { + Name::from_ascii(format!("{}.", self.0)).expect("normalized DNS name must remain valid") + } +} + +impl fmt::Display for NormalizedName { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.0) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct NameError; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn normalizes_case_and_trailing_dot() { + let lower = NormalizedName::parse("Db.Example.COM.").unwrap(); + assert_eq!(lower.as_str(), "db.example.com"); + assert_eq!(NormalizedName::parse("db.example.com").unwrap(), lower); + } + + #[test] + fn rejects_empty_root_ip_literals_and_invalid_labels() { + for raw in ["", ".", "192.0.2.10", "2001:db8::1", "bad name.example"] { + assert!(NormalizedName::parse(raw).is_err(), "accepted {raw:?}"); + } + } +} diff --git a/crates/openshell-supervisor-network/src/policy_dns/resolver.rs b/crates/openshell-supervisor-network/src/policy_dns/resolver.rs new file mode 100644 index 0000000000..8a31f77856 --- /dev/null +++ b/crates/openshell-supervisor-network/src/policy_dns/resolver.rs @@ -0,0 +1,479 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Bounded DNS exchange with an explicitly configured trusted resolver. +//! +//! This module never reads sandbox resolver state or `/etc/hosts`. The caller +//! supplies an already-parsed resolver socket address, and Hickory owns all DNS +//! wire encoding and decoding. + +use super::name::NormalizedName; +use hickory_proto::op::{Message, MessageType, OpCode, Query, ResponseCode}; +use hickory_proto::rr::{Name, RData, RecordType}; +use openshell_core::net::connect_tcp_nodelay_best_effort; +use std::collections::{BTreeMap, BTreeSet}; +use std::net::{IpAddr, SocketAddr}; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::UdpSocket; +use tokio::time::timeout; + +pub(crate) const DEFAULT_EXCHANGE_TIMEOUT: Duration = Duration::from_secs(2); +pub(crate) const MAX_DNS_MESSAGE_BYTES: usize = 8 * 1024; +pub(crate) const MAX_RETAINED_ADDRESSES: usize = 16; +pub(crate) const MAX_CNAME_HOPS: usize = 8; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub(crate) enum AddressFamily { + Ipv4, + Ipv6, +} + +impl AddressFamily { + pub(crate) fn record_type(self) -> RecordType { + match self { + Self::Ipv4 => RecordType::A, + Self::Ipv6 => RecordType::AAAA, + } + } + + pub(crate) fn accepts(self, address: IpAddr) -> bool { + matches!( + (self, address), + (Self::Ipv4, IpAddr::V4(_)) | (Self::Ipv6, IpAddr::V6(_)) + ) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct TrustedAnswer { + pub(crate) addresses: Vec, + pub(crate) ttl: Duration, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum ResolveError { + #[error("trusted DNS exchange timed out")] + Timeout, + #[error("trusted DNS I/O failed: {0}")] + Io(#[from] std::io::Error), + #[error("trusted DNS response exceeded the configured size bound")] + Oversized, + #[error("trusted DNS response was malformed or did not match the query")] + Malformed, + #[error("trusted DNS returned NXDOMAIN")] + NxDomain, + #[error("trusted DNS returned response code {0:?}")] + Response(ResponseCode), + #[error("trusted DNS returned no usable address records")] + NoData, + #[error("trusted DNS CNAME chain looped or exceeded the hop limit")] + CnameLimit, +} + +#[allow(async_fn_in_trait)] +pub(crate) trait TrustedResolver: Send + Sync { + async fn resolve( + &self, + name: &NormalizedName, + family: AddressFamily, + ) -> Result; +} + +/// A DNS client pinned to one operator-supplied upstream socket address. +pub(crate) struct SocketTrustedResolver { + server: SocketAddr, + exchange_timeout: Duration, +} + +impl SocketTrustedResolver { + pub(crate) fn new(server: SocketAddr) -> Self { + Self { + server, + exchange_timeout: DEFAULT_EXCHANGE_TIMEOUT, + } + } + + #[cfg(test)] + pub(crate) fn with_timeout(server: SocketAddr, exchange_timeout: Duration) -> Self { + Self { + server, + exchange_timeout, + } + } + + async fn exchange(&self, name: Name, record_type: RecordType) -> Result { + let query = Query::query(name, record_type); + let mut request = Message::query(); + let id = request.metadata.id; + request.metadata.recursion_desired = true; + request.queries.push(query.clone()); + let wire = request.to_vec().map_err(|_| ResolveError::Malformed)?; + + let udp_response = self.udp_exchange(&wire).await?; + let response = parse_response(&udp_response, id, &query)?; + if response.metadata.truncation { + let tcp_response = self.tcp_exchange(&wire).await?; + parse_response(&tcp_response, id, &query) + } else { + Ok(response) + } + } + + async fn udp_exchange(&self, request: &[u8]) -> Result, ResolveError> { + let bind = if self.server.is_ipv4() { + "0.0.0.0:0" + } else { + "[::]:0" + }; + let socket = UdpSocket::bind(bind).await?; + socket.connect(self.server).await?; + + timeout(self.exchange_timeout, socket.send(request)) + .await + .map_err(|_| ResolveError::Timeout)??; + let mut response = vec![0_u8; MAX_DNS_MESSAGE_BYTES + 1]; + let received = timeout(self.exchange_timeout, socket.recv(&mut response)) + .await + .map_err(|_| ResolveError::Timeout)??; + if received > MAX_DNS_MESSAGE_BYTES { + return Err(ResolveError::Oversized); + } + response.truncate(received); + Ok(response) + } + + async fn tcp_exchange(&self, request: &[u8]) -> Result, ResolveError> { + let mut stream = timeout( + self.exchange_timeout, + connect_tcp_nodelay_best_effort(&[self.server]), + ) + .await + .map_err(|_| ResolveError::Timeout)??; + + let request_len = u16::try_from(request.len()).map_err(|_| ResolveError::Oversized)?; + timeout(self.exchange_timeout, stream.write_u16(request_len)) + .await + .map_err(|_| ResolveError::Timeout)??; + timeout(self.exchange_timeout, stream.write_all(request)) + .await + .map_err(|_| ResolveError::Timeout)??; + + let response_len = timeout(self.exchange_timeout, stream.read_u16()) + .await + .map_err(|_| ResolveError::Timeout)?? as usize; + if response_len > MAX_DNS_MESSAGE_BYTES { + return Err(ResolveError::Oversized); + } + let mut response = vec![0_u8; response_len]; + timeout(self.exchange_timeout, stream.read_exact(&mut response)) + .await + .map_err(|_| ResolveError::Timeout)??; + Ok(response) + } +} + +impl TrustedResolver for SocketTrustedResolver { + async fn resolve( + &self, + name: &NormalizedName, + family: AddressFamily, + ) -> Result { + let mut current = name.as_absolute_name(); + let mut visited = BTreeSet::new(); + let mut chain_ttl = u32::MAX; + let mut cname_hops = 0; + visited.insert(canonical_name(¤t)); + + for _ in 0..=MAX_CNAME_HOPS { + let current_key = canonical_name(¤t); + let response = self.exchange(current.clone(), family.record_type()).await?; + let parsed = parse_answer_records(&response, family); + let mut cursor = current_key; + + loop { + if let Some(records) = parsed.addresses.get(&cursor) { + let mut addresses = records + .iter() + .map(|(address, _)| *address) + .collect::>(); + addresses.sort_unstable(); + addresses.dedup(); + addresses.truncate(MAX_RETAINED_ADDRESSES); + let address_ttl = records.iter().map(|(_, ttl)| *ttl).min().unwrap_or(1); + return Ok(TrustedAnswer { + addresses, + ttl: Duration::from_secs(u64::from(chain_ttl.min(address_ttl))), + }); + } + + let Some((target, ttl)) = parsed.cnames.get(&cursor) else { + return Err(ResolveError::NoData); + }; + cname_hops += 1; + if cname_hops > MAX_CNAME_HOPS { + return Err(ResolveError::CnameLimit); + } + chain_ttl = chain_ttl.min(*ttl); + let target_key = canonical_name(target); + if !visited.insert(target_key.clone()) { + return Err(ResolveError::CnameLimit); + } + cursor = target_key; + + if !parsed.addresses.contains_key(&cursor) && !parsed.cnames.contains_key(&cursor) { + current = target.clone(); + break; + } + } + } + + Err(ResolveError::CnameLimit) + } +} + +fn parse_response(wire: &[u8], id: u16, query: &Query) -> Result { + if wire.len() > MAX_DNS_MESSAGE_BYTES { + return Err(ResolveError::Oversized); + } + let response = Message::from_vec(wire).map_err(|_| ResolveError::Malformed)?; + if response.metadata.id != id + || response.metadata.message_type != MessageType::Response + || response.metadata.op_code != OpCode::Query + || response.queries.len() != 1 + || response.queries.first() != Some(query) + { + return Err(ResolveError::Malformed); + } + match response.metadata.response_code { + ResponseCode::NoError => Ok(response), + ResponseCode::NXDomain => Err(ResolveError::NxDomain), + code => Err(ResolveError::Response(code)), + } +} + +struct ParsedRecords { + addresses: BTreeMap>, + cnames: BTreeMap, +} + +fn parse_answer_records(message: &Message, family: AddressFamily) -> ParsedRecords { + let mut parsed = ParsedRecords { + addresses: BTreeMap::new(), + cnames: BTreeMap::new(), + }; + + for record in &message.answers { + let owner = canonical_name(&record.name); + match &record.data { + RData::A(value) if family == AddressFamily::Ipv4 => { + parsed + .addresses + .entry(owner) + .or_default() + .push((IpAddr::V4(value.0), record.ttl)); + } + RData::AAAA(value) if family == AddressFamily::Ipv6 => { + parsed + .addresses + .entry(owner) + .or_default() + .push((IpAddr::V6(value.0), record.ttl)); + } + RData::CNAME(target) => { + parsed + .cnames + .entry(owner) + .or_insert_with(|| (target.0.clone(), record.ttl)); + } + _ => {} + } + } + parsed +} + +fn canonical_name(name: &Name) -> String { + name.to_ascii().trim_end_matches('.').to_ascii_lowercase() +} + +#[cfg(test)] +mod tests { + use super::*; + use hickory_proto::rr::Record; + use hickory_proto::rr::rdata::{A, CNAME}; + use openshell_core::net::set_tcp_nodelay_best_effort; + use tokio::net::TcpListener; + + #[test] + fn answer_parser_keeps_only_requested_family_and_bounds_are_constants() { + let owner = Name::from_ascii("db.example.").unwrap(); + let mut message = Message::response(1, OpCode::Query); + message.add_answer(Record::from_rdata( + owner.clone(), + 120, + RData::A(A::new(203, 0, 113, 10)), + )); + message.add_answer(Record::from_rdata( + owner, + 120, + RData::AAAA("2001:db8::10".parse::().unwrap().into()), + )); + + let parsed = parse_answer_records(&message, AddressFamily::Ipv4); + assert_eq!(parsed.addresses["db.example"].len(), 1); + assert_eq!(MAX_RETAINED_ADDRESSES, 16); + assert_eq!(MAX_DNS_MESSAGE_BYTES, 8192); + } + + #[test] + fn parser_retains_cname_owner_target_and_ttl() { + let owner = Name::from_ascii("db.example.").unwrap(); + let target = Name::from_ascii("target.example.").unwrap(); + let mut message = Message::response(1, OpCode::Query); + message.add_answer(Record::from_rdata( + owner, + 17, + RData::CNAME(CNAME(target.clone())), + )); + let parsed = parse_answer_records(&message, AddressFamily::Ipv4); + assert_eq!(parsed.cnames["db.example"], (target, 17)); + } + + #[test] + fn response_validation_rejects_wrong_transaction_or_question() { + let query = Query::query(Name::from_ascii("db.example.").unwrap(), RecordType::A); + let mut response = Message::response(9, OpCode::Query); + response.queries.push(query.clone()); + let wire = response.to_vec().unwrap(); + assert!(matches!( + parse_response(&wire, 10, &query), + Err(ResolveError::Malformed) + )); + } + + #[tokio::test] + async fn truncated_udp_retries_over_tcp_and_follows_cname() { + let udp = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let server = udp.local_addr().unwrap(); + let tcp = TcpListener::bind(server).await.unwrap(); + + let udp_task = tokio::spawn(async move { + let mut wire = [0_u8; MAX_DNS_MESSAGE_BYTES]; + let (length, peer) = udp.recv_from(&mut wire).await.unwrap(); + let request = Message::from_vec(&wire[..length]).unwrap(); + let mut response = Message::response(request.metadata.id, OpCode::Query); + response.metadata.truncation = true; + response.queries = request.queries; + udp.send_to(&response.to_vec().unwrap(), peer) + .await + .unwrap(); + }); + let tcp_task = tokio::spawn(async move { + let (mut stream, _) = tcp.accept().await.unwrap(); + set_tcp_nodelay_best_effort(&stream); + let length = stream.read_u16().await.unwrap() as usize; + let mut wire = vec![0_u8; length]; + stream.read_exact(&mut wire).await.unwrap(); + let request = Message::from_vec(&wire).unwrap(); + let requested = request.queries[0].name.clone(); + let canonical = Name::from_ascii("canonical.example.").unwrap(); + let mut response = Message::response(request.metadata.id, OpCode::Query); + response.queries = request.queries; + response.add_answer(Record::from_rdata( + requested, + 12, + RData::CNAME(CNAME(canonical.clone())), + )); + response.add_answer(Record::from_rdata( + canonical, + 20, + RData::A(A::new(8, 8, 8, 8)), + )); + let wire = response.to_vec().unwrap(); + stream + .write_u16(u16::try_from(wire.len()).unwrap()) + .await + .unwrap(); + stream.write_all(&wire).await.unwrap(); + }); + + let resolver = SocketTrustedResolver::new(server); + let answer = resolver + .resolve( + &NormalizedName::parse("db.example").unwrap(), + AddressFamily::Ipv4, + ) + .await + .unwrap(); + assert_eq!(answer.addresses, ["8.8.8.8".parse::().unwrap()]); + assert_eq!(answer.ttl, Duration::from_secs(12)); + udp_task.await.unwrap(); + tcp_task.await.unwrap(); + } + + #[tokio::test] + async fn cname_hop_overflow_fails_closed() { + let udp = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let server = udp.local_addr().unwrap(); + let task = tokio::spawn(async move { + let mut wire = [0_u8; MAX_DNS_MESSAGE_BYTES]; + let (length, peer) = udp.recv_from(&mut wire).await.unwrap(); + let request = Message::from_vec(&wire[..length]).unwrap(); + let mut response = Message::response(request.metadata.id, OpCode::Query); + response.queries = request.queries.clone(); + let mut owner = request.queries[0].name.clone(); + for index in 0..=MAX_CNAME_HOPS { + let target = Name::from_ascii(format!("hop{index}.example.")).unwrap(); + response.add_answer(Record::from_rdata( + owner, + 10, + RData::CNAME(CNAME(target.clone())), + )); + owner = target; + } + response.add_answer(Record::from_rdata(owner, 10, RData::A(A::new(8, 8, 8, 8)))); + udp.send_to(&response.to_vec().unwrap(), peer) + .await + .unwrap(); + }); + let resolver = SocketTrustedResolver::new(server); + assert!(matches!( + resolver + .resolve( + &NormalizedName::parse("db.example").unwrap(), + AddressFamily::Ipv4 + ) + .await, + Err(ResolveError::CnameLimit) + )); + task.await.unwrap(); + } + + #[tokio::test] + async fn trusted_exchange_timeout_is_bounded() { + let udp = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let resolver = SocketTrustedResolver::with_timeout( + udp.local_addr().unwrap(), + Duration::from_millis(10), + ); + assert!(matches!( + resolver + .resolve( + &NormalizedName::parse("db.example").unwrap(), + AddressFamily::Ipv4 + ) + .await, + Err(ResolveError::Timeout) + )); + drop(udp); + } + + #[test] + fn oversized_response_is_rejected_before_decode() { + let query = Query::query(Name::from_ascii("db.example.").unwrap(), RecordType::A); + assert!(matches!( + parse_response(&vec![0; MAX_DNS_MESSAGE_BYTES + 1], 1, &query), + Err(ResolveError::Oversized) + )); + } +} diff --git a/crates/openshell-supervisor-network/src/policy_dns/store.rs b/crates/openshell-supervisor-network/src/policy_dns/store.rs new file mode 100644 index 0000000000..cc9d9f9680 --- /dev/null +++ b/crates/openshell-supervisor-network/src/policy_dns/store.rs @@ -0,0 +1,732 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Synthetic-address allocation and resolved endpoint mappings. + +use super::name::NormalizedName; +use super::resolver::AddressFamily; +use crate::proxy::destination::{ + DestinationRequest, DestinationValidationPlan, UpstreamConnector, build_pinned_validation_plan, + validate_destination, +}; +use std::collections::{BTreeMap, BTreeSet}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::ops::RangeInclusive; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; +use uuid::Uuid; + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub(crate) struct PolicyEndpointId { + pub(crate) policy_name: String, + pub(crate) endpoint_index: usize, +} + +#[derive(Debug, Clone)] +pub(crate) struct ResolvedPortContract { + pub(crate) endpoint_id: PolicyEndpointId, + pub(crate) port: u16, + pub(crate) destination_plan: DestinationValidationPlan, + pub(crate) pinned_addresses: Vec, +} + +#[derive(Debug, Clone)] +pub(crate) struct PublishRequest { + pub(crate) normalized_name: NormalizedName, + pub(crate) family: AddressFamily, + /// Digest of every compatible endpoint identity and its policy metadata. + /// A changed endpoint contract receives a new synthetic identity even when + /// it selects the same normalized name. + pub(crate) allocation_identity: [u8; 32], + pub(crate) policy_generation: u64, + pub(crate) ttl: Duration, + pub(crate) contracts: Vec, +} + +#[derive(Debug, Clone)] +pub(crate) struct ResolvedEndpointRecord { + pub(crate) synthetic_address: IpAddr, + pub(crate) normalized_name: NormalizedName, + pub(crate) family: AddressFamily, + pub(crate) policy_generation: u64, + pub(crate) mapping_generation: u64, + pub(crate) mapping_id: Uuid, + pub(crate) created_at: Instant, + pub(crate) expires_at: Instant, + pub(crate) contracts: Vec, +} + +impl ResolvedEndpointRecord { + pub(crate) fn allowed_ports(&self) -> BTreeSet { + self.contracts + .iter() + .map(|contract| contract.port) + .collect() + } +} + +#[derive(Debug, Clone)] +pub(crate) struct MappingLookup { + pub(crate) record: ResolvedEndpointRecord, + pub(crate) port: u16, +} + +impl MappingLookup { + pub(crate) fn endpoint_ids(&self) -> impl Iterator { + self.record + .contracts + .iter() + .filter(move |contract| contract.port == self.port) + .map(|contract| &contract.endpoint_id) + } + + /// Build the unopened connector for a process-authorized endpoint. + /// + /// Selecting by endpoint identity prevents a compatible endpoint record + /// from becoming a new policy precedence rule. The pinned destination mode + /// never resolves `normalized_name` again. + pub(crate) async fn connector_for( + &self, + endpoint_id: &PolicyEndpointId, + ) -> Result { + 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()) + .collect::>() + .into_iter() + .collect::>(); + if addresses.is_empty() { + return Err(MappingLookupError::EndpointMismatch); + } + let plan = build_pinned_validation_plan(addresses) + .map_err(|_| MappingLookupError::InvalidMapping)?; + validate_destination(DestinationRequest { + host: self.record.normalized_name.as_str(), + port: self.port, + sandbox_entrypoint_pid: 0, + plan: &plan, + }) + .await + .map_err(|_| MappingLookupError::InvalidMapping) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct SyntheticPools { + ipv4: RangeInclusive, + ipv6: RangeInclusive, +} + +impl SyntheticPools { + /// Construct injectable pools. Production runtime ranges are deliberately + /// selected only after PR3 checks route collisions in each namespace. + pub(crate) fn new( + ipv4: RangeInclusive, + ipv6: RangeInclusive, + ) -> Result { + if ipv4.is_empty() || ipv6.is_empty() { + return Err(StoreConfigError::InvalidPool); + } + for address in [IpAddr::V4(*ipv4.start()), IpAddr::V4(*ipv4.end())] { + if openshell_core::net::is_always_blocked_ip(address) { + return Err(StoreConfigError::InvalidPool); + } + } + for address in [IpAddr::V6(*ipv6.start()), IpAddr::V6(*ipv6.end())] { + if openshell_core::net::is_always_blocked_ip(address) { + return Err(StoreConfigError::InvalidPool); + } + } + Ok(Self { ipv4, ipv6 }) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct StoreConfig { + pub(crate) pools: SyntheticPools, + pub(crate) max_mappings: usize, +} + +impl StoreConfig { + pub(crate) fn new( + pools: SyntheticPools, + max_mappings: usize, + ) -> Result { + if max_mappings == 0 { + return Err(StoreConfigError::ZeroCapacity); + } + Ok(Self { + pools, + max_mappings, + }) + } +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum StoreConfigError { + #[error("synthetic address pool is empty or contains an always-blocked boundary")] + InvalidPool, + #[error("resolved endpoint store capacity must be non-zero")] + ZeroCapacity, +} + +#[derive(Debug, thiserror::Error, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PublishError { + #[error("policy generation changed before mapping publication")] + StalePolicy, + #[error("resolved endpoint publication was empty or invalid")] + InvalidMapping, + #[error("synthetic address pool is exhausted")] + PoolExhausted, + #[error("resolved endpoint store lock was poisoned")] + LockPoisoned, +} + +#[derive(Debug, thiserror::Error, Clone, Copy, PartialEq, Eq)] +pub(crate) enum MappingLookupError { + #[error("transparent TCP mapping is missing")] + Missing, + #[error("transparent TCP mapping is expired")] + Expired, + #[error("transparent TCP mapping belongs to a stale policy generation")] + StalePolicy, + #[error("transparent TCP mapping does not authorize the requested port")] + PortMismatch, + #[error("transparent TCP mapping does not contain the authorized endpoint")] + EndpointMismatch, + #[error("transparent TCP mapping is internally invalid")] + InvalidMapping, + #[error("resolved endpoint store lock was poisoned")] + LockPoisoned, +} + +#[derive(Default)] +struct PolicyDnsMetrics { + queries: AtomicU64, + refused: AtomicU64, + upstream_queries: AtomicU64, + no_valid_address: AtomicU64, + mappings_published: AtomicU64, + mappings_expired: AtomicU64, + pool_exhausted: AtomicU64, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct PolicyDnsMetricsSnapshot { + pub(crate) queries: u64, + pub(crate) refused: u64, + pub(crate) upstream_queries: u64, + pub(crate) no_valid_address: u64, + pub(crate) mappings_published: u64, + pub(crate) mappings_expired: u64, + pub(crate) pool_exhausted: u64, + pub(crate) active_mappings: usize, + pub(crate) allocated_identities: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +struct AllocationKey { + normalized_name: NormalizedName, + family: AddressFamily, + allocation_identity: [u8; 32], +} + +struct StoreState { + records: BTreeMap, + allocations: BTreeMap, + expired_allocations: BTreeSet, + next_ipv4: u32, + end_ipv4: u32, + next_ipv6: u128, + end_ipv6: u128, + next_mapping_generation: u64, +} + +pub(crate) struct ResolvedEndpointStore { + state: RwLock, + config: StoreConfig, + metrics: Arc, +} + +impl ResolvedEndpointStore { + pub(crate) fn new(config: StoreConfig) -> Self { + let next_ipv4 = u32::from(*config.pools.ipv4.start()); + let end_ipv4 = u32::from(*config.pools.ipv4.end()); + let next_ipv6 = u128::from(*config.pools.ipv6.start()); + let end_ipv6 = u128::from(*config.pools.ipv6.end()); + Self { + state: RwLock::new(StoreState { + records: BTreeMap::new(), + allocations: BTreeMap::new(), + expired_allocations: BTreeSet::new(), + next_ipv4, + end_ipv4, + next_ipv6, + end_ipv6, + next_mapping_generation: 0, + }), + config, + metrics: Arc::new(PolicyDnsMetrics::default()), + } + } + + pub(crate) fn note_query(&self) { + self.metrics.queries.fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn note_refused(&self) { + self.metrics.refused.fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn note_upstream_query(&self) { + self.metrics + .upstream_queries + .fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn note_no_valid_address(&self) { + self.metrics + .no_valid_address + .fetch_add(1, Ordering::Relaxed); + } + + pub(crate) fn publish( + &self, + request: PublishRequest, + current_policy_generation: u64, + now: Instant, + ) -> Result { + if request.policy_generation != current_policy_generation { + return Err(PublishError::StalePolicy); + } + if request.ttl.is_zero() + || request.contracts.is_empty() + || request.contracts.iter().any(|contract| { + contract.port == 0 + || contract.pinned_addresses.is_empty() + || contract + .pinned_addresses + .iter() + .any(|address| !request.family.accepts(*address)) + }) + { + return Err(PublishError::InvalidMapping); + } + + let key = AllocationKey { + normalized_name: request.normalized_name.clone(), + family: request.family, + allocation_identity: request.allocation_identity, + }; + let mut state = self.state.write().map_err(|_| PublishError::LockPoisoned)?; + let synthetic_address = if let Some(address) = state.allocations.get(&key) { + *address + } else { + if state.allocations.len() >= self.config.max_mappings { + self.metrics.pool_exhausted.fetch_add(1, Ordering::Relaxed); + return Err(PublishError::PoolExhausted); + } + let address = allocate_address(&mut state, request.family).ok_or_else(|| { + self.metrics.pool_exhausted.fetch_add(1, Ordering::Relaxed); + PublishError::PoolExhausted + })?; + state.allocations.insert(key, address); + address + }; + + // Defense in depth for callers outside the OPA generation guard: a + // delayed publication from an older generation must never replace a + // newer live correlation for the same stable allocation identity. + if state + .records + .get(&synthetic_address) + .is_some_and(|record| record.policy_generation > request.policy_generation) + { + return Err(PublishError::StalePolicy); + } + + state.next_mapping_generation = state.next_mapping_generation.saturating_add(1); + let record = ResolvedEndpointRecord { + synthetic_address, + normalized_name: request.normalized_name, + family: request.family, + policy_generation: request.policy_generation, + mapping_generation: state.next_mapping_generation, + mapping_id: Uuid::new_v4(), + created_at: now, + expires_at: now + request.ttl, + contracts: request.contracts, + }; + state.expired_allocations.remove(&synthetic_address); + state.records.insert(synthetic_address, record.clone()); + self.metrics + .mappings_published + .fetch_add(1, Ordering::Relaxed); + Ok(record) + } + + pub(crate) fn lookup( + &self, + synthetic_address: IpAddr, + port: u16, + current_policy_generation: u64, + now: Instant, + ) -> Result { + let state = self + .state + .read() + .map_err(|_| MappingLookupError::LockPoisoned)?; + let Some(record) = state.records.get(&synthetic_address) else { + return if state.expired_allocations.contains(&synthetic_address) { + Err(MappingLookupError::Expired) + } else { + Err(MappingLookupError::Missing) + }; + }; + if now >= record.expires_at { + return Err(MappingLookupError::Expired); + } + if record.policy_generation != current_policy_generation { + return Err(MappingLookupError::StalePolicy); + } + if !record + .contracts + .iter() + .any(|contract| contract.port == port) + { + return Err(MappingLookupError::PortMismatch); + } + Ok(MappingLookup { + record: record.clone(), + port, + }) + } + + /// Remove expired active records without freeing their synthetic identity. + pub(crate) fn expire(&self, now: Instant) -> Result { + let mut state = self + .state + .write() + .map_err(|_| MappingLookupError::LockPoisoned)?; + let expired = state + .records + .iter() + .filter_map(|(address, record)| (now >= record.expires_at).then_some(*address)) + .collect::>(); + for address in &expired { + state.records.remove(address); + state.expired_allocations.insert(*address); + } + self.metrics + .mappings_expired + .fetch_add(expired.len() as u64, Ordering::Relaxed); + Ok(expired.len()) + } + + pub(crate) fn metrics(&self, now: Instant) -> PolicyDnsMetricsSnapshot { + let state = self + .state + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + PolicyDnsMetricsSnapshot { + queries: self.metrics.queries.load(Ordering::Relaxed), + refused: self.metrics.refused.load(Ordering::Relaxed), + upstream_queries: self.metrics.upstream_queries.load(Ordering::Relaxed), + no_valid_address: self.metrics.no_valid_address.load(Ordering::Relaxed), + mappings_published: self.metrics.mappings_published.load(Ordering::Relaxed), + mappings_expired: self.metrics.mappings_expired.load(Ordering::Relaxed), + pool_exhausted: self.metrics.pool_exhausted.load(Ordering::Relaxed), + active_mappings: state + .records + .values() + .filter(|record| now < record.expires_at) + .count(), + allocated_identities: state.allocations.len(), + } + } +} + +fn allocate_address(state: &mut StoreState, family: AddressFamily) -> Option { + match family { + AddressFamily::Ipv4 if state.next_ipv4 <= state.end_ipv4 => { + let address = IpAddr::V4(Ipv4Addr::from(state.next_ipv4)); + state.next_ipv4 = state.next_ipv4.saturating_add(1); + Some(address) + } + AddressFamily::Ipv6 if state.next_ipv6 <= state.end_ipv6 => { + let address = IpAddr::V6(Ipv6Addr::from(state.next_ipv6)); + state.next_ipv6 = state.next_ipv6.saturating_add(1); + Some(address) + } + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::proxy::destination::{AddressAuthorization, DestinationValidationPlan}; + use std::sync::Barrier; + + fn store(max_mappings: usize) -> ResolvedEndpointStore { + let pools = SyntheticPools::new( + Ipv4Addr::new(198, 18, 0, 1)..=Ipv4Addr::new(198, 18, 0, 2), + "fd00:1::1".parse().unwrap()..="fd00:1::2".parse().unwrap(), + ) + .unwrap(); + ResolvedEndpointStore::new(StoreConfig::new(pools, max_mappings).unwrap()) + } + + fn request(name: &str, generation: u64, ttl: Duration) -> PublishRequest { + PublishRequest { + normalized_name: NormalizedName::parse(name).unwrap(), + family: AddressFamily::Ipv4, + allocation_identity: [1; 32], + policy_generation: generation, + ttl, + contracts: vec![ResolvedPortContract { + endpoint_id: PolicyEndpointId { + policy_name: "database".to_string(), + endpoint_index: 0, + }, + port: 5432, + destination_plan: DestinationValidationPlan { + address_authorization: AddressAuthorization::ExactDeclaredHost, + }, + pinned_addresses: vec!["203.0.113.8".parse().unwrap()], + }], + } + } + + #[test] + fn refresh_retains_synthetic_identity_and_changes_mapping_generation() { + let store = store(2); + let now = Instant::now(); + let first = store + .publish(request("db.example", 7, Duration::from_secs(10)), 7, now) + .unwrap(); + let second = store + .publish( + request("DB.EXAMPLE.", 7, Duration::from_secs(20)), + 7, + now + Duration::from_secs(1), + ) + .unwrap(); + assert_eq!(first.synthetic_address, second.synthetic_address); + assert_ne!(first.mapping_id, second.mapping_id); + assert!(second.mapping_generation > first.mapping_generation); + assert_eq!(first.policy_generation, second.policy_generation); + } + + #[test] + fn distinct_names_sharing_real_ip_get_distinct_correlations() { + let store = store(2); + let now = Instant::now(); + let left = store + .publish(request("left.example", 1, Duration::from_secs(10)), 1, now) + .unwrap(); + let right = store + .publish(request("right.example", 1, Duration::from_secs(10)), 1, now) + .unwrap(); + assert_ne!(left.synthetic_address, right.synthetic_address); + assert_eq!( + left.contracts[0].pinned_addresses, + right.contracts[0].pinned_addresses + ); + } + + #[test] + fn wrong_port_stale_generation_and_expiry_fail_closed() { + let store = store(2); + let now = Instant::now(); + let record = store + .publish(request("db.example", 4, Duration::from_secs(2)), 4, now) + .unwrap(); + assert!(matches!( + store.lookup(record.synthetic_address, 3306, 4, now), + Err(MappingLookupError::PortMismatch) + )); + assert!(matches!( + store.lookup(record.synthetic_address, 5432, 5, now), + Err(MappingLookupError::StalePolicy) + )); + assert!(matches!( + store.lookup( + record.synthetic_address, + 5432, + 4, + now + Duration::from_secs(2) + ), + Err(MappingLookupError::Expired) + )); + } + + #[test] + fn expiry_never_reassigns_synthetic_address_to_another_name() { + let store = store(2); + let now = Instant::now(); + let first = store + .publish(request("first.example", 1, Duration::from_secs(1)), 1, now) + .unwrap(); + assert_eq!(store.expire(now + Duration::from_secs(1)).unwrap(), 1); + let second = store + .publish( + request("second.example", 1, Duration::from_secs(10)), + 1, + now + Duration::from_secs(1), + ) + .unwrap(); + assert_ne!(first.synthetic_address, second.synthetic_address); + assert!(matches!( + store.lookup( + first.synthetic_address, + 5432, + 1, + now + Duration::from_secs(1) + ), + Err(MappingLookupError::Expired) + )); + } + + #[test] + fn changed_endpoint_contract_never_reuses_synthetic_identity() { + let store = store(2); + let now = Instant::now(); + let first = store + .publish(request("db.example", 1, Duration::from_secs(5)), 1, now) + .unwrap(); + let mut changed = request("db.example", 2, Duration::from_secs(5)); + changed.allocation_identity = [2; 32]; + let second = store.publish(changed, 2, now).unwrap(); + assert_ne!(first.synthetic_address, second.synthetic_address); + assert!(matches!( + store.lookup(first.synthetic_address, 5432, 2, now), + Err(MappingLookupError::StalePolicy) + )); + } + + #[test] + fn pool_exhaustion_and_stale_publication_publish_nothing() { + let store = store(1); + let now = Instant::now(); + assert!(matches!( + store.publish(request("stale.example", 1, Duration::from_secs(5)), 2, now), + Err(PublishError::StalePolicy) + )); + store + .publish(request("first.example", 2, Duration::from_secs(5)), 2, now) + .unwrap(); + assert!(matches!( + store.publish(request("second.example", 2, Duration::from_secs(5)), 2, now), + Err(PublishError::PoolExhausted) + )); + let metrics = store.metrics(now); + assert_eq!(metrics.allocated_identities, 1); + assert_eq!(metrics.active_mappings, 1); + assert_eq!(metrics.pool_exhausted, 1); + } + + #[test] + fn older_generation_cannot_replace_newer_live_mapping() { + let store = store(1); + let now = Instant::now(); + let newer = store + .publish(request("db.example", 2, Duration::from_secs(10)), 2, now) + .unwrap(); + + assert!(matches!( + store.publish( + request("db.example", 1, Duration::from_secs(10)), + 1, + now + Duration::from_secs(1), + ), + Err(PublishError::StalePolicy) + )); + + let mapping = store.lookup(newer.synthetic_address, 5432, 2, now).unwrap(); + assert_eq!(mapping.record.mapping_id, newer.mapping_id); + assert_eq!(mapping.record.policy_generation, 2); + } + + #[test] + fn real_address_never_inherits_synthetic_mapping() { + let store = store(1); + let now = Instant::now(); + store + .publish(request("db.example", 1, Duration::from_secs(5)), 1, now) + .unwrap(); + assert!(matches!( + store.lookup("203.0.113.8".parse().unwrap(), 5432, 1, now), + Err(MappingLookupError::Missing) + )); + } + + #[test] + fn ipv6_pool_allocates_only_ipv6_synthetic_addresses() { + let store = store(1); + let now = Instant::now(); + let mut request = request("db.example", 1, Duration::from_secs(5)); + request.family = AddressFamily::Ipv6; + request.contracts[0].pinned_addresses = vec!["2001:db8::8".parse().unwrap()]; + let record = store.publish(request, 1, now).unwrap(); + assert!(record.synthetic_address.is_ipv6()); + assert_eq!(record.family, AddressFamily::Ipv6); + } + + #[test] + fn concurrent_refreshes_never_publish_partial_records() { + let store = Arc::new(store(1)); + let barrier = Arc::new(Barrier::new(9)); + let now = Instant::now(); + let mut workers = Vec::new(); + for _ in 0..8 { + let store = store.clone(); + let barrier = barrier.clone(); + workers.push(std::thread::spawn(move || { + barrier.wait(); + store + .publish(request("db.example", 3, Duration::from_secs(5)), 3, now) + .unwrap() + })); + } + barrier.wait(); + let records = workers + .into_iter() + .map(|worker| worker.join().unwrap()) + .collect::>(); + assert!( + records + .iter() + .all(|record| record.synthetic_address == records[0].synthetic_address) + ); + let lookup = store + .lookup(records[0].synthetic_address, 5432, 3, now) + .unwrap(); + assert!(!lookup.record.contracts.is_empty()); + assert!(!lookup.record.contracts[0].pinned_addresses.is_empty()); + assert_eq!(store.metrics(now).allocated_identities, 1); + } + + #[tokio::test] + async fn connector_uses_only_pinned_addresses_and_endpoint_identity() { + let store = store(1); + let now = Instant::now(); + let record = store + .publish( + request("must-not-resolve.invalid", 1, Duration::from_secs(5)), + 1, + now, + ) + .unwrap(); + let lookup = store + .lookup(record.synthetic_address, 5432, 1, now) + .unwrap(); + let endpoint = lookup.endpoint_ids().next().unwrap().clone(); + let connector = lookup.connector_for(&endpoint).await.unwrap(); + assert_eq!(connector.addrs(), &["203.0.113.8:5432".parse().unwrap()]); + } +} diff --git a/crates/openshell-supervisor-network/src/policy_dns/wire.rs b/crates/openshell-supervisor-network/src/policy_dns/wire.rs new file mode 100644 index 0000000000..ecc520b428 --- /dev/null +++ b/crates/openshell-supervisor-network/src/policy_dns/wire.rs @@ -0,0 +1,281 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! DNS request and response wire handling. + +use super::resolver::{AddressFamily, MAX_DNS_MESSAGE_BYTES, ResolveError, TrustedResolver}; +use super::{PolicyDnsError, PolicyDnsService}; +use hickory_proto::op::{Message, MessageType, OpCode, ResponseCode}; +use hickory_proto::rr::rdata::{A, AAAA}; +use hickory_proto::rr::{DNSClass, RData, Record, RecordType}; +use std::net::IpAddr; +use std::time::Instant; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum WireError { + #[error("DNS response encoding failed")] + Encode, + #[error("DNS-over-TCP frame is invalid")] + InvalidTcpFrame, +} + +/// Handle one DNS datagram without binding a runtime listener. +pub(crate) async fn handle_udp_query( + service: &PolicyDnsService, + wire: &[u8], +) -> Result, WireError> { + let fallback_id = wire + .get(..2) + .map(|bytes| u16::from_be_bytes([bytes[0], bytes[1]])) + .unwrap_or_default(); + if wire.len() > MAX_DNS_MESSAGE_BYTES { + return encode_message(Message::error_msg( + fallback_id, + OpCode::Query, + ResponseCode::FormErr, + )); + } + let Ok(request) = Message::from_vec(wire) else { + return encode_message(Message::error_msg( + fallback_id, + OpCode::Query, + ResponseCode::FormErr, + )); + }; + let Some((query, family)) = validate_request(&request) else { + let code = if request.metadata.message_type != MessageType::Query + || request.metadata.op_code != OpCode::Query + || request.queries.len() != 1 + { + ResponseCode::FormErr + } else { + ResponseCode::NotImp + }; + return encode_message(response_with_code(&request, code)); + }; + + let raw_name = query.name.to_ascii(); + match service + .answer_query(&raw_name, family, Instant::now()) + .await + { + Ok(answer) => { + let rdata = match answer.address { + IpAddr::V4(address) if family == AddressFamily::Ipv4 => RData::A(A(address)), + IpAddr::V6(address) if family == AddressFamily::Ipv6 => RData::AAAA(AAAA(address)), + _ => return encode_message(response_with_code(&request, ResponseCode::ServFail)), + }; + let mut response = response_with_code(&request, ResponseCode::NoError); + response.answers.push(Record::from_rdata( + query.name.clone(), + u32::try_from(answer.ttl.as_secs()).unwrap_or(u32::MAX), + rdata, + )); + encode_message(response) + } + Err(PolicyDnsError::Ineligible | PolicyDnsError::TrustedGatewayUnavailable) => { + encode_message(response_with_code(&request, ResponseCode::Refused)) + } + Err(PolicyDnsError::Resolver(ResolveError::NxDomain)) => { + encode_message(response_with_code(&request, ResponseCode::NXDomain)) + } + Err(PolicyDnsError::InvalidName) => { + encode_message(response_with_code(&request, ResponseCode::FormErr)) + } + Err( + PolicyDnsError::Resolver(_) + | PolicyDnsError::NoValidAddress + | PolicyDnsError::StalePolicy + | PolicyDnsError::Publish(_) + | PolicyDnsError::Policy(_), + ) => encode_message(response_with_code(&request, ResponseCode::ServFail)), + } +} + +/// Handle exactly one length-prefixed DNS-over-TCP message. +pub(crate) async fn handle_tcp_query( + service: &PolicyDnsService, + frame: &[u8], +) -> Result, WireError> { + let declared = frame + .get(..2) + .map(|bytes| usize::from(u16::from_be_bytes([bytes[0], bytes[1]]))) + .ok_or(WireError::InvalidTcpFrame)?; + if declared > MAX_DNS_MESSAGE_BYTES || frame.len() != declared + 2 { + return Err(WireError::InvalidTcpFrame); + } + let response = handle_udp_query(service, &frame[2..]).await?; + let length = u16::try_from(response.len()).map_err(|_| WireError::Encode)?; + let mut framed = Vec::with_capacity(response.len() + 2); + framed.extend_from_slice(&length.to_be_bytes()); + framed.extend_from_slice(&response); + Ok(framed) +} + +fn validate_request(request: &Message) -> Option<(&hickory_proto::op::Query, AddressFamily)> { + if request.metadata.message_type != MessageType::Query + || request.metadata.op_code != OpCode::Query + || request.queries.len() != 1 + { + return None; + } + let query = request.queries.first()?; + if query.query_class != DNSClass::IN { + return None; + } + let family = match query.query_type { + RecordType::A => AddressFamily::Ipv4, + RecordType::AAAA => AddressFamily::Ipv6, + _ => return None, + }; + Some((query, family)) +} + +fn response_with_code(request: &Message, code: ResponseCode) -> Message { + let mut response = Message::response(request.metadata.id, request.metadata.op_code); + response.metadata.recursion_desired = request.metadata.recursion_desired; + response.metadata.recursion_available = true; + response.metadata.response_code = code; + response.queries.clone_from(&request.queries); + response +} + +fn encode_message(message: Message) -> Result, WireError> { + message.to_vec().map_err(|_| WireError::Encode) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::opa::OpaEngine; + use crate::policy_dns::name::NormalizedName; + use crate::policy_dns::resolver::TrustedAnswer; + use crate::policy_dns::store::{ResolvedEndpointStore, StoreConfig, SyntheticPools}; + use hickory_proto::op::Query; + use hickory_proto::rr::Name; + use std::net::{Ipv4Addr, Ipv6Addr}; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + struct FakeResolver { + calls: AtomicUsize, + } + + impl TrustedResolver for FakeResolver { + async fn resolve( + &self, + _name: &NormalizedName, + family: AddressFamily, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(TrustedAnswer { + addresses: match family { + AddressFamily::Ipv4 => vec!["8.8.8.8".parse().unwrap()], + AddressFamily::Ipv6 => vec!["2001:4860:4860::8888".parse().unwrap()], + }, + ttl: Duration::from_secs(10), + }) + } + } + + fn service() -> PolicyDnsService { + let yaml = r" +network_policies: + database: + name: database + endpoints: [{ host: db.example, port: 5432, protocol: tcp }] + binaries: [{ path: /usr/bin/psql }] +filesystem_policy: { include_workdir: true, read_only: [], read_write: [] } +landlock: { compatibility: best_effort } +process: { run_as_user: sandbox, run_as_group: sandbox } +"; + let policy = Arc::new( + OpaEngine::from_strings(include_str!("../../data/sandbox-policy.rego"), yaml).unwrap(), + ); + let pools = SyntheticPools::new( + Ipv4Addr::new(198, 18, 0, 1)..=Ipv4Addr::new(198, 18, 0, 4), + "fd00:1::1".parse::().unwrap()..="fd00:1::4".parse::().unwrap(), + ) + .unwrap(); + PolicyDnsService::new( + policy, + FakeResolver { + calls: AtomicUsize::new(0), + }, + Arc::new(ResolvedEndpointStore::new( + StoreConfig::new(pools, 8).unwrap(), + )), + None, + ) + } + + fn request(name: &str, record_type: RecordType) -> Vec { + let mut message = Message::new(42, MessageType::Query, OpCode::Query); + message.metadata.recursion_desired = true; + message + .queries + .push(Query::query(Name::from_ascii(name).unwrap(), record_type)); + message.to_vec().unwrap() + } + + #[tokio::test] + async fn udp_and_tcp_queries_return_synthetic_answers() { + let service = service(); + let udp = handle_udp_query(&service, &request("DB.EXAMPLE.", RecordType::A)) + .await + .unwrap(); + let udp_message = Message::from_vec(&udp).unwrap(); + assert_eq!(udp_message.metadata.response_code, ResponseCode::NoError); + assert!(matches!(udp_message.answers[0].data, RData::A(_))); + + let query = request("db.example.", RecordType::A); + let mut frame = Vec::with_capacity(query.len() + 2); + frame.extend_from_slice(&u16::try_from(query.len()).unwrap().to_be_bytes()); + frame.extend_from_slice(&query); + let tcp = handle_tcp_query(&service, &frame).await.unwrap(); + let declared = usize::from(u16::from_be_bytes([tcp[0], tcp[1]])); + assert_eq!(declared, tcp.len() - 2); + assert_eq!( + Message::from_vec(&tcp[2..]).unwrap().metadata.response_code, + ResponseCode::NoError + ); + + let ipv6 = handle_udp_query(&service, &request("db.example.", RecordType::AAAA)) + .await + .unwrap(); + assert!(matches!( + Message::from_vec(&ipv6).unwrap().answers[0].data, + RData::AAAA(_) + )); + } + + #[tokio::test] + async fn ineligible_query_is_refused_without_upstream_call() { + let service = service(); + let wire = handle_udp_query(&service, &request("other.example.", RecordType::A)) + .await + .unwrap(); + assert_eq!( + Message::from_vec(&wire).unwrap().metadata.response_code, + ResponseCode::Refused + ); + assert_eq!(service.resolver.calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn unsupported_type_is_not_implemented_and_malformed_tcp_is_rejected() { + let service = service(); + let wire = handle_udp_query(&service, &request("db.example.", RecordType::TXT)) + .await + .unwrap(); + assert_eq!( + Message::from_vec(&wire).unwrap().metadata.response_code, + ResponseCode::NotImp + ); + assert!(matches!( + handle_tcp_query(&service, &[0, 10, 1, 2]).await, + Err(WireError::InvalidTcpFrame) + )); + } +} diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 04e91d4230..81c69dca59 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -3,7 +3,7 @@ //! HTTP CONNECT proxy with OPA policy evaluation and process-identity binding. -mod destination; +pub(crate) mod destination; mod egress; mod relay; @@ -2808,7 +2808,7 @@ fn normalize_host_lookup_key(host: &str) -> &str { /// Returns `true` if `host` is one of the well-known driver-injected aliases /// for the host machine (e.g. `host.openshell.internal`). -fn is_host_gateway_alias(host: &str) -> bool { +pub(crate) fn is_host_gateway_alias(host: &str) -> bool { let h = normalize_host_lookup_key(host); HOST_GATEWAY_ALIASES .iter() diff --git a/crates/openshell-supervisor-network/src/proxy/destination.rs b/crates/openshell-supervisor-network/src/proxy/destination.rs index 4f94a08743..1ce514133a 100644 --- a/crates/openshell-supervisor-network/src/proxy/destination.rs +++ b/crates/openshell-supervisor-network/src/proxy/destination.rs @@ -1,21 +1,27 @@ // SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +#![allow( + clippy::redundant_pub_crate, + reason = "the destination primitives intentionally remain internal to the proxy crate" +)] + //! Shared external destination validation and upstream dial boundary. use super::{ - implicit_allowed_ips_for_ip_host, is_host_gateway_alias, parse_allowed_ips, - resolve_and_check_allowed_ips, resolve_and_check_declared_endpoint, - resolve_and_check_trusted_gateway, resolve_and_reject_internal, + BLOCKED_CONTROL_PLANE_PORTS, implicit_allowed_ips_for_ip_host, is_cloud_metadata_ip, + is_host_gateway_alias, is_link_local_ip, parse_allowed_ips, resolve_and_check_allowed_ips, + resolve_and_check_declared_endpoint, resolve_and_check_trusted_gateway, + resolve_and_reject_internal, }; use ipnet::IpNet; -use openshell_core::net::connect_tcp_nodelay_best_effort; +use openshell_core::net::{connect_tcp_nodelay_best_effort, is_always_blocked_ip, is_internal_ip}; use std::net::{IpAddr, SocketAddr}; use tokio::net::TcpStream; /// Address-validation mode selected from the current endpoint configuration. #[derive(Debug, Clone, PartialEq, Eq)] -pub(super) enum AddressAuthorization { +pub(crate) enum AddressAuthorization { DefaultPublicOnly, ExplicitAllowedIps(Vec), ExactDeclaredHost, @@ -32,16 +38,16 @@ pub(super) enum AddressAuthorization { /// Fully materialized input to shared destination validation. #[derive(Debug, Clone, PartialEq, Eq)] -pub(super) struct DestinationValidationPlan { - pub(super) address_authorization: AddressAuthorization, +pub(crate) struct DestinationValidationPlan { + pub(crate) address_authorization: AddressAuthorization, } /// Inputs needed to apply the current SSRF and endpoint destination policy. -pub(super) struct DestinationRequest<'a> { - pub(super) host: &'a str, - pub(super) port: u16, - pub(super) sandbox_entrypoint_pid: u32, - pub(super) plan: &'a DestinationValidationPlan, +pub(crate) struct DestinationRequest<'a> { + pub(crate) host: &'a str, + pub(crate) port: u16, + pub(crate) sandbox_entrypoint_pid: u32, + pub(crate) plan: &'a DestinationValidationPlan, } /// Destination-validation branch that rejected an egress request. @@ -49,7 +55,7 @@ pub(super) struct DestinationRequest<'a> { /// Adapters use this classification to preserve their existing HTTP response /// and OCSF message shapes while sharing the underlying validation logic. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(super) enum DestinationDenialKind { +pub(crate) enum DestinationDenialKind { TrustedGateway, InvalidAllowedIps, AllowedIps, @@ -58,9 +64,9 @@ pub(super) enum DestinationDenialKind { } #[derive(Debug)] -pub(super) struct DestinationDenial { - pub(super) kind: DestinationDenialKind, - pub(super) reason: String, +pub(crate) struct DestinationDenial { + pub(crate) kind: DestinationDenialKind, + pub(crate) reason: String, } impl DestinationDenial { @@ -70,7 +76,7 @@ impl DestinationDenial { } /// Select one current destination-validation mode without changing precedence. -pub(super) fn build_validation_plan( +pub(crate) fn build_validation_plan( host: &str, normalized_host: &str, trusted_host_gateway: Option, @@ -103,8 +109,8 @@ pub(super) fn build_validation_plan( /// Build the destination mode used by policy DNS after it has validated and /// pinned a non-empty answer set for an endpoint. -#[allow(dead_code, reason = "used when the policy DNS adapter lands")] -pub(super) fn build_pinned_validation_plan( +#[allow(dead_code, reason = "used by the policy DNS adapter")] +pub(crate) fn build_pinned_validation_plan( addresses: Vec, ) -> Result { if addresses.is_empty() { @@ -119,25 +125,140 @@ pub(super) fn build_pinned_validation_plan( }) } +/// Filter resolver-provided addresses through a materialized destination plan. +/// +/// This is the address-only policy-DNS boundary: it never reads a hosts file, +/// invokes a system lookup, or otherwise resolves `host`. Unlike CONNECT's +/// all-or-nothing validation, prohibited answers are removed so a trusted DNS +/// response containing both usable and unusable addresses can retain only the +/// usable subset. +#[allow(dead_code, reason = "used by the policy DNS adapter")] +pub(crate) fn filter_resolved_addresses( + plan: &DestinationValidationPlan, + host: &str, + port: u16, + resolved_ips: &[IpAddr], +) -> Result, DestinationDenial> { + let (kind, control_plane_blocked) = match &plan.address_authorization { + AddressAuthorization::TrustedGatewayAlias { .. } => { + (DestinationDenialKind::TrustedGateway, true) + } + AddressAuthorization::ExplicitAllowedIps(_) + | AddressAuthorization::ImplicitIpLiteral(_) => (DestinationDenialKind::AllowedIps, true), + AddressAuthorization::ExactDeclaredHost => (DestinationDenialKind::DeclaredEndpoint, true), + AddressAuthorization::DefaultPublicOnly => (DestinationDenialKind::InternalAddress, false), + AddressAuthorization::PinnedResolved(_) => (DestinationDenialKind::AllowedIps, false), + }; + + if control_plane_blocked && BLOCKED_CONTROL_PLANE_PORTS.contains(&port) { + return Err(DestinationDenial::new( + kind, + format!("port {port} is a blocked control-plane port, connection rejected"), + )); + } + + let mut allowed = Vec::new(); + let mut first_rejection = None; + for &ip in resolved_ips { + let rejection = match &plan.address_authorization { + AddressAuthorization::DefaultPublicOnly if is_internal_ip(ip) => Some(format!( + "{host} resolves to internal address {ip}, connection rejected" + )), + AddressAuthorization::ExplicitAllowedIps(networks) => { + if is_always_blocked_ip(ip) { + Some(format!( + "{host} resolves to always-blocked address {ip}, connection rejected" + )) + } else if !networks.iter().any(|network| network.contains(&ip)) { + Some(format!( + "{host} resolves to {ip} which is not in allowed_ips, connection rejected" + )) + } else { + None + } + } + AddressAuthorization::ImplicitIpLiteral(expected_ip) => { + if is_always_blocked_ip(ip) { + Some(format!( + "{host} resolves to always-blocked address {ip}, connection rejected" + )) + } else if ip != *expected_ip { + Some(format!( + "{host} resolves to {ip} which is not in allowed_ips, connection rejected" + )) + } else { + None + } + } + AddressAuthorization::ExactDeclaredHost if is_always_blocked_ip(ip) => Some(format!( + "{host} resolves to always-blocked address {ip}, connection rejected" + )), + AddressAuthorization::TrustedGatewayAlias { expected_ip } => { + if is_cloud_metadata_ip(ip) { + Some(format!( + "{host} resolves to cloud metadata address {ip}, connection rejected" + )) + } else if ip != *expected_ip { + Some(format!( + "{host} resolves to {ip} which does not match trusted host gateway \ + {expected_ip}, connection rejected" + )) + } else if !is_link_local_ip(ip) { + Some(format!( + "{host} resolves to non-link-local address {ip}, connection rejected" + )) + } else { + None + } + } + AddressAuthorization::PinnedResolved(pinned) if !pinned.contains(&ip) => Some(format!( + "{host} resolves to unpinned address {ip}, connection rejected" + )), + AddressAuthorization::DefaultPublicOnly + | AddressAuthorization::ExactDeclaredHost + | AddressAuthorization::PinnedResolved(_) => None, + }; + if let Some(reason) = rejection { + first_rejection.get_or_insert(reason); + } else if !allowed.contains(&ip) { + allowed.push(ip); + } + } + + if allowed.is_empty() { + return Err(DestinationDenial::new( + kind, + first_rejection.unwrap_or_else(|| { + format!( + "DNS resolution returned no addresses for {}", + super::normalize_host_lookup_key(host) + ) + }), + )); + } + + Ok(allowed) +} + /// Validated, but not yet opened, upstream destination. /// /// The explicit proxy adapter controls when `connect` is called so CONNECT and /// forward HTTP retain their current upstream-dial timing during the refactor. -pub(super) struct UpstreamConnector { +pub(crate) struct UpstreamConnector { host: String, port: u16, addrs: Vec, } impl UpstreamConnector { - pub(super) fn addrs(&self) -> &[SocketAddr] { + pub(crate) fn addrs(&self) -> &[SocketAddr] { &self.addrs } /// Opens the connection with `TCP_NODELAY` set: this is the upstream dial /// boundary for latency-sensitive proxied request/response traffic, where /// Nagle would stall sub-MSS writes on delayed ACKs. - pub(super) async fn connect(&self) -> std::io::Result { + pub(crate) async fn connect(&self) -> std::io::Result { tracing::debug!( host = %self.host, port = self.port, @@ -147,7 +268,7 @@ impl UpstreamConnector { connect_tcp_nodelay_best_effort(self.addrs.as_slice()).await } - fn new(host: &str, port: u16, addrs: Vec) -> Self { + pub(crate) fn new(host: &str, port: u16, addrs: Vec) -> Self { Self { host: host.to_string(), port, @@ -157,7 +278,7 @@ impl UpstreamConnector { } /// Resolve and validate a destination using the existing proxy security rules. -pub(super) async fn validate_destination( +pub(crate) async fn validate_destination( request: DestinationRequest<'_>, ) -> Result { let DestinationRequest { @@ -317,6 +438,82 @@ mod tests { assert!(denial.reason.contains("empty pinned address set")); } + #[test] + fn address_filter_retains_public_answer_from_mixed_set() { + let plan = DestinationValidationPlan { + address_authorization: AddressAuthorization::DefaultPublicOnly, + }; + let public: IpAddr = "8.8.8.8".parse().unwrap(); + let private: IpAddr = "10.1.2.3".parse().unwrap(); + + let allowed = + filter_resolved_addresses(&plan, "mixed.example", 443, &[private, public]).unwrap(); + + assert_eq!(allowed, vec![public]); + } + + #[test] + fn address_filter_exact_host_allows_private_but_not_always_blocked() { + let plan = DestinationValidationPlan { + address_authorization: AddressAuthorization::ExactDeclaredHost, + }; + let private: IpAddr = "10.1.2.3".parse().unwrap(); + let loopback: IpAddr = "127.0.0.1".parse().unwrap(); + + let allowed = + filter_resolved_addresses(&plan, "private.example", 443, &[loopback, private]).unwrap(); + + assert_eq!(allowed, vec![private]); + } + + #[test] + fn address_filter_enforces_allowed_ips() { + let plan = DestinationValidationPlan { + address_authorization: AddressAuthorization::ExplicitAllowedIps(vec![ + "10.2.0.0/16".parse().unwrap(), + ]), + }; + let included: IpAddr = "10.2.3.4".parse().unwrap(); + let excluded: IpAddr = "10.3.4.5".parse().unwrap(); + + let allowed = + filter_resolved_addresses(&plan, "allowlisted.example", 443, &[excluded, included]) + .unwrap(); + + assert_eq!(allowed, vec![included]); + } + + #[test] + fn address_filter_rejects_always_blocked_only_answer() { + let plan = DestinationValidationPlan { + address_authorization: AddressAuthorization::ExactDeclaredHost, + }; + + let denial = filter_resolved_addresses( + &plan, + "loopback.example", + 443, + &["127.0.0.1".parse().unwrap()], + ) + .expect_err("loopback must not survive filtering"); + + assert_eq!(denial.kind, DestinationDenialKind::DeclaredEndpoint); + } + + #[test] + fn address_filter_rejects_control_plane_port() { + let plan = DestinationValidationPlan { + address_authorization: AddressAuthorization::ExactDeclaredHost, + }; + + let denial = + filter_resolved_addresses(&plan, "api.example", 6443, &["8.8.8.8".parse().unwrap()]) + .expect_err("control-plane port must remain blocked"); + + assert_eq!(denial.kind, DestinationDenialKind::DeclaredEndpoint); + assert!(denial.reason.contains("blocked control-plane port")); + } + #[test] fn validation_mode_precedence_is_explicit_and_stable() { let trusted_ip = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2));