From 44e6c01f8e15ddf1cbc50020e63cfc6104ad6b6d Mon Sep 17 00:00:00 2001 From: craig Date: Tue, 6 Oct 2026 17:17:42 -0700 Subject: [PATCH] feat(policy-advisor): support reviewed endpoint CA trust Signed-off-by: craig --- Cargo.lock | 3 + architecture/security-policy.md | 10 + crates/openshell-core/src/gateway_fleet.rs | 175 ++++++++ crates/openshell-policy/Cargo.toml | 5 + crates/openshell-policy/src/ambiguity.rs | 17 + crates/openshell-policy/src/lib.rs | 160 +++++++ crates/openshell-policy/src/merge.rs | 122 +++++- crates/openshell-providers/src/profiles.rs | 1 + crates/openshell-sandbox/src/lib.rs | 98 +++-- .../src/mechanistic_mapper.rs | 70 +++- crates/openshell-server/src/grpc/policy.rs | 125 ++++++ .../tests/prover_policy_round_trip.rs | 147 +++++++ .../data/sandbox-policy.rego | 11 +- .../src/l7/tls.rs | 394 ++++++++++++++++++ .../openshell-supervisor-network/src/opa.rs | 94 +++++ .../src/policy_local.rs | 290 ++++++++++++- .../openshell-supervisor-network/src/proxy.rs | 238 +++++++---- .../openshell-supervisor-network/src/run.rs | 5 +- .../src/skills/policy-advisor/SKILL.md | 2 + .../src/skills/policy_advisor.md | 6 + docs/reference/policy-schema.mdx | 1 + docs/sandboxes/policy-advisor.mdx | 4 +- proto/sandbox.proto | 3 + .../v1/internal/converter/coverage_test.go | 1 + .../v1/internal/converter/network_policy.go | 2 + .../internal/converter/network_policy_test.go | 4 + sdk/go/openshell/v1/types/network_policy.go | 12 +- sdk/go/proto/sandboxv1/sandbox.pb.go | 22 +- 28 files changed, 1872 insertions(+), 150 deletions(-) create mode 100644 crates/openshell-server/tests/prover_policy_round_trip.rs diff --git a/Cargo.lock b/Cargo.lock index 59b92b4185..fdd150c96a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4358,6 +4358,9 @@ dependencies = [ "noyalib", "openshell-core", "prost-types", + "rcgen", + "rustls", + "rustls-pemfile", "serde", "serde_json", ] diff --git a/architecture/security-policy.md b/architecture/security-policy.md index 62f5837e70..a5245bccc5 100644 --- a/architecture/security-policy.md +++ b/architecture/security-policy.md @@ -142,6 +142,16 @@ flag defaults to `false` and is security-flagged in policy approval flows. Incremental merges only ever add the flag to a matching endpoint; clearing it requires removing the endpoint or replacing the policy. +TLS endpoints may specify `upstream_ca_pem` in their policy. The supervisor +adds that PEM certificate bundle only to the selected endpoint's upstream +rustls verifier, alongside its normal roots. CONNECT still terminates client +TLS and uses the parsed HTTP relay, so credential rewrite, middleware, and +redaction retain their normal boundary. Policy validation rejects this field +on `tls: skip` and `protocol: tcp` endpoints, which do not use the verifier. +Policy advisor may propose the same endpoint field. Its security-notes gate +requires human review even when the prover reports an empty delta; approval +then uses the normal policy merge and hot reload path. + The network supervisor independently enforces the same boundary. Credentialed WebSocket upgrades use the parsed relay, binary frames fail closed, and text placeholders require rewrite. REST bodies can continue streaming when body diff --git a/crates/openshell-core/src/gateway_fleet.rs b/crates/openshell-core/src/gateway_fleet.rs index c069954740..1def547605 100644 --- a/crates/openshell-core/src/gateway_fleet.rs +++ b/crates/openshell-core/src/gateway_fleet.rs @@ -259,6 +259,83 @@ fn member_score(member: &str, sandbox_id: &str) -> u64 { u64::from_be_bytes(score) } +/// Split a gateway endpoint URL into the scheme and port to reuse when +/// addressing individual replicas. +/// +/// A replica is dialed at the same scheme and port as the configured +/// endpoint, differing only in host — the fleet is uniform by construction, +/// being one Deployment behind one Service. +#[must_use] +pub fn endpoint_shape(endpoint: &str) -> Option<(String, u16)> { + let (scheme, rest) = endpoint.split_once("://")?; + let host_and_port = rest.split('/').next().unwrap_or(rest); + let port = host_and_port.rsplit_once(':')?.1.parse::().ok()?; + Some((scheme.to_string(), port)) +} + +/// Prefix a resolved member address (`host:port`) with the configured scheme. +#[must_use] +pub fn member_endpoint(scheme: &str, member: &str) -> String { + format!("{scheme}://{member}") +} + +/// A gateway endpoint for a unary RPC, addressed at a fleet replica. +/// +/// The configured endpoint is not a fallback when a fleet is configured: it +/// is then the headless `Service`, which is deliberately absent from the +/// gateway's serving certificate, so dialling it fails hostname +/// verification. This resolves membership itself and addresses a subset +/// member directly; `attempt` rotates through the subset so a replica that +/// has since died costs one attempt rather than every attempt. Any replica +/// can serve a unary store read, so a caller whose hash key differs from +/// another caller's still reaches a working replica. +/// +/// Single-endpoint mode — no fleet configured, or an endpoint whose scheme +/// and port cannot be reused — is the one case where the configured endpoint +/// *is* a single gateway, and then it is returned unchanged. +/// +/// `None` means membership could not be resolved. Callers are retry loops, so +/// that costs one backoff. +pub async fn unary_endpoint( + configured_endpoint: &str, + sandbox_id: &str, + attempt: usize, +) -> Option { + let Some((scheme, port)) = endpoint_shape(configured_endpoint) else { + return Some(configured_endpoint.to_string()); + }; + let Some(fleet) = FleetConfig::from_env(port) else { + return Some(configured_endpoint.to_string()); + }; + + let members = match resolve_members(&fleet.dns_name, port).await { + Ok(members) if !members.is_empty() => members, + Ok(_) => { + tracing::warn!( + fleet_dns_name = %fleet.dns_name, + "gateway fleet resolved to no endpoints; cannot address a replica yet" + ); + return None; + } + Err(error) => { + tracing::warn!( + fleet_dns_name = %fleet.dns_name, + error = %error, + "could not resolve the gateway fleet; cannot address a replica yet" + ); + return None; + } + }; + + // The subset this sandbox's sessions would use, so a unary RPC warms the + // same replicas rather than a third set. Any replica can serve a store + // read, so falling back to the full membership is harmless. + let subset = subset_for(sandbox_id, &members, fleet.subset_size); + let pool = if subset.is_empty() { members } else { subset }; + + Some(member_endpoint(&scheme, &pool[attempt % pool.len()])) +} + #[cfg(test)] mod tests { use super::*; @@ -526,3 +603,101 @@ mod tests { } } } + +#[cfg(test)] +mod unary_endpoint_tests { + use super::*; + use std::sync::Mutex; + + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + fn fleet_env_lock() -> std::sync::MutexGuard<'static, ()> { + ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + struct EnvVarGuard { + key: &'static str, + previous: Option, + } + + impl EnvVarGuard { + #[allow(unsafe_code)] // Tests serialize process-wide environment changes with ENV_LOCK. + fn set(key: &'static str, value: &str) -> Self { + let previous = std::env::var_os(key); + // SAFETY: test-process scoped and serialized by ENV_LOCK. + unsafe { + std::env::set_var(key, value); + } + Self { key, previous } + } + } + + #[allow(unsafe_code)] // Tests serialize process-wide environment changes with ENV_LOCK. + impl Drop for EnvVarGuard { + fn drop(&mut self) { + unsafe { + match &self.previous { + Some(value) => std::env::set_var(self.key, value), + None => std::env::remove_var(self.key), + } + } + } + } + + /// Single-endpoint mode — no fleet configured — returns the configured + /// endpoint unchanged, so callers without a fleet keep working. + // + // The ENV_LOCK guard is held across the await on purpose: `FleetConfig` is + // read from the process environment inside `unary_endpoint`, so the + // mutation this test makes must stay serialized with every other + // env-reading test for the duration of the call. + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn unary_endpoint_without_fleet_returns_configured_endpoint() { + let _guard = fleet_env_lock(); + let _fleet = EnvVarGuard::set(FLEET_DNS_NAME_ENV, ""); + + let got = unary_endpoint("https://openshell.sandbox.svc:8080", "sbx", 0) + .await + .expect("single-endpoint mode always resolves"); + assert_eq!(got, "https://openshell.sandbox.svc:8080"); + } + + /// An endpoint whose scheme/port cannot be reused cannot be re-addressed + /// at a replica, so it is returned unchanged rather than dropped. The + /// fleet being configured is what makes this the pass-through path that + /// matters: with no fleet the same return happens for a different reason. + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn unary_endpoint_passes_through_an_endpoint_it_cannot_rebuild() { + let _guard = fleet_env_lock(); + let _fleet = EnvVarGuard::set( + FLEET_DNS_NAME_ENV, + "openshell-headless.sandbox.svc.cluster.local", + ); + let _size = EnvVarGuard::set(FLEET_SUBSET_SIZE_ENV, "2"); + + let got = unary_endpoint("openshell", "sbx", 0) + .await + .expect("unrebuildable endpoints are passed through, not dropped"); + assert_eq!(got, "openshell"); + } + + /// `endpoint_shape`/`member_endpoint` round-trip: a rebuilt replica + /// address keeps the configured scheme and port and differs only in + /// host. + #[test] + fn endpoint_shape_round_trips_with_member_endpoint() { + let (scheme, port) = endpoint_shape("https://openshell.sandbox.svc:8080").unwrap(); + assert_eq!(scheme, "https"); + assert_eq!(port, 8080); + assert_eq!( + member_endpoint(&scheme, "10-42-0-7.openshell-headless.ns.svc:8080"), + "https://10-42-0-7.openshell-headless.ns.svc:8080" + ); + assert_eq!(endpoint_shape("https://openshell"), None); + assert_eq!(endpoint_shape("openshell:8080"), None); + } +} diff --git a/crates/openshell-policy/Cargo.toml b/crates/openshell-policy/Cargo.toml index cb32584186..6589c4462f 100644 --- a/crates/openshell-policy/Cargo.toml +++ b/crates/openshell-policy/Cargo.toml @@ -18,6 +18,11 @@ serde_json = { workspace = true } serde_yml = { workspace = true } miette = { workspace = true } prost-types = { workspace = true } +rustls = { workspace = true } +rustls-pemfile = { workspace = true } [lints] workspace = true + +[dev-dependencies] +rcgen = { workspace = true } diff --git a/crates/openshell-policy/src/ambiguity.rs b/crates/openshell-policy/src/ambiguity.rs index 2c0b1b1944..0a3fa5c522 100644 --- a/crates/openshell-policy/src/ambiguity.rs +++ b/crates/openshell-policy/src/ambiguity.rs @@ -181,6 +181,9 @@ fn connection_conflicts(left: &NetworkEndpoint, right: &NetworkEndpoint) -> Vec< &normalized_strings(&left.allowed_ips), &normalized_strings(&right.allowed_ips), ); + if left.upstream_ca_pem != right.upstream_ca_pem { + conflicts.push("upstream_ca_pem differs".to_string()); + } conflicts } @@ -1066,4 +1069,18 @@ mod tests { .any(|conflict| conflict.contains("transparent_tcp_eligible")) ); } + + #[test] + fn different_upstream_ca_pem_is_ambiguous_without_disclosing_certificates() { + let mut left = endpoint("api.example.com", 443); + left.upstream_ca_pem = "private-left-pem".to_string(); + let mut right = endpoint("api.example.com", 443); + right.upstream_ca_pem = "private-right-pem".to_string(); + let ambiguities = find_endpoint_ambiguities(&policy_with(left, right)); + assert_eq!(ambiguities.len(), 1); + let diagnostic = ambiguities[0].to_string(); + assert!(diagnostic.contains("upstream_ca_pem differs")); + assert!(!diagnostic.contains("private-left-pem")); + assert!(!diagnostic.contains("private-right-pem")); + } } diff --git a/crates/openshell-policy/src/lib.rs b/crates/openshell-policy/src/lib.rs index b2c15e4595..fc10b72e6c 100644 --- a/crates/openshell-policy/src/lib.rs +++ b/crates/openshell-policy/src/lib.rs @@ -129,6 +129,8 @@ struct NetworkEndpointDef { protocol: String, #[serde(default, skip_serializing_if = "String::is_empty")] tls: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + upstream_ca_pem: Option, #[serde(default, skip_serializing_if = "String::is_empty")] enforcement: String, #[serde(default, skip_serializing_if = "String::is_empty")] @@ -845,6 +847,7 @@ fn to_proto(raw: PolicyFile) -> Result { ports: normalized_ports, protocol: protocol.clone(), tls: e.tls, + upstream_ca_pem: e.upstream_ca_pem.unwrap_or_default(), enforcement: e.enforcement, access: e.access, rules: allow_rules @@ -1013,6 +1016,8 @@ fn from_proto(policy: &SandboxPolicy) -> PolicyFile { ports, protocol, tls: e.tls.clone(), + upstream_ca_pem: (!e.upstream_ca_pem.is_empty()) + .then(|| e.upstream_ca_pem.clone()), enforcement: e.enforcement.clone(), access: e.access.clone(), rules, @@ -1164,6 +1169,17 @@ pub fn parse_sandbox_policy(yaml: &str) -> Result { let raw: PolicyFile = serde_yml::from_str(yaml) .into_diagnostic() .wrap_err("failed to parse sandbox policy YAML")?; + for rule in raw.network_policies.values() { + for endpoint in &rule.endpoints { + if endpoint + .upstream_ca_pem + .as_deref() + .is_some_and(str::is_empty) + { + return Err(miette::miette!("upstream_ca_pem cannot be empty")); + } + } + } validate_mcp_version_schema(&raw)?; to_proto(raw) } @@ -1354,6 +1370,13 @@ pub enum PolicyViolation { endpoint_index: usize, reason: String, }, + /// Overlapping endpoints disagree on their supplemental upstream CA. + ConflictingUpstreamCa { + left_policy: String, + left_endpoint_index: usize, + right_policy: String, + right_endpoint_index: usize, + }, /// A middleware configuration is structurally invalid. InvalidMiddlewareConfig { name: String, reason: String }, /// Too many middleware configurations are attached to one policy. @@ -1516,6 +1539,15 @@ impl fmt::Display for PolicyViolation { f, "network policy '{policy_name}': endpoint {endpoint_index} has invalid L7 configuration: {reason}" ), + Self::ConflictingUpstreamCa { + left_policy, + left_endpoint_index, + right_policy, + right_endpoint_index, + } => write!( + f, + "network policies '{left_policy}' endpoint[{left_endpoint_index}] and '{right_policy}' endpoint[{right_endpoint_index}] overlap with differing upstream_ca_pem" + ), Self::InvalidMiddlewareConfig { name, reason } => { write!(f, "middleware config '{name}' is invalid: {reason}") } @@ -1817,7 +1849,22 @@ fn validate_sandbox_policy_with_mcp_presence( .unwrap_or(false), }; let mut l7_errors = validate_l7_endpoint_semantics(&fields); + if !ep.upstream_ca_pem.is_empty() { + if ep.tls.eq_ignore_ascii_case("skip") { + l7_errors.push("upstream_ca_pem cannot be used with tls: skip".to_string()); + } + let usable = validate_upstream_ca_pem(&ep.upstream_ca_pem); + if !usable { + l7_errors.push( + "upstream_ca_pem must contain at least one usable PEM certificate" + .to_string(), + ); + } + } let mut explicit_tcp_fields = Vec::new(); + if !ep.upstream_ca_pem.is_empty() { + explicit_tcp_fields.push("upstream_ca_pem"); + } if !ep.enforcement.is_empty() { explicit_tcp_fields.push("enforcement"); } @@ -1913,6 +1960,24 @@ fn validate_sandbox_policy_with_mcp_presence( } } + // A request must never select among different trust roots for the same + // destination. Reuse the runtime ambiguity detector so wildcard and + // multi-port overlaps follow the same matching semantics as the proxy. + for ambiguity in find_endpoint_ambiguities(policy) { + if ambiguity + .conflicts + .iter() + .any(|conflict| conflict == "upstream_ca_pem differs") + { + violations.push(PolicyViolation::ConflictingUpstreamCa { + left_policy: ambiguity.left_policy, + left_endpoint_index: ambiguity.left_endpoint_index, + right_policy: ambiguity.right_policy, + right_endpoint_index: ambiguity.right_endpoint_index, + }); + } + } + violations.extend(middleware::validate(policy)); if violations.is_empty() { @@ -1922,6 +1987,101 @@ fn validate_sandbox_policy_with_mcp_presence( } } +/// Accept only usable certificate PEM blocks; private keys and unrelated PEM +/// material must never be carried in an endpoint policy. +pub fn validate_upstream_ca_pem(pem: &str) -> bool { + let mut reader = std::io::BufReader::new(pem.as_bytes()); + let mut certs = Vec::new(); + loop { + match rustls_pemfile::read_one(&mut reader) { + Ok(Some(rustls_pemfile::Item::X509Certificate(cert))) => certs.push(cert), + Ok(None) => break, + _ => return false, + } + } + if certs.is_empty() { + return false; + } + let mut store = rustls::RootCertStore::empty(); + let expected = certs.len(); + let (added, ignored) = store.add_parsable_certificates(certs); + added == expected && ignored == 0 +} + +#[cfg(test)] +mod upstream_ca_tests { + use super::*; + + #[test] + fn supplemental_ca_round_trips_and_rejects_unsafe_combinations() { + let cert = rcgen::generate_simple_self_signed(vec!["api.example.test".into()]).unwrap(); + let pem = cert.cert.pem(); + let indented_pem = pem + .lines() + .map(|line| format!(" {line}")) + .collect::>() + .join("\n"); + let yaml = format!( + "version: 1\nnetwork_policies:\n api:\n endpoints:\n - host: api.example.test\n port: 443\n upstream_ca_pem: |\n{indented_pem}\n" + ); + let policy = parse_sandbox_policy(&yaml).unwrap(); + assert_eq!( + policy.network_policies["api"].endpoints[0] + .upstream_ca_pem + .trim(), + pem.trim() + ); + assert_eq!( + parse_sandbox_policy(&serialize_sandbox_policy(&policy).unwrap()).unwrap(), + policy + ); + let mut skip = policy.clone(); + skip.network_policies.get_mut("api").unwrap().endpoints[0].tls = "skip".into(); + assert!(validate_sandbox_policy(&skip).unwrap_err().iter().any(|v| { + v.to_string() + .contains("upstream_ca_pem cannot be used with tls: skip") + })); + let mut tcp = policy; + tcp.network_policies.get_mut("api").unwrap().endpoints[0].protocol = "tcp".into(); + assert!( + validate_sandbox_policy(&tcp) + .unwrap_err() + .iter() + .any(|v| v.to_string().contains("upstream_ca_pem")) + ); + } + + #[test] + fn supplemental_ca_rejects_empty_invalid_and_private_key_material() { + assert!(!validate_upstream_ca_pem("")); + assert!(!validate_upstream_ca_pem("not a certificate")); + assert!(!validate_upstream_ca_pem( + "-----BEGIN PRIVATE KEY-----\nAA==\n-----END PRIVATE KEY-----" + )); + let yaml = "version: 1\nnetwork_policies:\n api:\n endpoints:\n - host: api.example.test\n port: 443\n upstream_ca_pem: ''\n"; + assert!( + parse_sandbox_policy(yaml) + .unwrap_err() + .to_string() + .contains("upstream_ca_pem cannot be empty") + ); + let mut policy = parse_sandbox_policy("version: 1\nnetwork_policies:\n api:\n endpoints:\n - host: api.example.test\n port: 443\n").unwrap(); + for invalid in [ + "not a certificate", + "-----BEGIN PRIVATE KEY-----\nAA==\n-----END PRIVATE KEY-----", + ] { + policy.network_policies.get_mut("api").unwrap().endpoints[0].upstream_ca_pem = + invalid.into(); + assert!( + validate_sandbox_policy(&policy) + .unwrap_err() + .iter() + .any(|v| v.to_string().contains("upstream_ca_pem must contain")) + ); + } + } +} + fn host_wildcard_shape_invalid(host: &str) -> bool { if host == "*" || host == "**" { return true; diff --git a/crates/openshell-policy/src/merge.rs b/crates/openshell-policy/src/merge.rs index e6d72d4899..c59507f53a 100644 --- a/crates/openshell-policy/src/merge.rs +++ b/crates/openshell-policy/src/merge.rs @@ -302,6 +302,12 @@ pub enum PolicyMergeError { /// Rendered effective contract the operation asked for. incoming: String, }, + /// An overlapping endpoint already trusts a different supplemental CA. + UpstreamCaConflict { + operation_index: usize, + host: String, + port: u32, + }, /// Newly added binary scope would inherit an existing endpoint /// authorization that the incoming rule did not declare. NewBinaryWouldInheritAuthorization { @@ -419,6 +425,14 @@ impl std::fmt::Display for PolicyMergeError { f, "merge operation {operation_index} cannot combine MCP contracts at {host}:{port}: existing {existing}, incoming {incoming}" ), + Self::UpstreamCaConflict { + operation_index, + host, + port, + } => write!( + f, + "merge operation {operation_index} cannot change upstream_ca_pem for {host}:{port}; replace the endpoint policy" + ), Self::NewBinaryWouldInheritAuthorization { operation_index, rule_name, @@ -675,6 +689,9 @@ fn endpoint_attributes_cover(loaded: &NetworkEndpoint, proposed: &NetworkEndpoin if !proposed.tls.is_empty() && effective_tls(&loaded.tls) != effective_tls(&proposed.tls) { return false; } + if !proposed.upstream_ca_pem.is_empty() && loaded.upstream_ca_pem != proposed.upstream_ca_pem { + return false; + } if !proposed.enforcement.is_empty() && effective_enforcement(&loaded.enforcement) != effective_enforcement(&proposed.enforcement) @@ -1293,6 +1310,7 @@ fn is_authorization_inheritance_conflict(error: &PolicyMergeError) -> bool { // inspection contract per host and port rather than per rule, so two // contracts stay ambiguous however they are split. PolicyMergeError::McpContractConflict { .. } + | PolicyMergeError::UpstreamCaConflict { .. } | PolicyMergeError::ConflictingInspectionContracts { .. } => false, // Reports an unsupported or missing state in the policy the fold @@ -1461,6 +1479,19 @@ fn merge_endpoint( }, warnings, ); + if existing.upstream_ca_pem.is_empty() { + existing + .upstream_ca_pem + .clone_from(&incoming.upstream_ca_pem); + } else if !incoming.upstream_ca_pem.is_empty() + && existing.upstream_ca_pem != incoming.upstream_ca_pem + { + return Err(PolicyMergeError::UpstreamCaConflict { + operation_index, + host, + port, + }); + } if !incoming.rules.is_empty() { expand_existing_access(existing, &host, port, warnings)?; @@ -1728,6 +1759,9 @@ fn adopt_unset_retained_fields( if adopted.tls.is_empty() { adopted.tls.clone_from(&merged.tls); } + if adopted.upstream_ca_pem.is_empty() { + adopted.upstream_ca_pem.clone_from(&merged.upstream_ca_pem); + } if adopted.enforcement.is_empty() { adopted.enforcement.clone_from(&merged.enforcement); } @@ -2216,8 +2250,8 @@ mod tests { use super::{ ANY_BINARY_SCOPE, DEFAULT_JSON_RPC_MAX_BODY_BYTES, PolicyMergeError, PolicyMergeOp, - PolicyMergeWarning, canonical_ports, canonicalize_advisor_add_rule, generated_rule_name, - merge_policy, policy_covers_rule, + PolicyMergeWarning, adopt_unset_retained_fields, canonical_ports, + canonicalize_advisor_add_rule, generated_rule_name, merge_policy, policy_covers_rule, }; use crate::{restrictive_default_policy, validate_sandbox_policy}; use openshell_core::{ @@ -3699,6 +3733,90 @@ mod tests { assert_eq!(rule.binaries.len(), 2); } + #[test] + fn add_rule_preserves_requested_upstream_ca_and_coverage() { + let pem = rcgen::generate_simple_self_signed(vec!["api.example.com".into()]) + .unwrap() + .cert + .pem(); + let existing = rule_with_authorizations( + "api", + vec![endpoint("api.example.com", 443)], + &["/usr/bin/client"], + ); + let mut incoming_endpoint = endpoint("api.example.com", 443); + incoming_endpoint.upstream_ca_pem = pem.clone(); + let incoming = + rule_with_authorizations("api", vec![incoming_endpoint], &["/usr/bin/client"]); + let merged = merge_policy( + policy_with_rule("api", existing), + &[PolicyMergeOp::AddRule { + rule_name: "api".into(), + rule: incoming.clone(), + }], + ) + .unwrap() + .policy; + assert_eq!( + merged.network_policies["api"].endpoints[0].upstream_ca_pem, + pem + ); + assert!(policy_covers_rule(&merged, &incoming)); + let mut without_ca = merged; + without_ca + .network_policies + .get_mut("api") + .unwrap() + .endpoints[0] + .upstream_ca_pem + .clear(); + assert!(!policy_covers_rule(&without_ca, &incoming)); + } + + #[test] + fn add_rule_rejects_conflicting_upstream_ca_without_exposing_pem() { + let first = rcgen::generate_simple_self_signed(vec!["api.example.com".into()]) + .unwrap() + .cert + .pem(); + let second = rcgen::generate_simple_self_signed(vec!["api.example.com".into()]) + .unwrap() + .cert + .pem(); + let mut existing_endpoint = endpoint("api.example.com", 443); + existing_endpoint.upstream_ca_pem = first.clone(); + let mut incoming_endpoint = endpoint("api.example.com", 443); + incoming_endpoint.upstream_ca_pem = second.clone(); + let existing = + rule_with_authorizations("api", vec![existing_endpoint], &["/usr/bin/client"]); + let incoming = + rule_with_authorizations("api", vec![incoming_endpoint], &["/usr/bin/client"]); + let error = merge_policy( + policy_with_rule("api", existing), + &[PolicyMergeOp::AddRule { + rule_name: "api".into(), + rule: incoming, + }], + ) + .unwrap_err(); + assert!(matches!(error, PolicyMergeError::UpstreamCaConflict { .. })); + let message = error.to_string(); + assert!(message.contains("upstream_ca_pem")); + assert!(!message.contains(&first)); + assert!(!message.contains(&second)); + } + + #[test] + fn inheritance_adopts_unset_upstream_ca() { + let declared = endpoint("api.example.com", 443); + let mut merged = declared.clone(); + merged.upstream_ca_pem = "retained-ca".into(); + assert_eq!( + adopt_unset_retained_fields(&declared, &merged).upstream_ca_pem, + "retained-ca" + ); + } + #[test] fn add_rule_user_binary_clears_advisor_marker_for_same_path() { let mut policy = restrictive_default_policy(); diff --git a/crates/openshell-providers/src/profiles.rs b/crates/openshell-providers/src/profiles.rs index 9cdf2257ad..c51015f0ad 100644 --- a/crates/openshell-providers/src/profiles.rs +++ b/crates/openshell-providers/src/profiles.rs @@ -1431,6 +1431,7 @@ fn endpoint_to_proto(endpoint: &EndpointProfile) -> NetworkEndpoint { port: endpoint.port, protocol: endpoint.protocol.clone(), tls: endpoint.tls.clone(), + upstream_ca_pem: String::new(), enforcement: endpoint.enforcement.clone(), access: endpoint.access.clone(), rules: endpoint diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 3421d6cde2..396a5d59a6 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -674,17 +674,23 @@ pub async fn run_sandbox( let aggregator = denial_aggregator::DenialAggregator::new(rx, flush_interval_secs); let denial_workspace_gate = workspace_rx.clone(); let denial_workspace_rx = workspace_rx.clone(); + // Fleet subset routing hashes the sandbox UUID, matching the + // supervisor sessions; the name fallback covers sandboxes that only + // have a name. + let agg_id = sandbox_id.clone().unwrap_or_else(|| agg_name.clone()); tokio::spawn(async move { aggregator .run( |summaries| { let endpoint = agg_endpoint.clone(); + let sandbox_id = agg_id.clone(); let sandbox_name = agg_name.clone(); let workspace = denial_workspace_rx.borrow().clone(); async move { if let Err(e) = flush_proposals_to_gateway( &endpoint, + &sandbox_id, &sandbox_name, &workspace, summaries, @@ -719,17 +725,21 @@ pub async fn run_sandbox( let aggregator = activity_aggregator::ActivityAggregator::new(rx, flush_interval_secs); let activity_workspace_gate = workspace_rx.clone(); let activity_workspace_rx = workspace_rx.clone(); + // Same fleet routing identity as the denial flush. + let agg_id = sandbox_id.clone().unwrap_or_else(|| agg_name.clone()); tokio::spawn(async move { aggregator .run( move |summary| { let endpoint = agg_endpoint.clone(); + let sandbox_id = agg_id.clone(); let sandbox_name = agg_name.clone(); let workspace = activity_workspace_rx.borrow().clone(); async move { if let Err(e) = flush_activity_to_gateway( &endpoint, + &sandbox_id, &sandbox_name, &workspace, summary, @@ -1506,6 +1516,7 @@ fn process_policy_for_topology( /// Flush aggregated denial summaries to the gateway via `SubmitPolicyAnalysis`. async fn flush_proposals_to_gateway( endpoint: &str, + sandbox_id: &str, sandbox_name: &str, workspace: &str, summaries: Vec, @@ -1513,9 +1524,6 @@ async fn flush_proposals_to_gateway( use openshell_core::grpc_client::CachedOpenShellClient; use openshell_core::proto::{DenialSummary, L7RequestSample}; - let client = CachedOpenShellClient::connect(endpoint).await?; - client.set_workspace(workspace.to_string()); - let proto_summaries: Vec = summaries .into_iter() .map(|s| DenialSummary { @@ -1560,15 +1568,32 @@ async fn flush_proposals_to_gateway( "Flushed denial analysis to gateway" ); - client - .submit_policy_analysis( - sandbox_name, - proto_summaries, - proposals, - Vec::new(), - "mechanistic", - ) - .await?; + // The configured endpoint is the headless Service — a discovery address + // the gateway's serving certificate deliberately omits, so dialling it + // fails the TLS handshake. Address a fleet replica instead, like every + // other unary RPC this sandbox makes. + grpc_retry_on_fleet("Denial analysis flush", endpoint, sandbox_id, |target| { + // Cloned per attempt: the retry calls this more than once. + let workspace = workspace.to_string(); + let sandbox_name = sandbox_name.to_string(); + let proto_summaries = proto_summaries.clone(); + let proposals = proposals.clone(); + async move { + let client = CachedOpenShellClient::connect(&target).await?; + client.set_workspace(workspace); + client + .submit_policy_analysis( + &sandbox_name, + proto_summaries, + proposals, + Vec::new(), + "mechanistic", + ) + .await?; + Ok(()) + } + }) + .await?; Ok(()) } @@ -1576,6 +1601,7 @@ async fn flush_proposals_to_gateway( /// Flush an anonymous activity summary to the gateway via `SubmitPolicyAnalysis`. async fn flush_activity_to_gateway( endpoint: &str, + sandbox_id: &str, sandbox_name: &str, workspace: &str, summary: activity_aggregator::FlushableActivitySummary, @@ -1583,9 +1609,6 @@ async fn flush_activity_to_gateway( use openshell_core::grpc_client::CachedOpenShellClient; use openshell_core::proto::{DenialGroupCount, NetworkActivitySummary}; - let client = CachedOpenShellClient::connect(endpoint).await?; - client.set_workspace(workspace.to_string()); - let proto_summary = NetworkActivitySummary { network_activity_count: summary.network_activity_count, denied_action_count: summary.denied_action_count, @@ -1606,15 +1629,29 @@ async fn flush_activity_to_gateway( "Flushed activity summary to gateway" ); - client - .submit_policy_analysis( - sandbox_name, - Vec::new(), - Vec::new(), - vec![proto_summary], - "activity", - ) - .await?; + // Same fleet contract as the denial flush: the configured endpoint is a + // discovery address, never dialled. + grpc_retry_on_fleet("Activity summary flush", endpoint, sandbox_id, |target| { + // Cloned per attempt: the retry calls this more than once. + let workspace = workspace.to_string(); + let sandbox_name = sandbox_name.to_string(); + let proto_summary = proto_summary.clone(); + async move { + let client = CachedOpenShellClient::connect(&target).await?; + client.set_workspace(workspace); + client + .submit_policy_analysis( + &sandbox_name, + Vec::new(), + Vec::new(), + vec![proto_summary], + "activity", + ) + .await?; + Ok(()) + } + }) + .await?; Ok(()) } @@ -3484,7 +3521,18 @@ async fn report_initial_policy_failure( }) else { return; }; - let client = match openshell_core::grpc_client::CachedOpenShellClient::connect(endpoint).await { + // Fleet-addressed like every other unary RPC: the configured endpoint is + // a discovery address the serving certificate deliberately omits. + let client = match grpc_retry_on_fleet( + "Initial policy failure connect", + endpoint, + sandbox_id, + |target| async move { + openshell_core::grpc_client::CachedOpenShellClient::connect(&target).await + }, + ) + .await + { Ok(client) => client, Err(e) => { warn!(error = %e, "Failed to connect to report initial policy failure"); diff --git a/crates/openshell-sandbox/src/mechanistic_mapper.rs b/crates/openshell-sandbox/src/mechanistic_mapper.rs index 9be5f8e438..77ceded8a5 100644 --- a/crates/openshell-sandbox/src/mechanistic_mapper.rs +++ b/crates/openshell-sandbox/src/mechanistic_mapper.rs @@ -142,7 +142,7 @@ pub fn generate_proposals(summaries: &[DenialSummary]) -> Vec { } }; - let binaries: Vec = if binary.is_empty() { + let binaries: Vec = if is_unknown_binary(binary) { vec![] } else { let mut proposal_binary = NetworkBinary { @@ -169,7 +169,7 @@ pub fn generate_proposals(summaries: &[DenialSummary]) -> Vec { let confidence = compute_confidence(total_count, *port as u16, is_ssrf); // Generate rationale. - let binary_list = if binary.is_empty() { + let binary_list = if is_unknown_binary(binary) { "unknown binary".to_string() } else { short_binary_name(binary) @@ -425,6 +425,18 @@ fn short_binary_name(path: &str) -> String { path.rsplit('/').next().unwrap_or(path).to_string() } +/// The network supervisor emits `"-"` as the binary placeholder when the +/// connecting process's executable could not be resolved (the same +/// convention as the OCSF activity feed). A placeholder must never be copied +/// into a proposed rule's `binaries` list: a literal `-` binary path renders +/// the serialized policy YAML ambiguous (`path: -` parses as a block-sequence +/// indicator), breaking every consumer that round-trips the policy through +/// YAML — the gateway prover validation, which rejects the proposal as +/// "validation unavailable", and the sandbox's `policy.local` reload. +fn is_unknown_binary(binary: &str) -> bool { + binary.is_empty() || binary == "-" +} + /// Check if a destination host is always-blocked. /// /// For literal IP hosts, checks against [`is_always_blocked_ip`]. @@ -548,6 +560,60 @@ mod tests { assert!(rule.endpoints[0].allowed_ips.is_empty()); } + #[test] + fn test_generate_proposals_placeholder_binary_is_unknown() { + // The supervisor emits "-" when the connecting process's executable + // could not be resolved. The mapper must treat it as "unknown" — + // copying the placeholder into the rule breaks the gateway prover, + // whose YAML round-trip cannot re-parse `path: -` (block-sequence + // indicator), and poisons the sandbox policy on approval. + let summaries = vec![DenialSummary { + sandbox_id: "test".to_string(), + host: "example.com".to_string(), + port: 443, + binary: "-".to_string(), + ancestors: vec![], + deny_reason: "no matching policy".to_string(), + first_seen_ms: 1000, + last_seen_ms: 2000, + count: 3, + suppressed_count: 0, + total_count: 3, + sample_cmdlines: vec![], + binary_sha256: String::new(), + persistent: false, + denial_stage: "connect".to_string(), + l7_request_samples: vec![], + l7_inspection_active: false, + }]; + + let proposals = generate_proposals(&summaries); + assert_eq!(proposals.len(), 1); + let rule = proposals[0].proposed_rule.as_ref().unwrap(); + assert!( + rule.binaries.is_empty(), + "placeholder binary must not enter the rule" + ); + assert!(proposals[0].rationale.contains("unknown binary")); + + // The proposed rule must survive the same serialize → parse + // round-trip the gateway prover runs on the merged policy. + let merged = openshell_policy::merge_policy( + openshell_core::proto::SandboxPolicy { + version: 1, + ..Default::default() + }, + &[openshell_policy::PolicyMergeOp::AddRule { + rule_name: proposals[0].rule_name.clone(), + rule: rule.clone(), + }], + ) + .expect("merge") + .policy; + let yaml = openshell_policy::serialize_sandbox_policy(&merged).expect("serialize"); + openshell_policy::parse_sandbox_policy(&yaml).expect("policy YAML must round-trip"); + } + #[test] fn test_generate_proposals_with_l7_samples() { use openshell_core::proto::L7RequestSample; diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 054a3c34c6..3ea1c9f3d7 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -6048,6 +6048,15 @@ fn generate_security_notes(rule: &NetworkPolicyRule) -> String { for endpoint in &rule.endpoints { let host = endpoint.host.to_lowercase(); + // Supplemental roots change which upstream identities this sandbox + // trusts. Require human review even when the prover finds no reach or + // capability delta; auto mode must never silently approve a CA change. + if !endpoint.upstream_ca_pem.is_empty() { + notes.push(format!( + "Endpoint '{host}' adds a supplemental upstream CA and requires review." + )); + } + if endpoint.allow_uninspected_credentials { notes.push(format!( "Endpoint '{host}' explicitly allows credentials on traffic OpenShell cannot inspect or rewrite." @@ -6355,6 +6364,7 @@ fn map_policy_merge_error(error: openshell_policy::PolicyMergeError) -> Status { } openshell_policy::PolicyMergeError::InvalidInputPolicy { .. } | openshell_policy::PolicyMergeError::McpContractConflict { .. } + | openshell_policy::PolicyMergeError::UpstreamCaConflict { .. } | openshell_policy::PolicyMergeError::NewBinaryWouldInheritAuthorization { .. } | openshell_policy::PolicyMergeError::ExistingBinariesWouldInheritAuthorization { .. @@ -7125,6 +7135,20 @@ mod tests { }) } + #[test] + fn security_notes_flag_supplemental_ca() { + let rule = NetworkPolicyRule { + endpoints: vec![NetworkEndpoint { + host: "api.example.test".into(), + port: 443, + upstream_ca_pem: "-----BEGIN CERTIFICATE-----\nreviewed separately\n".into(), + ..Default::default() + }], + ..Default::default() + }; + assert!(generate_security_notes(&rule).contains("supplemental upstream CA")); + } + fn mcp_policy_with_versions(versions: &[&str]) -> ProtoSandboxPolicy { let mut policy = openshell_policy::restrictive_default_policy(); policy.network_policies.insert( @@ -13143,6 +13167,107 @@ mod tests { ); } + /// A reviewed CA proposal enters the sandbox-owned policy and reload + /// coverage must include its trust bundle, not just the destination. + #[tokio::test] + async fn approved_agent_ca_proposal_persists_and_requires_ca_reload() { + let state = test_server_state().await; + let sandbox_id = "sb-agent-ca"; + let sandbox_name = "agent-ca"; + state + .store + .put_message(&test_sandbox( + sandbox_id, + sandbox_name, + ProtoSandboxPolicy::default(), + vec![], + )) + .await + .unwrap(); + seed_sandbox_approval_mode(&state, sandbox_name, "auto").await; + let pem = rcgen::generate_simple_self_signed(vec!["api.example.test".into()]) + .unwrap() + .cert + .pem(); + let proposed_rule = NetworkPolicyRule { + name: "internal_api".into(), + endpoints: vec![NetworkEndpoint { + host: "api.example.test".into(), + port: 443, + upstream_ca_pem: pem.clone(), + ..Default::default() + }], + binaries: vec![NetworkBinary { + path: "/usr/bin/curl".into(), + ..Default::default() + }], + }; + handle_submit_policy_analysis( + &state, + with_user(Request::new(SubmitPolicyAnalysisRequest { + name: sandbox_name.into(), + analysis_mode: "agent_authored".into(), + proposed_chunks: vec![PolicyChunk { + rule_name: "internal_api".into(), + proposed_rule: Some(proposed_rule.clone()), + ..Default::default() + }], + ..Default::default() + })), + ) + .await + .unwrap(); + let draft = handle_get_draft_policy( + &state, + with_user(Request::new(GetDraftPolicyRequest { + name: sandbox_name.into(), + workspace: "default".into(), + ..Default::default() + })), + ) + .await + .unwrap() + .into_inner(); + let chunk = &draft.chunks[0]; + assert_eq!(chunk.status, "pending"); + assert!(chunk.security_notes.contains("supplemental upstream CA")); + assert_eq!( + chunk.proposed_rule.as_ref().unwrap().endpoints[0].upstream_ca_pem, + pem + ); + handle_approve_draft_chunk( + &state, + authed_request(ApproveDraftChunkRequest { + name: sandbox_name.into(), + chunk_id: chunk.id.clone(), + workspace: "default".into(), + review_token: chunk.review_token.clone(), + }), + ) + .await + .unwrap(); + let stored = state + .store + .get_latest_policy(sandbox_id) + .await + .unwrap() + .unwrap(); + let mut policy = ProtoSandboxPolicy::decode(stored.policy_payload.as_slice()).unwrap(); + assert_eq!( + policy.network_policies["internal_api"].endpoints[0].upstream_ca_pem, + pem + ); + assert!(policy_covers_rule(&policy, &proposed_rule)); + policy + .network_policies + .get_mut("internal_api") + .unwrap() + .endpoints[0] + .upstream_ca_pem + .clear(); + assert!(!policy_covers_rule(&policy, &proposed_rule)); + } + #[tokio::test] async fn mechanistic_dedup_auto_approval_rechecks_edited_stored_rule() { let state = test_server_state().await; diff --git a/crates/openshell-server/tests/prover_policy_round_trip.rs b/crates/openshell-server/tests/prover_policy_round_trip.rs new file mode 100644 index 0000000000..d466bb3548 --- /dev/null +++ b/crates/openshell-server/tests/prover_policy_round_trip.rs @@ -0,0 +1,147 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Regression for the incident where the gateway prover rejected every +//! mechanistic advisor proposal with "parse policy failed: parsing policy +//! YAML": the supervisor's binary placeholder `-` was copied verbatim into +//! the proposed rule's `binaries` list, and `serialize_sandbox_policy` +//! emits it as the plain YAML scalar `path: -`, which libyaml re-parses as +//! a block-sequence indicator. The placeholder is now filtered at the +//! source (`mechanistic_mapper::is_unknown_binary`, covered by that +//! crate's `test_generate_proposals_placeholder_binary_is_unknown`). +//! +//! This test pins the gateway-side contract independently: the candidate +//! policy the prover validates for a standard mechanistic proposal — the +//! chart's baseline policy merged with an advisor-proposed `allow` rule +//! that has NO binaries — must survive the exact round-trip +//! `run_prover_findings` performs: `merge_policy` → +//! `serialize_sandbox_policy` → prover `parse_policy_str`. +use openshell_core::proto::{ + FilesystemPolicy, LandlockPolicy, NetworkEndpoint, NetworkPolicyRule, ProcessPolicy, + SandboxPolicy, +}; +use openshell_policy::{PolicyMergeOp, merge_policy, serialize_sandbox_policy}; +use openshell_prover::policy::parse_policy_str; + +fn chart_baseline_policy() -> SandboxPolicy { + SandboxPolicy { + version: 1, + filesystem: Some(FilesystemPolicy { + include_workdir: true, + read_only: [ + "/usr", + "/app", + "/lib", + "/lib64", + "/bin", + "/sbin", + "/proc", + "/dev/urandom", + "/etc", + "/var/lib/dpkg", + "/var/log", + ] + .into_iter() + .map(String::from) + .collect(), + read_write: ["/sandbox", "/tmp", "/dev/null"] + .into_iter() + .map(String::from) + .collect(), + }), + landlock: Some(LandlockPolicy { + compatibility: "best_effort".to_string(), + }), + process: Some(ProcessPolicy { + run_as_user: "sandbox".to_string(), + run_as_group: "sandbox".to_string(), + }), + ..Default::default() + } +} + +#[test] +fn mechanistic_candidate_policy_survives_prover_round_trip() { + // The rule shape the (fixed) mechanistic mapper emits for a 443 CONNECT + // denial with an unresolvable binary: advisor-proposed endpoint, no + // binaries list. + let rule = NetworkPolicyRule { + name: "allow_example_com_443".to_string(), + endpoints: vec![NetworkEndpoint { + host: "example.com".to_string(), + port: 443, + ports: vec![443], + advisor_proposed: true, + ..Default::default() + }], + binaries: vec![], + }; + + let merged = merge_policy( + chart_baseline_policy(), + &[PolicyMergeOp::AddRule { + rule_name: "allow_example_com_443".to_string(), + rule, + }], + ) + .expect("merge") + .policy; + + let yaml = serialize_sandbox_policy(&merged).expect("serialize"); + let model = parse_policy_str(&yaml) + .unwrap_or_else(|e| panic!("prover must parse the candidate policy YAML: {e:?}")); + assert_eq!(model.network_policies.len(), 1); +} + +#[test] +fn ca_endpoints_survive_prover_round_trip() { + // Scenario-4 preflight: an advisor rule carrying upstream_ca_pem (PEM + // is multiline) must survive the same round-trip. The merge layer + // validates PEM usability, so this is a real self-signed test CA + // (CN=Scenario4 Test CA, generated with `openssl req -x509`, 30 days). + let pem = "-----BEGIN CERTIFICATE-----\n\ +MIIC+DCCAeCgAwIBAgIUNEncXTLhXQA4KBuy9ftpw916z5AwDQYJKoZIhvcNAQEL\n\ +BQAwHDEaMBgGA1UEAwwRU2NlbmFyaW80IFRlc3QgQ0EwHhcNMjYxMDA2MjIxMTAz\n\ +WhcNMjYxMTA1MjIxMTAzWjAcMRowGAYDVQQDDBFTY2VuYXJpbzQgVGVzdCBDQTCC\n\ +ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAOWjtqT4lhauCFHVwWriaE/d\n\ +ZGtuwdoEpwOoVuXnZNQpGpS3ixnqgj9TkrXZ+FpR00HKLKZBi3l+unTld0PPKuX2\n\ +PQemNPVQjr0qC/CL4lINO+lgZhfj/P/F9PGba8Hn/Xb61uyOKT0bemlv4FkcwlWH\n\ +Pz+Z4TwuI8kFqoFOyKCNsaL2QunuHkHMXV5YXPhRm1DDoXBhsc54T3iUwoT6NGre\n\ +d3q56X2nUMbCprjRHbnyo7ojQUZSKR7LulILJut2fkXPKnZiC8DbRZ677TylmMPg\n\ +unN4gNqcqDyThgcEU9q7CRIOcJrwem/jrSzHutle6rJbPSy7eNJvHrT4UKkVOl8C\n\ +AwEAAaMyMDAwHQYDVR0OBBYEFFEQDdwGHtgjBzqMDGovE/GtocmQMA8GA1UdEwEB\n\ +/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAA9SL33DKzU3WcA1xMg8j/ytcDQB\n\ +ssVipgLXjPB63udmu+r/p0Pff7hHychWZLipTrEmecoDzo5NzivFJ/S5U5e5LfPv\n\ +02JKXQyRztxPdOW2xQCOx3JHAljiU/ONBVEwxkbnL6jMyOzE4wWyzQpgdZf9RzW3\n\ +S5hY8l/LtmqVrccQ/Krq0IoYIp5wz3QdWZqLjo48ckEiRr/Q+vpDK4WQ8RVUqDyK\n\ +r4k+hyyDcMqhRI/PzW0h8WKOONv7dAKD1KnBwT6cm3Bd8J3k2tDbf4FHCVkNEOmj\n\ +YP4/4Z280GURxsg2At97z3Tl05WTvsb1rwZ4KvK7OQJcoyMAPmFFGg5ocSI=\n\ +-----END CERTIFICATE-----\n"; + let rule = NetworkPolicyRule { + name: "allow_ca_host_8443".to_string(), + endpoints: vec![NetworkEndpoint { + host: "ca-host.internal".to_string(), + port: 8443, + ports: vec![8443], + upstream_ca_pem: pem.to_string(), + advisor_proposed: true, + ..Default::default() + }], + binaries: vec![], + }; + let merged = merge_policy( + chart_baseline_policy(), + &[PolicyMergeOp::AddRule { + rule_name: rule.name.clone(), + rule, + }], + ) + .expect("merge") + .policy; + let yaml = serialize_sandbox_policy(&merged).expect("serialize"); + let model = parse_policy_str(&yaml) + .unwrap_or_else(|e| panic!("prover must parse CA-bearing candidate policy: {e:?}")); + assert_eq!(model.network_policies.len(), 1); + let ep = &model.network_policies.values().next().unwrap().endpoints[0]; + assert_eq!(ep.host, "ca-host.internal"); +} diff --git a/crates/openshell-supervisor-network/data/sandbox-policy.rego b/crates/openshell-supervisor-network/data/sandbox-policy.rego index e3fa6d36b4..f3abcf2ffa 100644 --- a/crates/openshell-supervisor-network/data/sandbox-policy.rego +++ b/crates/openshell-supervisor-network/data/sandbox-policy.rego @@ -876,10 +876,11 @@ command_matches(actual, expected) if { upper(actual) == upper(expected) } -# --- Matched endpoint config (for L7 and allowed_ips extraction) --- +# --- Matched endpoint config (for L7, allowed_ips, and upstream_ca_pem extraction) --- # Returns the raw endpoint object for the matched policy + host:port. # Used by Rust to extract L7 config (protocol, tls, enforcement, -# allow_encoded_slash) and/or allowed_ips for SSRF allowlist validation. +# allow_encoded_slash), allowed_ips for SSRF validation, and upstream_ca_pem +# for endpoint-scoped upstream TLS verification. # Per-policy helper: returns matching endpoint configs for a single policy. _policy_endpoint_configs(policy) := [ep | @@ -1023,7 +1024,7 @@ endpoint_path_matches_request(ep, request) if { } # An endpoint has extended config if it specifies an L7 protocol, allowed_ips, -# or an explicit tls mode (e.g. tls: skip). Explicit protocol "tcp" is the +# upstream_ca_pem, or an explicit tls mode (e.g. tls: skip). Explicit protocol "tcp" is the # authored spelling of plain L4 behavior and does not select an L7 config. endpoint_has_extended_config(ep) if { protocol := object.get(ep, "protocol", "") @@ -1038,3 +1039,7 @@ endpoint_has_extended_config(ep) if { endpoint_has_extended_config(ep) if { ep.tls } + +endpoint_has_extended_config(ep) if { + ep.upstream_ca_pem +} diff --git a/crates/openshell-supervisor-network/src/l7/tls.rs b/crates/openshell-supervisor-network/src/l7/tls.rs index 943117620b..e2df8e7a1e 100644 --- a/crates/openshell-supervisor-network/src/l7/tls.rs +++ b/crates/openshell-supervisor-network/src/l7/tls.rs @@ -139,6 +139,8 @@ impl CertCache { pub struct ProxyTlsState { cert_cache: CertCache, upstream_config: Arc, + upstream_roots: Option, + endpoint_configs: Mutex>>, } impl ProxyTlsState { @@ -147,9 +149,17 @@ impl ProxyTlsState { Self { cert_cache, upstream_config, + upstream_roots: None, + endpoint_configs: Mutex::new(HashMap::new()), } } + /// Attach the same system roots used by the default upstream config. + pub fn with_upstream_roots(mut self, system_ca_bundle: &str) -> Result { + self.upstream_roots = Some(build_upstream_root_store(system_ca_bundle)?); + Ok(self) + } + /// Get or generate a leaf cert for the hostname and return a TLS acceptor. fn acceptor_for(&self, hostname: &str) -> Result { let leaf = self.cert_cache.get_or_generate(hostname)?; @@ -165,6 +175,390 @@ impl ProxyTlsState { pub fn upstream_config(&self) -> &Arc { &self.upstream_config } + + /// Return an endpoint-only upstream verifier with system roots plus its CA. + /// The PEM content is the cache key, so changed policy content cannot reuse + /// an old verifier; the bounded cache avoids retaining removed policies. + pub fn endpoint_upstream_config(&self, pem: &str) -> Result> { + if pem.is_empty() { + return Ok(Arc::clone(&self.upstream_config)); + } + let mut cache = self + .endpoint_configs + .lock() + .map_err(|_| miette!("endpoint TLS config cache lock poisoned"))?; + if let Some(config) = cache.get(pem) { + return Ok(Arc::clone(config)); + } + let mut roots = self + .upstream_roots + .clone() + .ok_or_else(|| miette!("system upstream roots unavailable"))?; + let (added, ignored) = load_pem_certs_into_store(&mut roots, pem); + if added == 0 || ignored > 0 { + return Err(miette!( + "endpoint upstream CA contains no usable certificates or malformed certificates" + )); + } + let mut config = ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + config.alpn_protocols = vec![b"http/1.1".to_vec()]; + let config = Arc::new(config); + if cache.len() >= MAX_CACHED_CERTS { + cache.clear(); + } + cache.insert(pem.to_string(), Arc::clone(&config)); + Ok(config) + } +} + +#[cfg(test)] +mod endpoint_ca_tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + #[tokio::test] + async fn self_signed_ca_true_leaf_is_rejected_as_end_entity() { + // A single self-signed cert minted by `openssl req -x509` carries + // CA:TRUE, and webpki refuses a CA certificate used as an end-entity + // leaf (CaUsedAsEndEntity) even when it is also the trust anchor. + // Appliances must serve a leaf signed by a dedicated CA and propose + // that CA; this pins the rejection so a regression in the product's + // verification is caught. + let key = KeyPair::generate().unwrap(); + let mut params = CertificateParams::default(); + params.is_ca = IsCa::Ca(rcgen::BasicConstraints::Unconstrained); + params + .distinguished_name + .push(DnType::CommonName, "appliance"); + params.subject_alt_names = vec![rcgen::SanType::IpAddress("10.43.156.92".parse().unwrap())]; + let cert = params.self_signed(&key).unwrap(); + let pem = cert.pem(); + + let leaf = CertificateDer::from(cert.der().to_vec()); + let server_key = PrivateKeyDer::Pkcs8(rustls::pki_types::PrivatePkcs8KeyDer::from( + key.serialize_der(), + )); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + // The client aborts the handshake after the verification failure, so + // the server side must tolerate an AlertReceived shutdown. + let server = tokio::spawn(async move { + if let Ok((socket, _)) = listener.accept().await { + let config = ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![leaf], server_key) + .unwrap(); + let acceptor = TlsAcceptor::from(Arc::new(config)); + let Ok(mut tls) = acceptor.accept(socket).await else { + return; + }; + let mut buf = [0u8; 512]; + let _ = tls.read(&mut buf).await; + let _ = tls + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .await; + let _ = tls.shutdown().await; + } + }); + + let proxy_state = ProxyTlsState::new( + CertCache::new(SandboxCa::generate().unwrap()), + build_upstream_client_config("").unwrap(), + ) + .with_upstream_roots("") + .unwrap(); + let config = proxy_state.endpoint_upstream_config(&pem).unwrap(); + let socket = TcpStream::connect(addr).await.unwrap(); + let Err(error) = tls_connect_upstream(socket, "10.43.156.92", &config).await else { + panic!("CA:TRUE leaf must be rejected as an end entity"); + }; + assert!( + error.to_string().contains("CaUsedAsEndEntity"), + "unexpected error: {error}" + ); + let _ = server.await; + } + + #[tokio::test] + async fn ca_signed_leaf_with_ip_san_verifies_via_endpoint_ca() { + // Appliance contract: a dedicated CA (proposed via upstream_ca_pem) + // signs a leaf carrying the destination's IP SAN. The proxy's + // endpoint-scoped upstream config must verify the leaf for the + // IP-literal host. + let ca_key = KeyPair::generate().unwrap(); + let mut ca_params = CertificateParams::default(); + ca_params.is_ca = IsCa::Ca(rcgen::BasicConstraints::Unconstrained); + ca_params + .distinguished_name + .push(DnType::CommonName, "appliance-ca"); + let ca = ca_params.self_signed(&ca_key).unwrap(); + let ca_pem = ca.pem(); + + let leaf_key = KeyPair::generate().unwrap(); + let mut leaf_params = CertificateParams::default(); + leaf_params + .distinguished_name + .push(DnType::CommonName, "appliance"); + leaf_params.subject_alt_names = + vec![rcgen::SanType::IpAddress("10.43.156.92".parse().unwrap())]; + let leaf_cert = leaf_params.signed_by(&leaf_key, &ca, &ca_key).unwrap(); + let leaf = CertificateDer::from(leaf_cert.der().to_vec()); + let server_key = PrivateKeyDer::Pkcs8(rustls::pki_types::PrivatePkcs8KeyDer::from( + leaf_key.serialize_der(), + )); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let config = ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![leaf], server_key) + .unwrap(); + let acceptor = TlsAcceptor::from(Arc::new(config)); + let mut tls = acceptor.accept(socket).await.unwrap(); + let mut buf = [0u8; 512]; + let _ = tls.read(&mut buf).await.unwrap(); + tls.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .await + .unwrap(); + tls.shutdown().await.unwrap(); + }); + + let proxy_state = ProxyTlsState::new( + CertCache::new(SandboxCa::generate().unwrap()), + build_upstream_client_config("").unwrap(), + ) + .with_upstream_roots("") + .unwrap(); + let config = proxy_state.endpoint_upstream_config(&ca_pem).unwrap(); + let socket = TcpStream::connect(addr).await.unwrap(); + let mut tls = tls_connect_upstream(socket, "10.43.156.92", &config) + .await + .expect("CA-signed leaf with matching IP SAN must verify"); + tls.write_all(b"GET / HTTP/1.1\r\nHost: 10.43.156.92\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut response = [0u8; 128]; + let n = tokio::time::timeout(std::time::Duration::from_secs(3), tls.read(&mut response)) + .await + .unwrap() + .unwrap(); + assert!(String::from_utf8_lossy(&response[..n]).starts_with("HTTP/1.1 200")); + server.await.unwrap(); + } + + fn client_config_with_test_ca(pem: &str) -> Arc { + let mut roots = rustls::RootCertStore::empty(); + assert_eq!(load_pem_certs_into_store(&mut roots, pem), (1, 0)); + let mut config = ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + config.alpn_protocols = vec![b"http/1.1".to_vec()]; + Arc::new(config) + } + + #[tokio::test] + async fn endpoint_ca_trust_is_scoped_to_its_upstream_config() { + let upstream_ca = SandboxCa::generate().unwrap(); + #[cfg(feature = "bundled-ca-roots")] + let system_ca = SandboxCa::generate().unwrap(); + #[cfg(feature = "bundled-ca-roots")] + let system_pem = system_ca.cert_pem().to_string(); + #[cfg(not(feature = "bundled-ca-roots"))] + let system_pem = String::new(); + let server = Arc::new(ProxyTlsState::new( + CertCache::new(upstream_ca), + build_upstream_client_config("").unwrap(), + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let pem = server.cert_cache.ca.cert_pem().to_string(); + tokio::spawn(async move { + for _ in 0..2 { + let (stream, _) = listener.accept().await.unwrap(); + let server = Arc::clone(&server); + tokio::spawn(async move { + let _ = tls_terminate_client(stream, &server, "private.example.test").await; + }); + } + }); + let client = ProxyTlsState::new( + CertCache::new(SandboxCa::generate().unwrap()), + build_upstream_client_config(&system_pem).unwrap(), + ) + .with_upstream_roots(&system_pem) + .unwrap(); + let scoped = client.endpoint_upstream_config(&pem).unwrap(); + let stream = TcpStream::connect(address).await.unwrap(); + tls_connect_upstream(stream, "private.example.test", &scoped) + .await + .expect("endpoint CA should verify upstream"); + let stream = TcpStream::connect(address).await.unwrap(); + assert!( + tls_connect_upstream(stream, "private.example.test", client.upstream_config()) + .await + .is_err(), + "default upstream config must not trust endpoint CA" + ); + assert!(Arc::ptr_eq( + &scoped, + &client.endpoint_upstream_config(&pem).unwrap() + )); + assert!(!client.upstream_roots.as_ref().unwrap().roots.is_empty()); + + #[cfg(feature = "bundled-ca-roots")] + { + let system_server = Arc::new(ProxyTlsState::new( + CertCache::new(system_ca), + build_upstream_client_config("").unwrap(), + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + for _ in 0..2 { + let (stream, _) = listener.accept().await.unwrap(); + let _ = + tls_terminate_client(stream, &system_server, "system.example.test").await; + } + }); + let stream = TcpStream::connect(address).await.unwrap(); + tls_connect_upstream(stream, "system.example.test", client.upstream_config()) + .await + .expect("default config must retain system roots"); + let stream = TcpStream::connect(address).await.unwrap(); + tls_connect_upstream(stream, "system.example.test", &scoped) + .await + .expect("endpoint config must retain system roots"); + } + } + + #[tokio::test] + async fn endpoint_ca_keeps_tls_termination_and_credential_relay() { + use crate::l7::relay::{L7EvalContext, relay_passthrough_with_credentials}; + use crate::opa::OpaEngine; + use openshell_core::secrets::SecretResolver; + + let proxy_ca = SandboxCa::generate().unwrap(); + let proxy_pem = proxy_ca.cert_pem().to_string(); + let upstream_ca = SandboxCa::generate().unwrap(); + let upstream_pem = upstream_ca.cert_pem().to_string(); + let upstream_state = ProxyTlsState::new( + CertCache::new(upstream_ca), + build_upstream_client_config("").unwrap(), + ); + let upstream_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_addr = upstream_listener.local_addr().unwrap(); + let (captured_tx, captured_rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + let (socket, _) = upstream_listener.accept().await.unwrap(); + let mut tls = tls_terminate_client(socket, &upstream_state, "api.example.test") + .await + .unwrap(); + let mut request = Vec::new(); + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + let mut buf = [0u8; 512]; + let n = tokio::time::timeout(std::time::Duration::from_secs(3), tls.read(&mut buf)) + .await + .unwrap() + .unwrap(); + assert!( + n > 0 && request.len() + n <= 2048, + "HTTP request headers incomplete or too large" + ); + request.extend_from_slice(&buf[..n]); + } + captured_tx + .send(String::from_utf8_lossy(&request).to_string()) + .unwrap(); + tls.write_all( + b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ) + .await + .unwrap(); + tls.shutdown().await.unwrap(); + }); + + let (child_env, resolver) = SecretResolver::from_provider_env( + std::iter::once(("API_TOKEN".to_string(), "real-secret".to_string())).collect(), + ); + let placeholder = child_env.get("API_TOKEN").unwrap().clone(); + let proxy_state = ProxyTlsState::new( + CertCache::new(proxy_ca), + build_upstream_client_config("").unwrap(), + ) + .with_upstream_roots("") + .unwrap(); + let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_addr = proxy_listener.local_addr().unwrap(); + let proxy_task = tokio::spawn(async move { + let (client, _) = proxy_listener.accept().await.unwrap(); + let mut client_tls = tls_terminate_client(client, &proxy_state, "api.example.test") + .await + .unwrap(); + let upstream = TcpStream::connect(upstream_addr).await.unwrap(); + let config = proxy_state.endpoint_upstream_config(&upstream_pem).unwrap(); + let mut upstream_tls = tls_connect_upstream(upstream, "api.example.test", &config) + .await + .unwrap(); + let engine = OpaEngine::from_strings( + include_str!("../../data/sandbox-policy.rego"), + "network_policies: {}\n", + ) + .unwrap(); + let guard = engine + .generation_guard(engine.current_generation()) + .unwrap(); + let ctx = L7EvalContext { + host: "api.example.test".into(), + port: 443, + request_default_port: Some(443), + secret_resolver: resolver.map(Arc::new), + ..Default::default() + }; + relay_passthrough_with_credentials( + &mut client_tls, + &mut upstream_tls, + &ctx, + &guard, + None, + ) + .await + .unwrap(); + }); + let client_config = client_config_with_test_ca(&proxy_pem); + let socket = TcpStream::connect(proxy_addr).await.unwrap(); + let mut client_tls = tls_connect_upstream(socket, "api.example.test", &client_config) + .await + .unwrap(); + client_tls.write_all(format!("GET / HTTP/1.1\r\nHost: api.example.test\r\nAuthorization: Bearer {placeholder}\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap(); + let request = tokio::time::timeout(std::time::Duration::from_secs(3), captured_rx) + .await + .unwrap() + .unwrap(); + assert!( + request.contains("Authorization: Bearer real-secret"), + "upstream received: {request:?}" + ); + assert!(!request.contains(&placeholder)); + let mut response = [0u8; 256]; + let n = tokio::time::timeout( + std::time::Duration::from_secs(3), + client_tls.read(&mut response), + ) + .await + .unwrap() + .unwrap(); + assert!(String::from_utf8_lossy(&response[..n]).starts_with("HTTP/1.1 204")); + drop(client_tls); + tokio::time::timeout(std::time::Duration::from_secs(3), proxy_task) + .await + .unwrap() + .unwrap(); + } } /// Accept TLS from a sandbox client, presenting a dynamic cert for the hostname. diff --git a/crates/openshell-supervisor-network/src/opa.rs b/crates/openshell-supervisor-network/src/opa.rs index 98632686e9..dbc6e9ec37 100644 --- a/crates/openshell-supervisor-network/src/opa.rs +++ b/crates/openshell-supervisor-network/src/opa.rs @@ -1955,6 +1955,9 @@ fn proto_to_opa_data_json(proto: &ProtoSandboxPolicy, entrypoint_pid: u32) -> St if !e.tls.is_empty() { ep["tls"] = e.tls.clone().into(); } + if !e.upstream_ca_pem.is_empty() { + ep["upstream_ca_pem"] = e.upstream_ca_pem.clone().into(); + } if !e.enforcement.is_empty() { ep["enforcement"] = e.enforcement.clone().into(); } @@ -4960,6 +4963,97 @@ network_policies: assert_eq!(l7.enforcement, crate::l7::EnforcementMode::Enforce); } + #[test] + fn endpoint_ca_survives_proto_to_opa_selection_without_l7_protocol() { + let mut policy = test_proto(); + let ca_pem = crate::l7::tls::SandboxCa::generate() + .unwrap() + .cert_pem() + .to_string(); + let endpoint = &mut policy + .network_policies + .get_mut("claude_code") + .unwrap() + .endpoints[0]; + endpoint.upstream_ca_pem = ca_pem.clone(); + let engine = OpaEngine::from_proto(&policy).unwrap(); + let input = NetworkInput { + host: "api.anthropic.com".into(), + port: 443, + binary_path: PathBuf::from("/usr/local/bin/claude"), + binary_sha256: "unused".into(), + ancestors: vec![], + cmdline_paths: vec![], + }; + let config = engine.query_endpoint_config(&input).unwrap().unwrap(); + assert_eq!( + get_str(&config, "upstream_ca_pem").as_deref(), + Some(ca_pem.as_str()) + ); + } + + #[test] + fn endpoint_ca_survives_full_egress_authorization_decision() { + // The CONNECT handler reads the supplemental upstream CA from + // `EgressAuthorization::endpoint_configs`; this exercises the whole + // proto -> OPA data -> egress_authorization chain, not just + // query_endpoint_config. + let mut policy = test_proto(); + let ca_pem = crate::l7::tls::SandboxCa::generate() + .unwrap() + .cert_pem() + .to_string(); + policy.network_policies.insert( + "ca_lane".to_string(), + NetworkPolicyRule { + name: "ca_lane".to_string(), + endpoints: vec![NetworkEndpoint { + host: "10.43.156.92".to_string(), + port: 443, + upstream_ca_pem: ca_pem.clone(), + ..Default::default() + }], + binaries: vec![NetworkBinary { + path: "/usr/bin/curl".to_string(), + ..Default::default() + }], + }, + ); + let engine = OpaEngine::from_proto(&policy).unwrap(); + let input = NetworkInput { + host: "10.43.156.92".into(), + port: 443, + binary_path: PathBuf::from("/usr/bin/curl"), + binary_sha256: "unused".into(), + ancestors: vec![], + cmdline_paths: vec![], + }; + let authorization = engine.authorize_egress(&input).unwrap(); + assert!( + matches!(authorization.action, NetworkAction::Allow { .. }), + "expected allow, got {:?}", + authorization.action + ); + let config = authorization + .endpoint_configs + .first() + .expect("endpoint_configs must carry the matched endpoint"); + let extracted = match config { + regorus::Value::Object(fields) => fields + .get(®orus::Value::String("upstream_ca_pem".into())) + .map(|value| match value { + regorus::Value::String(s) => s.as_ref().to_string(), + _ => String::new(), + }), + _ => None, + } + .unwrap_or_default(); + assert_eq!( + extracted, ca_pem, + "upstream_ca_pem must survive into the egress decision endpoint_configs" + ); + } + #[test] fn explicit_tcp_authorizes_as_l4_without_endpoint_config() { let engine = l7_engine(); diff --git a/crates/openshell-supervisor-network/src/policy_local.rs b/crates/openshell-supervisor-network/src/policy_local.rs index 520a22adba..a478bf1cbd 100644 --- a/crates/openshell-supervisor-network/src/policy_local.rs +++ b/crates/openshell-supervisor-network/src/policy_local.rs @@ -487,6 +487,58 @@ fn collect_shorthand_log_files(log_dir: &Path, max_files: usize) -> std::io::Res .collect()) } +/// Connect a gateway client for a unary policy-local RPC, addressing a fleet +/// replica rather than the configured endpoint. +/// +/// The configured endpoint is the headless Service when a fleet is +/// configured — a discovery address the gateway's serving certificate +/// deliberately omits, so dialling it fails the TLS handshake. Address a +/// replica instead, like every other unary RPC this sandbox makes (see +/// `flush_proposals_to_gateway` in the sandbox crate). +/// +/// `sandbox_key` only picks which replicas this call warms; any replica can +/// serve these store reads, so the sandbox *name* this caller has is fine. +async fn connect_gateway_for_rpc( + endpoint: &str, + sandbox_key: &str, +) -> std::result::Result +{ + let mut last_error = None; + for attempt in 0..3 { + let Some(target) = + openshell_core::gateway_fleet::unary_endpoint(endpoint, sandbox_key, attempt).await + else { + last_error = Some(( + 503, + error_payload( + "gateway_unavailable", + "gateway fleet membership is not resolved yet; retry shortly".to_string(), + ), + )); + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + continue; + }; + match openshell_core::grpc_client::CachedOpenShellClient::connect(&target).await { + Ok(client) => return Ok(client), + Err(error) => { + last_error = Some(( + 502, + error_payload("gateway_connect_failed", error.to_string()), + )); + if attempt + 1 < 3 { + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + } + } + } + Err(last_error.unwrap_or_else(|| { + ( + 503, + error_payload("gateway_unavailable", "no gateway endpoint".to_string()), + ) + })) +} + async fn submit_proposal(ctx: &PolicyLocalContext, body: &[u8]) -> (u16, serde_json::Value) { let Some(endpoint) = ctx.gateway_endpoint.as_deref() else { return ( @@ -528,21 +580,11 @@ async fn submit_proposal(ctx: &PolicyLocalContext, body: &[u8]) -> (u16, serde_j Err(error) => return (400, error_payload("invalid_proposal", error)), }; - let client = match openshell_core::grpc_client::CachedOpenShellClient::connect(endpoint).await { - Ok(client) => { - client.set_workspace(workspace); - client - } - Err(error) => { - return ( - 502, - serde_json::json!({ - "error": "gateway_connect_failed", - "detail": error.to_string() - }), - ); - } + let client = match connect_gateway_for_rpc(endpoint, sandbox_name).await { + Ok(client) => client, + Err((status, error)) => return (status, error), }; + client.set_workspace(workspace); // Pre-compute the audit summaries before handing `chunks` to the // gateway client (which consumes the vec). The summaries pair up with @@ -994,9 +1036,7 @@ async fn open_lookup_session( ), )); } - let client = openshell_core::grpc_client::CachedOpenShellClient::connect(endpoint) - .await - .map_err(|e| (502, error_payload("gateway_connect_failed", e.to_string())))?; + let client = connect_gateway_for_rpc(endpoint, sandbox_name).await?; client.set_workspace(workspace); Ok(LookupSession { client, @@ -1138,6 +1178,16 @@ fn network_endpoint_from_json( if endpoint.host.trim().is_empty() { return Err("endpoint.host is required".to_string()); } + if endpoint.upstream_ca_pem.as_deref() == Some("") { + return Err("upstream_ca_pem cannot be empty".to_string()); + } + if endpoint + .upstream_ca_pem + .as_deref() + .is_some_and(|pem| !openshell_policy::validate_upstream_ca_pem(pem)) + { + return Err("upstream_ca_pem must contain at least one usable PEM certificate".to_string()); + } if let Some(reason) = openshell_policy::agent_authored_transport_rejection(&endpoint.protocol, &endpoint.tls) { @@ -1199,6 +1249,7 @@ fn network_endpoint_from_json( port, protocol: endpoint.protocol, tls: endpoint.tls, + upstream_ca_pem: endpoint.upstream_ca_pem.unwrap_or_default(), enforcement: endpoint.enforcement, access: endpoint.access, rules, @@ -1370,6 +1421,8 @@ struct NetworkEndpointJson { #[serde(default)] tls: String, #[serde(default)] + upstream_ca_pem: Option, + #[serde(default)] enforcement: String, #[serde(default)] access: String, @@ -1482,6 +1535,179 @@ mod tests { ); } + #[test] + fn proposal_ca_bundle_is_preserved_and_validated() { + let cert = rcgen::generate_simple_self_signed(vec!["api.example.test".into()]).unwrap(); + let pem = cert.cert.pem(); + let proposal = |ca: &str, protocol: &str, tls: &str| { + serde_json::to_vec(&serde_json::json!({ + "operations": [{"addRule": { + "ruleName": "internal_api", + "rule": {"endpoints": [{ + "host": "api.example.test", "port": 443, + "protocol": protocol, "tls": tls, + "access": "full", + "upstream_ca_pem": ca + }]} + }}] + })) + .unwrap() + }; + let chunks = proposal_chunks_from_body(&proposal(&pem, "rest", "terminate")).unwrap(); + assert_eq!( + chunks[0].proposed_rule.as_ref().unwrap().endpoints[0].upstream_ca_pem, + pem + ); + for invalid in [ + "", + "garbage", + "-----BEGIN PRIVATE KEY-----\nAA==\n-----END PRIVATE KEY-----", + ] { + assert!(proposal_chunks_from_body(&proposal(invalid, "rest", "terminate")).is_err()); + } + let cert_with_key = + format!("{pem}-----BEGIN PRIVATE KEY-----\nAA==\n-----END PRIVATE KEY-----"); + assert!(proposal_chunks_from_body(&proposal(&cert_with_key, "rest", "terminate")).is_err()); + for (protocol, tls) in [("tcp", "terminate"), ("rest", "skip")] { + assert!(proposal_chunks_from_body(&proposal(&pem, protocol, tls)).is_err()); + } + let other_pem = rcgen::generate_simple_self_signed(vec!["api.example.test".into()]) + .unwrap() + .cert + .pem(); + let overlapping = serde_json::to_vec(&serde_json::json!({ + "operations": [{"addRule": { + "ruleName": "internal_api", + "rule": {"endpoints": [ + {"host": "api.example.test", "port": 443, "access": "full", "upstream_ca_pem": pem}, + {"host": "api.example.test", "port": 443, "access": "full", "upstream_ca_pem": other_pem} + ]} + }}] + })).unwrap(); + let chunk = proposal_chunks_from_body(&overlapping).unwrap().remove(0); + let result = openshell_policy::merge_policy( + ProtoSandboxPolicy { + version: 1, + ..Default::default() + }, + &[openshell_policy::PolicyMergeOp::AddRule { + rule_name: chunk.rule_name, + rule: chunk.proposed_rule.unwrap(), + }], + ); + let error = result.expect_err("overlapping destinations cannot carry conflicting CAs"); + assert!(error.to_string().contains("upstream_ca_pem")); + } + + /// Exercise the advisor wire rule through merge and the supervisor's + /// upstream verifier, with no trust added to an unrelated endpoint. + #[tokio::test] + async fn proposed_ca_dials_only_its_selected_destination() { + use crate::l7::tls::{ + CertCache, ProxyTlsState, SandboxCa, build_upstream_client_config, + tls_connect_upstream, tls_terminate_client, + }; + use crate::opa::{NetworkInput, OpaEngine}; + use openshell_policy::{PolicyMergeOp, merge_policy}; + + let customer_ca = SandboxCa::generate().unwrap(); + let pem = customer_ca.cert_pem().to_string(); + let server = Arc::new(ProxyTlsState::new( + CertCache::new(customer_ca), + build_upstream_client_config("").unwrap(), + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + for _ in 0..2 { + let (stream, _) = listener.accept().await.unwrap(); + let server = Arc::clone(&server); + tokio::spawn(async move { + let _ = tls_terminate_client(stream, &server, "private.example.test").await; + }); + } + }); + + let body = serde_json::to_vec(&serde_json::json!({ + "operations": [{"addRule": { + "ruleName": "private_api", + "rule": {"binaries": [{"path": "/usr/bin/curl"}], "endpoints": [{ + "host": "private.example.test", "port": 443, + "access": "full", "upstream_ca_pem": pem + }, { + "host": "other.example.test", "port": 443, + "access": "full" + }]} + }}] + })) + .unwrap(); + let chunk = proposal_chunks_from_body(&body).unwrap().remove(0); + let proposed = chunk.proposed_rule.unwrap(); + let merged = merge_policy( + ProtoSandboxPolicy { + version: 1, + ..Default::default() + }, + &[PolicyMergeOp::AddRule { + rule_name: chunk.rule_name, + rule: proposed, + }], + ) + .unwrap() + .policy; + let engine = OpaEngine::from_proto(&merged).unwrap(); + let selected_ca = |host: &str| { + let config = engine + .query_endpoint_config(&NetworkInput { + host: host.into(), + port: 443, + binary_path: PathBuf::from("/usr/bin/curl"), + binary_sha256: "unused".into(), + ancestors: vec![], + cmdline_paths: vec![], + }) + .unwrap(); + match config { + Some(regorus::Value::Object(fields)) => fields + .get(®orus::Value::String("upstream_ca_pem".into())) + .and_then(|value| match value { + regorus::Value::String(value) => Some(value.to_string()), + _ => None, + }) + .unwrap_or_default(), + None => String::new(), + _ => panic!("expected endpoint config object"), + } + }; + let selected_pem = selected_ca("private.example.test"); + assert_eq!(selected_pem, pem); + assert!(selected_ca("other.example.test").is_empty()); + + let client = ProxyTlsState::new( + CertCache::new(SandboxCa::generate().unwrap()), + build_upstream_client_config("").unwrap(), + ) + .with_upstream_roots("") + .unwrap(); + let scoped = client.endpoint_upstream_config(&selected_pem).unwrap(); + tls_connect_upstream( + tokio::net::TcpStream::connect(address).await.unwrap(), + "private.example.test", + &scoped, + ) + .await + .unwrap(); + assert!( + tls_connect_upstream( + tokio::net::TcpStream::connect(address).await.unwrap(), + "private.example.test", + client.upstream_config(), + ) + .await + .is_err() + ); + } + #[test] fn proposal_chunks_from_body_rejects_query_in_l7_path() { let body = br#"{ @@ -2084,6 +2310,36 @@ mod tests { ); } + /// A destination-only reload is insufficient for a CA proposal. The + /// in-process policy update completes the wait once the CA is installed. + #[tokio::test(start_paused = true)] + async fn wait_for_ca_proposal_requires_ca_bearing_reload() { + let pem = crate::l7::tls::SandboxCa::generate() + .unwrap() + .cert_pem() + .to_string(); + let mut proposed = proposed_curl_rule_for_github(); + proposed.endpoints[0].upstream_ca_pem = pem; + let mut initial = proposed.clone(); + initial.endpoints[0].upstream_ca_pem.clear(); + let ctx = Arc::new(PolicyLocalContext::new( + Some(policy_with_rule(initial)), + None, + None, + AgentProposals::new(false), + test_workspace_rx(), + )); + let reloads = Arc::clone(&ctx); + let covering = proposed.clone(); + let install = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + reloads.set_current_policy(policy_with_rule(covering)).await; + }); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2); + assert!(wait_for_local_policy_to_cover(&ctx, &proposed, deadline).await); + install.await.unwrap(); + } + #[tokio::test] async fn wait_does_not_wake_on_unrelated_policy_change() { // John's false-wakeup case: a *different* rule gets added to the diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 7df34f9149..77a5bff7f1 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -1898,12 +1898,10 @@ async fn handle_tcp_connection( emit_activity(&activity_tx, true, "connect_policy"); respond( &mut client, - &build_json_error_response_with_reason( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("CONNECT {host_lc}:{port} not permitted by policy"), - &deny_reason, + Some(&deny_reason), + agent_proposals.enabled(), ), ) .await?; @@ -1920,6 +1918,7 @@ async fn handle_tcp_connection( port, activity_tx.as_ref(), error, + agent_proposals.enabled(), ) .await?; return Ok(()); @@ -2090,8 +2089,15 @@ async fn handle_tcp_connection( if let Err(error) = relay::validate_route_generation(l7_route, connect_generation_guard.captured_generation()) { - reject_stale_connect_policy(&mut client, &host_lc, port, activity_tx.as_ref(), error) - .await?; + reject_stale_connect_policy( + &mut client, + &host_lc, + port, + activity_tx.as_ref(), + error, + agent_proposals.enabled(), + ) + .await?; return Ok(()); } @@ -2111,14 +2117,22 @@ async fn handle_tcp_connection( connect_generation_guard.captured_generation(), connect_generation_guard.current_generation(), ), + agent_proposals.enabled(), ) .await?; return Ok(()); }; let mut upstream = upstream_result.into_diagnostic()?; if let Err(error) = connect_generation_guard.ensure_current() { - reject_stale_connect_policy(&mut client, &host_lc, port, activity_tx.as_ref(), error) - .await?; + reject_stale_connect_policy( + &mut client, + &host_lc, + port, + activity_tx.as_ref(), + error, + agent_proposals.enabled(), + ) + .await?; return Ok(()); } @@ -2219,8 +2233,24 @@ async fn handle_tcp_connection( let tls_result = async { let mut tls_client = crate::l7::tls::tls_terminate_client(client, tls, &host_lc).await?; + let endpoint_ca = decision + .endpoint + .policy_configs + .first() + .and_then(|config| match config { + regorus::Value::Object(fields) => { + fields.get(®orus::Value::String("upstream_ca_pem".into())) + } + _ => None, + }) + .and_then(|value| match value { + regorus::Value::String(value) => Some(value.as_ref()), + _ => None, + }) + .unwrap_or(""); + let upstream_config = tls.endpoint_upstream_config(endpoint_ca)?; let mut tls_upstream = - crate::l7::tls::tls_connect_upstream(upstream, &host_lc, tls.upstream_config()) + crate::l7::tls::tls_connect_upstream(upstream, &host_lc, &upstream_config) .await?; let Some(relay_context) = relay::prepare_http_relay(l7_route, &opa_engine, &decision, &ctx) @@ -3428,6 +3458,7 @@ async fn reject_stale_connect_policy( port: u16, activity_tx: Option<&ActivitySender>, error: miette::Report, + agent_proposals_enabled: bool, ) -> Result<()> { warn!( host, @@ -3439,11 +3470,10 @@ async fn reject_stale_connect_policy( emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("CONNECT {host}:{port} not permitted because policy changed"), + None, + agent_proposals_enabled, ), ) .await @@ -5007,11 +5037,10 @@ async fn handle_forward_proxy( emit_activity_simple(activity_tx, true, "forward_policy"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{telemetry_path} not permitted by policy"), + None, + agent_proposals.enabled(), ), ) .await?; @@ -5049,11 +5078,10 @@ async fn handle_forward_proxy( emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{telemetry_path} not permitted by policy"), + None, + agent_proposals.enabled(), ), ) .await?; @@ -5131,6 +5159,7 @@ async fn handle_forward_proxy( .as_ref() .map(|ctx| ctx.workspace()) .unwrap_or_default(); + let agent_proposals_enabled = agent_proposals.enabled(); let mut l7_ctx = relay::http_context( &decision, provider_credentials, @@ -5173,11 +5202,10 @@ async fn handle_forward_proxy( emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{telemetry_path} not permitted by policy"), + None, + agent_proposals_enabled, ), ) .await?; @@ -5198,13 +5226,12 @@ async fn handle_forward_proxy( emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!( "{method} {host_lc}:{port}{telemetry_path} not permitted by policy" ), + None, + agent_proposals_enabled, ), ) .await?; @@ -5229,13 +5256,12 @@ async fn handle_forward_proxy( emit_activity_simple(activity_tx, true, "l7_policy"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!( "{method} {host_lc}:{port}{telemetry_path} did not match an L7 endpoint path" ), + None, + agent_proposals_enabled, ), ) .await?; @@ -5561,13 +5587,12 @@ async fn handle_forward_proxy( ); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!( "{method} {host_lc}:{port}{telemetry_path} denied by L7 policy: {reason}" ), + None, + agent_proposals_enabled, ), ) .await?; @@ -5663,11 +5688,10 @@ async fn handle_forward_proxy( emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{telemetry_path} not permitted by policy"), + None, + agent_proposals_enabled, ), ) .await?; @@ -5697,11 +5721,10 @@ async fn handle_forward_proxy( ); respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{telemetry_path} not permitted by policy"), + None, + agent_proposals_enabled, ), ) .await?; @@ -5925,11 +5948,10 @@ async fn handle_forward_proxy( } respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{telemetry_path} not permitted by policy"), + None, + agent_proposals_enabled, ), ) .await?; @@ -6013,11 +6035,10 @@ async fn handle_forward_proxy( } respond( client, - &build_json_error_response( - 403, - "Forbidden", - "policy_denied", + &build_policy_deny_response( &format!("{method} {host_lc}:{port}{path} not permitted by policy"), + None, + agent_proposals_enabled, ), ) .await?; @@ -6192,11 +6213,7 @@ async fn respond(client: &mut TcpStream, bytes: &[u8]) -> Result<()> { /// Returns bytes ready to write to the client socket. The body is a JSON /// object with `error` and `detail` fields, matching the format used by the /// L7 deny path in `l7/rest.rs`. -fn build_json_error_response(status: u16, status_text: &str, error: &str, detail: &str) -> Vec { - let body = serde_json::json!({ - "error": error, - "detail": detail, - }); +fn build_json_body_response(status: u16, status_text: &str, body: serde_json::Value) -> Vec { let body_str = body.to_string(); format!( "HTTP/1.1 {status} {status_text}\r\n\ @@ -6211,32 +6228,34 @@ fn build_json_error_response(status: u16, status_text: &str, error: &str, detail .into_bytes() } -fn build_json_error_response_with_reason( - status: u16, - status_text: &str, - error: &str, +fn build_json_error_response(status: u16, status_text: &str, error: &str, detail: &str) -> Vec { + build_json_body_response( + status, + status_text, + serde_json::json!({ + "error": error, + "detail": detail, + }), + ) +} + +/// Build a policy denial with optional decision reason and advisor guidance. +fn build_policy_deny_response( detail: &str, - reason: &str, + reason: Option<&str>, + agent_proposals_enabled: bool, ) -> Vec { let mut body = serde_json::json!({ - "error": error, + "error": "policy_denied", "detail": detail, }); - if !reason.is_empty() { + if let Some(reason) = reason.filter(|reason| !reason.is_empty()) { body["reason"] = serde_json::json!(reason); } - let body_str = body.to_string(); - format!( - "HTTP/1.1 {status} {status_text}\r\n\ - Content-Type: application/json\r\n\ - Content-Length: {}\r\n\ - Connection: close\r\n\ - \r\n\ - {}", - body_str.len(), - body_str, - ) - .into_bytes() + if let Some(guidance) = crate::policy_local::agent_guidance_for(agent_proposals_enabled) { + body["agent_guidance"] = serde_json::json!(guidance); + } + build_json_body_response(403, "Forbidden", body) } fn build_middleware_deny_response( @@ -6961,14 +6980,59 @@ network_policies: assert!(body.get("reason_code").is_none()); } + #[test] + fn policy_deny_response_points_at_policy_local_when_advisor_enabled() { + let response = + build_policy_deny_response("GET host.example:443/ not permitted by policy", None, true); + let response = String::from_utf8(response).expect("UTF-8 error response"); + assert!(response.starts_with("HTTP/1.1 403 Forbidden")); + let (_, body) = response.split_once("\r\n\r\n").expect("HTTP response"); + let body: serde_json::Value = serde_json::from_str(body).expect("JSON response"); + assert_eq!(body["error"], "policy_denied"); + let guidance = body["agent_guidance"].as_str().expect("guidance present"); + assert!(guidance.contains("policy.local"), "guidance: {guidance}"); + assert!(guidance.contains("policy_advisor"), "guidance: {guidance}"); + + let response = build_policy_deny_response( + "CONNECT host.example:443 not permitted by policy", + Some("binary '/usr/bin/curl' not allowed in policy 'allow_api'"), + true, + ); + let response = String::from_utf8(response).expect("UTF-8 error response"); + let (_, body) = response.split_once("\r\n\r\n").expect("HTTP response"); + let body: serde_json::Value = serde_json::from_str(body).expect("JSON response"); + assert_eq!(body["error"], "policy_denied"); + assert!(body["reason"].as_str().unwrap().contains("not allowed")); + assert!( + body["agent_guidance"] + .as_str() + .unwrap() + .contains("policy.local") + ); + } + + #[test] + fn policy_deny_response_omits_guidance_when_advisor_disabled() { + let response = build_policy_deny_response( + "GET host.example:443/ not permitted by policy", + None, + false, + ); + let response = String::from_utf8(response).expect("UTF-8 error response"); + let (_, body) = response.split_once("\r\n\r\n").expect("HTTP response"); + let body: serde_json::Value = serde_json::from_str(body).expect("JSON response"); + assert_eq!(body["error"], "policy_denied"); + assert!(body.get("agent_guidance").is_none()); + } + #[test] fn policy_deny_response_includes_reason() { - let response = build_json_error_response_with_reason( - 403, - "Forbidden", - "policy_denied", + let response = build_policy_deny_response( "CONNECT api.example.com:443 not permitted by policy", - "binary '/usr/bin/node' not allowed in policy 'allow_api' (ancestors: [/usr/local/bin/claude])", + Some( + "binary '/usr/bin/node' not allowed in policy 'allow_api' (ancestors: [/usr/local/bin/claude])", + ), + false, ); let response = String::from_utf8(response).expect("UTF-8 error response"); assert!(response.starts_with("HTTP/1.1 403 Forbidden")); @@ -6988,12 +7052,10 @@ network_policies: #[test] fn policy_deny_response_omits_empty_reason() { - let response = build_json_error_response_with_reason( - 403, - "Forbidden", - "policy_denied", + let response = build_policy_deny_response( "CONNECT api.example.com:443 not permitted by policy", - "", + Some(""), + false, ); let response = String::from_utf8(response).expect("UTF-8 error response"); let (_, body) = response.split_once("\r\n\r\n").expect("HTTP response"); diff --git a/crates/openshell-supervisor-network/src/run.rs b/crates/openshell-supervisor-network/src/run.rs index f6a38025c4..bb34aea6d2 100644 --- a/crates/openshell-supervisor-network/src/run.rs +++ b/crates/openshell-supervisor-network/src/run.rs @@ -350,7 +350,10 @@ pub async fn run_networking( let upstream_config = build_upstream_client_config(&system_ca_bundle)?; let cert_cache = CertCache::new(ca); - let state = Arc::new(ProxyTlsState::new(cert_cache, upstream_config)); + let state = Arc::new( + ProxyTlsState::new(cert_cache, upstream_config) + .with_upstream_roots(&system_ca_bundle)?, + ); ocsf_emit!( ConfigStateChangeBuilder::new(ocsf_ctx()) .severity(SeverityId::Informational) diff --git a/crates/openshell-supervisor-process/src/skills/policy-advisor/SKILL.md b/crates/openshell-supervisor-process/src/skills/policy-advisor/SKILL.md index f05ba6a0be..81a9f598e1 100644 --- a/crates/openshell-supervisor-process/src/skills/policy-advisor/SKILL.md +++ b/crates/openshell-supervisor-process/src/skills/policy-advisor/SKILL.md @@ -6,3 +6,5 @@ description: Use when an OpenShell sandbox returns policy_denied, mentions polic # OpenShell Policy Advisor When a request fails with `policy_denied`, do not treat the denial as final if the user task still needs the request. Read `/etc/openshell/skills/policy_advisor.md`, inspect `http://policy.local/v1/policy/current`, submit the narrowest proposal to `http://policy.local/v1/proposals`, wait on `/v1/proposals/{chunk_id}/wait?timeout=300`, and retry only after approval with `policy_reloaded: true`. + +For a self-signed internal HTTPS service, an endpoint proposal may include `upstream_ca_pem` containing the service owner's PEM CA certificate bundle. Scope it to the exact host and port. This requires developer review even in auto mode. diff --git a/crates/openshell-supervisor-process/src/skills/policy_advisor.md b/crates/openshell-supervisor-process/src/skills/policy_advisor.md index 8d792aaedb..eb3d622563 100644 --- a/crates/openshell-supervisor-process/src/skills/policy_advisor.md +++ b/crates/openshell-supervisor-process/src/skills/policy_advisor.md @@ -40,6 +40,12 @@ The sandbox-local policy API is reachable at `http://policy.local`: The proposal body takes an `intent_summary` and one or more `addRule` operations. Each `addRule` carries a complete narrow `NetworkPolicyRule`. +For a self-signed internal HTTPS service, include `upstream_ca_pem` as a PEM +certificate bundle string on its exact host:port endpoint. Obtain the CA from +the service owner; never submit a private key. Keep normal TLS inspection +enabled. A supplemental CA always requires developer review, including in +auto mode. Retry only after approval and `policy_reloaded: true`. + ## Workflow 1. Read the denial response body. Use `layer`, `method`, `path`, `host`, diff --git a/docs/reference/policy-schema.mdx b/docs/reference/policy-schema.mdx index 29eafa60de..679c44f85b 100644 --- a/docs/reference/policy-schema.mdx +++ b/docs/reference/policy-schema.mdx @@ -165,6 +165,7 @@ Each endpoint defines a reachable destination and optional inspection rules. | `path` | string | No | Optional HTTP path glob used to select between L7 endpoints that share the same host and port. Empty means all paths. Use this when REST and GraphQL live under the same host, such as `/repos/**` and `/graphql`. | | `protocol` | string | No | Set to `tcp` with a valid DNS hostname to allow native TCP clients through policy DNS and transparent capture without payload inspection. Omit the field for L4 passthrough through an explicit proxy, including legacy hostless `allowed_ips` endpoints. Set to `rest` for HTTP method/path inspection, `websocket` for RFC 6455 upgrade and client text-message inspection, `graphql` for GraphQL-over-HTTP operation inspection, `mcp` for MCP Streamable HTTP request inspection, or `json-rpc` for generic JSON-RPC-over-HTTP method inspection. WebSocket endpoints can also use GraphQL operation rules for GraphQL-over-WebSocket traffic. Provider-credentialed endpoints require an inspected protocol unless `allow_uninspected_credentials` is explicitly set. | | `tls` | string | No | TLS handling mode. The proxy auto-detects TLS by peeking the first bytes of each connection and terminates it for inspected HTTPS traffic, so this field is optional in most cases. Set to `skip` to disable auto-detection for edge cases such as client-certificate mTLS or non-standard protocols. Provider-credentialed endpoints reject `tls: skip` unless `allow_uninspected_credentials` is explicitly set. The values `terminate` and `passthrough` are deprecated and log a warning; they are still accepted for backward compatibility but have no effect on behavior. | +| `upstream_ca_pem` | string | No | PEM-encoded CA certificate bundle trusted only for the supervisor's TLS connection to this endpoint. Use a YAML block scalar containing one or more `-----BEGIN CERTIFICATE-----` blocks. The supervisor adds these certificates to its normal system roots for this endpoint only; it still terminates client TLS, inspects requests, and rewrites credentials. Cannot be combined with `tls: skip` or `protocol: tcp`. Do not include private keys. | | `enforcement` | string | No | `enforce` actively blocks disallowed requests. `audit` logs violations but allows traffic through. | | `access` | string | No | Access preset. One of `read-only`, `read-write`, or `full`. Mutually exclusive with `rules`. Not valid on `protocol: mcp` or `protocol: json-rpc`; MCP uses explicit rules unless `mcp.allow_all_known_mcp_methods: true` enables the endpoint method profile, and JSON-RPC always uses explicit rules. | | `rules` | list of allow rule objects | No | Fine-grained protocol-specific allow rules. Mutually exclusive with `access`. | diff --git a/docs/sandboxes/policy-advisor.mdx b/docs/sandboxes/policy-advisor.mdx index 80884620d5..24ea08d43d 100644 --- a/docs/sandboxes/policy-advisor.mdx +++ b/docs/sandboxes/policy-advisor.mdx @@ -189,7 +189,9 @@ For REST APIs, prefer L7 rules over broad L4 access. A good proposal allows one } ``` -The current `policy.local` JSON shape covers explicit-proxy endpoints and REST method or path rules. Agent-authored proposals cannot set `protocol: tcp` or `tls: skip`, because those modes bypass application-authority inspection. When a task requires native TCP or a raw TLS tunnel, a developer must add the rule through the normal policy-authoring workflow. Omitting `protocol` remains supported and retains the explicit proxy's default TLS termination and HTTP authority checks. Use [Customize Sandbox Policies](/sandboxes/policies) or [Policy Schema Reference](/reference/policy-schema) for policy fields that are not part of the agent-authored proposal surface, such as WebSocket credential rewrite, GraphQL operation matching, endpoint path scoping, and provider-owned policy bundles. +The current `policy.local` JSON shape covers explicit-proxy endpoints and REST method or path rules. For a self-signed internal HTTPS service, an endpoint may also carry `"upstream_ca_pem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----\n"`: a JSON string containing one or more PEM CA certificates for that exact host and port. Get the CA bundle from the service owner; private keys, malformed bundles, and an explicitly empty value are rejected. Conflicting CA bundles on overlapping endpoints cannot be merged. This adds trust only to the selected endpoint's upstream TLS verifier; normal TLS inspection remains active. A CA-bearing proposal receives a security note and always waits for human review, even in auto mode. After approval, the sandbox hot-reloads the policy without a restart. + +Agent-authored proposals cannot set `protocol: tcp` or `tls: skip`, because those modes bypass application-authority inspection and cannot use a supplemental CA. When a task requires native TCP or a raw TLS tunnel, a developer must add the rule through the normal policy-authoring workflow. Omitting `protocol` remains supported and retains the explicit proxy's default TLS termination and HTTP authority checks. Use [Customize Sandbox Policies](/sandboxes/policies) or [Policy Schema Reference](/reference/policy-schema) for policy fields that are not part of the agent-authored proposal surface, such as WebSocket credential rewrite, GraphQL operation matching, endpoint path scoping, and provider-owned policy bundles. Policy advisor proposals do not add `allowed_ips` automatically. If an advisor-proposed hostname resolves to an internal or private address, OpenShell's SSRF protections still block the connection until a developer explicitly adds the required `allowed_ips` entry. diff --git a/proto/sandbox.proto b/proto/sandbox.proto index c2b61d0b3a..29e40c69f6 100644 --- a/proto/sandbox.proto +++ b/proto/sandbox.proto @@ -194,6 +194,9 @@ message NetworkEndpoint { // Internal gateway-derived marker indicating that this endpoint belongs to // an attached credentialed provider. User-authored values are ignored. bool provider_credentialed = 26; + // Additional PEM-encoded CA certificates trusted only when the supervisor + // re-encrypts TLS to this endpoint. Does not disable TLS inspection. + string upstream_ca_pem = 27; } // MCP options are grouped so MCP-specific policy can grow without adding more diff --git a/sdk/go/openshell/v1/internal/converter/coverage_test.go b/sdk/go/openshell/v1/internal/converter/coverage_test.go index 16929f672c..32b34eade3 100644 --- a/sdk/go/openshell/v1/internal/converter/coverage_test.go +++ b/sdk/go/openshell/v1/internal/converter/coverage_test.go @@ -180,6 +180,7 @@ func TestConverterCoversAllProtoFields_NetworkEndpoint(t *testing.T) { "ports": true, "protocol": true, "tls": true, + "upstream_ca_pem": true, "enforcement": true, "access": true, "rules": true, diff --git a/sdk/go/openshell/v1/internal/converter/network_policy.go b/sdk/go/openshell/v1/internal/converter/network_policy.go index 2451f6f426..f45d975772 100644 --- a/sdk/go/openshell/v1/internal/converter/network_policy.go +++ b/sdk/go/openshell/v1/internal/converter/network_policy.go @@ -68,6 +68,7 @@ func policyNetworkEndpointFromProto(ep *sbv1.NetworkEndpoint) types.PolicyNetwor Port: ep.GetPort(), Protocol: ep.GetProtocol(), TLS: ep.GetTls(), + UpstreamCAPEM: ep.GetUpstreamCaPem(), Enforcement: ep.GetEnforcement(), Access: ep.GetAccess(), AllowEncodedSlash: ep.GetAllowEncodedSlash(), @@ -130,6 +131,7 @@ func policyNetworkEndpointToProto(ep *types.PolicyNetworkEndpoint) *sbv1.Network Port: ep.Port, Protocol: ep.Protocol, Tls: ep.TLS, + UpstreamCaPem: ep.UpstreamCAPEM, Enforcement: ep.Enforcement, Access: ep.Access, AllowEncodedSlash: ep.AllowEncodedSlash, diff --git a/sdk/go/openshell/v1/internal/converter/network_policy_test.go b/sdk/go/openshell/v1/internal/converter/network_policy_test.go index 5f83a82645..0b93b9b89b 100644 --- a/sdk/go/openshell/v1/internal/converter/network_policy_test.go +++ b/sdk/go/openshell/v1/internal/converter/network_policy_test.go @@ -23,6 +23,7 @@ func TestNetworkPolicyRuleFromProto(t *testing.T) { Port: 443, Protocol: "rest", Tls: "strict", + UpstreamCaPem: "test-ca-pem", Enforcement: "enforce", Access: "allow", Ports: []uint32{80, 443}, @@ -102,6 +103,7 @@ func TestNetworkPolicyRuleFromProto(t *testing.T) { assert.Equal(t, uint32(443), ep.Port) assert.Equal(t, "rest", ep.Protocol) assert.Equal(t, "strict", ep.TLS) + assert.Equal(t, "test-ca-pem", ep.UpstreamCAPEM) assert.Equal(t, "enforce", ep.Enforcement) assert.Equal(t, "allow", ep.Access) assert.Equal(t, []uint32{80, 443}, ep.Ports) @@ -182,6 +184,7 @@ func TestNetworkPolicyRuleRoundTrip(t *testing.T) { Port: 8080, Protocol: "graphql", TLS: "permissive", + UpstreamCAPEM: "roundtrip-ca-pem", Enforcement: "audit", Access: "allow", Ports: []uint32{8080, 8443}, @@ -257,6 +260,7 @@ func TestNetworkPolicyRuleRoundTrip(t *testing.T) { assert.Equal(t, original.Endpoints[0].Port, roundTrip.Endpoints[0].Port) assert.Equal(t, original.Endpoints[0].Protocol, roundTrip.Endpoints[0].Protocol) assert.Equal(t, original.Endpoints[0].TLS, roundTrip.Endpoints[0].TLS) + assert.Equal(t, original.Endpoints[0].UpstreamCAPEM, roundTrip.Endpoints[0].UpstreamCAPEM) assert.Equal(t, original.Endpoints[0].Enforcement, roundTrip.Endpoints[0].Enforcement) assert.Equal(t, original.Endpoints[0].Access, roundTrip.Endpoints[0].Access) assert.Equal(t, original.Endpoints[0].Ports, roundTrip.Endpoints[0].Ports) diff --git a/sdk/go/openshell/v1/types/network_policy.go b/sdk/go/openshell/v1/types/network_policy.go index 69e44e1242..f0731b23e7 100644 --- a/sdk/go/openshell/v1/types/network_policy.go +++ b/sdk/go/openshell/v1/types/network_policy.go @@ -17,11 +17,13 @@ type NetworkPolicyRule struct { // as used in sandbox network policy rules. This is distinct from [NetworkEndpoint] // which is the simplified profile-level endpoint (Host, Port, Protocol only). type PolicyNetworkEndpoint struct { - Host string - Port uint32 - Ports []uint32 - Protocol string - TLS string + Host string + Port uint32 + Ports []uint32 + Protocol string + TLS string + // UpstreamCAPEM adds trust anchors only for this endpoint's upstream TLS connection. + UpstreamCAPEM string Enforcement string Access string Rules []L7Rule diff --git a/sdk/go/proto/sandboxv1/sandbox.pb.go b/sdk/go/proto/sandboxv1/sandbox.pb.go index 989589002b..7d62b58f7b 100644 --- a/sdk/go/proto/sandboxv1/sandbox.pb.go +++ b/sdk/go/proto/sandboxv1/sandbox.pb.go @@ -655,7 +655,7 @@ type NetworkEndpoint struct { // Endpoint protocol. "tcp" and "" select L4-only handling; "rest", // "websocket", "graphql", "sql", "json-rpc", and "mcp" select L7 inspection. Protocol string `protobuf:"bytes,3,opt,name=protocol,proto3" json:"protocol,omitempty"` - // TLS handling: "terminate" or "passthrough" (default). + // TLS handling: "auto" (default) or "skip" (raw tunnel). Tls string `protobuf:"bytes,4,opt,name=tls,proto3" json:"tls,omitempty"` // Enforcement mode: "enforce" or "audit" (default). Enforcement string `protobuf:"bytes,5,opt,name=enforcement,proto3" json:"enforcement,omitempty"` @@ -738,8 +738,11 @@ type NetworkEndpoint struct { // Internal gateway-derived marker indicating that this endpoint belongs to // an attached credentialed provider. User-authored values are ignored. ProviderCredentialed bool `protobuf:"varint,26,opt,name=provider_credentialed,json=providerCredentialed,proto3" json:"provider_credentialed,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Additional PEM-encoded CA certificates trusted only when the supervisor + // re-encrypts TLS to this endpoint. Does not disable TLS inspection. + UpstreamCaPem string `protobuf:"bytes,27,opt,name=upstream_ca_pem,json=upstreamCaPem,proto3" json:"upstream_ca_pem,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *NetworkEndpoint) Reset() { @@ -954,6 +957,13 @@ func (x *NetworkEndpoint) GetProviderCredentialed() bool { return false } +func (x *NetworkEndpoint) GetUpstreamCaPem() string { + if x != nil { + return x.UpstreamCaPem + } + return "" +} + // MCP options are grouped so MCP-specific policy can grow without adding more // top-level NetworkEndpoint fields. OpenShell owns the supported revision // profiles instead of treating dependency enums as the policy contract. @@ -2113,8 +2123,7 @@ const file_sandbox_proto_rawDesc = "" + "\ainclude\x18\x01 \x03(\tR\ainclude\x12\x18\n" + "\aexclude\x18\x02 \x03(\tR\aexclude\"6\n" + "\x18NetworkCredentialBinding\x12\x1a\n" + - "\bprovider\x18\x01 \x01(\tR\bprovider\"\xdc\n" + - "\n" + + "\bprovider\x18\x01 \x01(\tR\bprovider\"\x84\v\n" + "\x0fNetworkEndpoint\x12\x12\n" + "\x04host\x18\x01 \x01(\tR\x04host\x12\x12\n" + "\x04port\x18\x02 \x01(\rR\x04port\x12\x1a\n" + @@ -2144,7 +2153,8 @@ const file_sandbox_proto_rawDesc = "" + "\x03mcp\x18\x17 \x01(\v2 .openshell.sandbox.v1.McpOptionsR\x03mcp\x12]\n" + "\x12credential_binding\x18\x18 \x01(\v2..openshell.sandbox.v1.NetworkCredentialBindingR\x11credentialBinding\x12B\n" + "\x1dallow_uninspected_credentials\x18\x19 \x01(\bR\x1ballowUninspectedCredentials\x123\n" + - "\x15provider_credentialed\x18\x1a \x01(\bR\x14providerCredentialed\x1ar\n" + + "\x15provider_credentialed\x18\x1a \x01(\bR\x14providerCredentialed\x12&\n" + + "\x0fupstream_ca_pem\x18\x1b \x01(\tR\rupstreamCaPem\x1ar\n" + "\x1cGraphqlPersistedQueriesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12<\n" + "\x05value\x18\x02 \x01(\v2&.openshell.sandbox.v1.GraphqlOperationR\x05value:\x028\x01\"\xd2\x01\n" +