// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 //! HTTP CONNECT proxy with OPA policy evaluation and process-identity binding. use crate::identity::BinaryIdentityCache; use crate::l7::tls::ProxyTlsState; use crate::opa::{NetworkAction, OpaEngine, PolicyGenerationGuard}; use crate::policy_local::{POLICY_LOCAL_HOST, PolicyLocalContext}; use miette::{IntoDiagnostic, Result}; use openshell_core::activity::{ActivitySender, try_record_activity}; use openshell_core::denial::DenialEvent; use openshell_core::net::{is_always_blocked_ip, is_internal_ip, is_link_local_ip}; use openshell_core::policy::ProxyPolicy; use openshell_core::provider_credentials::ProviderCredentialState; use openshell_core::secrets::{self, SecretResolver, rewrite_header_line_checked}; use openshell_ocsf::{ ActionId, ActivityId, DispositionId, Endpoint, HttpActivityBuilder, HttpRequest, NetworkActivityBuilder, Process, SeverityId, StatusId, Url as OcsfUrl, ocsf_emit, }; use std::net::{IpAddr, SocketAddr}; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; use tokio::io::{ AsyncRead as TokioAsyncRead, AsyncReadExt, AsyncWrite as TokioAsyncWrite, AsyncWriteExt, }; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::mpsc; use tokio::task::JoinHandle; use tracing::{debug, warn}; const MAX_HEADER_BYTES: usize = 8192; const TUNNEL_PROTOCOL_PEEK_BYTES: usize = crate::l7::rest::HTTP2_PRIOR_KNOWLEDGE_PREFACE.len(); #[cfg(not(test))] const TUNNEL_PROTOCOL_PEEK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(100); #[cfg(test)] const TUNNEL_PROTOCOL_PEEK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(10); #[cfg(not(test))] const TUNNEL_PROTOCOL_PEEK_POLL: std::time::Duration = std::time::Duration::from_millis(5); #[cfg(test)] const TUNNEL_PROTOCOL_PEEK_POLL: std::time::Duration = std::time::Duration::from_millis(1); const INFERENCE_LOCAL_HOST: &str = "inference.local"; const INFERENCE_LOCAL_PORT: u16 = 443; #[cfg(target_os = "linux")] const SIDECAR_SUPERVISOR_TOPOLOGY: &str = "sidecar"; /// Hostnames injected by compute drivers as `/etc/hosts` aliases for the host /// machine. Traffic to these names is eligible for the trusted-gateway SSRF /// exemption when the resolved IP matches the driver-injected value read from /// `/etc/hosts` at proxy startup. const HOST_GATEWAY_ALIASES: &[&str] = &[ "host.openshell.internal", "host.containers.internal", "host.docker.internal", ]; /// Cloud instance metadata IPs that are NEVER exempted from SSRF blocking, /// even when they coincidentally match a host-gateway alias resolution. /// This list covers the well-known IMDS endpoints across major cloud providers. const CLOUD_METADATA_IPS: &[IpAddr] = &[ // AWS / GCP / Azure instance metadata service IpAddr::V4(std::net::Ipv4Addr::new(169, 254, 169, 254)), ]; /// Maximum total bytes for a streaming inference response body (32 MiB). #[cfg(not(test))] const MAX_STREAMING_BODY: usize = 32 * 1024 * 1024; // Keep unit tests deterministic without pushing tens of MiB through loopback. #[cfg(test)] const MAX_STREAMING_BODY: usize = 1024; /// Idle timeout per chunk when relaying streaming inference responses. /// /// Reasoning models (e.g. nemotron-3-super, o1, o3) can pause for 60+ seconds /// between "thinking" and output phases. 120s provides headroom while still /// catching genuinely stuck streams. #[cfg(not(test))] const CHUNK_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120); // Exercise idle-timeout truncation without slowing the full package test suite. #[cfg(test)] const CHUNK_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(100); /// Result of a proxy CONNECT policy decision. struct ConnectDecision { action: NetworkAction, /// Policy generation used for the L4 network decision. generation: u64, /// Resolved binary path. binary: Option, /// PID owning the socket. binary_pid: Option, /// Ancestor binary paths from process tree walk. ancestors: Vec, /// Cmdline-derived absolute paths (for script detection). cmdline_paths: Vec, } /// Outcome of an inference interception attempt. /// /// Returned by [`handle_inference_interception`] so the call site can emit /// a structured CONNECT deny log when the connection is not successfully routed. #[derive(Debug)] enum InferenceOutcome { /// At least one request was successfully routed to a local inference backend. Routed, /// The connection was denied (TLS failure, non-inference request, etc.). Denied { reason: String }, } /// Inference routing context for sandbox-local execution. /// /// Holds a `Router` (HTTP client) and cached sets of resolved routes. /// User routes serve `inference.local` traffic; system routes are consumed /// in-process by the supervisor for platform functions (e.g. agent harness). pub struct InferenceContext { pub patterns: Vec, router: openshell_router::Router, /// Routes for the user-facing `inference.local` endpoint. routes: Arc>>, /// Routes for supervisor-only system inference (`sandbox-system`). system_routes: Arc>>, } impl InferenceContext { // `router`/`routes` are intentionally distinct nouns (the router and the // route list it consumes); both names are clearer than alternatives. #[allow(clippy::similar_names)] pub fn new( patterns: Vec, router: openshell_router::Router, routes: Vec, system_routes: Vec, ) -> Self { Self { patterns, router, routes: Arc::new(tokio::sync::RwLock::new(routes)), system_routes: Arc::new(tokio::sync::RwLock::new(system_routes)), } } /// Get a handle to the user route cache for background refresh. pub fn route_cache( &self, ) -> Arc>> { self.routes.clone() } /// Get a handle to the system route cache for background refresh. pub fn system_route_cache( &self, ) -> Arc>> { self.system_routes.clone() } /// Make an inference call using system routes (supervisor-only). /// /// This is the in-process API for platform functions. It bypasses the /// CONNECT proxy entirely — the supervisor calls the router directly /// from the host network namespace. pub async fn system_inference( &self, protocol: &str, method: &str, path: &str, headers: Vec<(String, String)>, body: bytes::Bytes, ) -> Result { let routes = self.system_routes.read().await; self.router .proxy_with_candidates(protocol, method, path, headers, body, &routes) .await } } #[derive(Debug)] pub struct ProxyHandle { #[allow(dead_code)] http_addr: Option, join: JoinHandle<()>, } impl ProxyHandle { /// Start the proxy with OPA engine for policy evaluation. /// /// The proxy uses OPA for network decisions with process-identity binding /// via `/proc/net/tcp`. All connections are evaluated through OPA policy. #[allow(clippy::too_many_arguments)] pub(crate) async fn start_with_bind_addr( policy: &ProxyPolicy, bind_addr: Option, opa_engine: Arc, identity_cache: Arc, entrypoint_pid: Arc, tls_state: Option>, inference_ctx: Option>, provider_credentials: Option, policy_local_ctx: Option>, denial_tx: Option>, activity_tx: Option, engine_ready: tokio::sync::watch::Receiver, ) -> Result { // Use override bind_addr, fall back to policy http_addr, then default // to loopback:3128. The default allows the proxy to function when no // network namespace is available (e.g. missing CAP_NET_ADMIN) and the // policy doesn't specify an explicit address. let default_addr: SocketAddr = ([127, 0, 0, 1], 3128).into(); let http_addr = bind_addr.or(policy.http_addr).unwrap_or(default_addr); // Only enforce loopback restriction when not using network namespace override if bind_addr.is_none() && !http_addr.ip().is_loopback() { return Err(miette::miette!( "Proxy http_addr must be loopback-only: {http_addr}" )); } let listener = TcpListener::bind(http_addr).await.into_diagnostic()?; let local_addr = listener.local_addr().into_diagnostic()?; { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Listen) .severity(SeverityId::Informational) .status(StatusId::Success) .dst_endpoint(Endpoint::from_ip(local_addr.ip(), local_addr.port())) .message(format!("Proxy listening on {local_addr}")) .build(); ocsf_emit!(event); } // Detect the trusted host gateway IP from /etc/hosts before user code // runs. This is read once at startup so later /etc/hosts modifications // by sandbox workloads cannot influence the stored value. let trusted_host_gateway: Arc> = Arc::new(detect_trusted_host_gateway()); if let Some(ref ip) = *trusted_host_gateway { tracing::info!( %ip, "Trusted host gateway detected from /etc/hosts; \ host-gateway aliases exempt from SSRF always-blocked check" ); } let join = tokio::spawn(async move { // Wait for the OPA engine's symlink resolution reload to complete // before accepting connections. This prevents requests from // observing a generation transition mid-flight, which would cause // the generation guard to reject them with a 403. // // The TCP listener is already bound, so the OS backlog queues // incoming SYN packets during this wait. Once we start accepting, // queued connections drain immediately. let mut engine_ready = engine_ready; match tokio::time::timeout( std::time::Duration::from_secs(15), engine_ready.wait_for(|v| *v), ) .await { Ok(_) => {} Err(_) => { warn!( "Engine readiness signal not received within 15s; \ proceeding with proxy accept loop" ); } } loop { match listener.accept().await { Ok((stream, _addr)) => { let opa = opa_engine.clone(); let cache = identity_cache.clone(); let spid = entrypoint_pid.clone(); let tls = tls_state.clone(); let inf = inference_ctx.clone(); let policy_local = policy_local_ctx.clone(); let gw = trusted_host_gateway.clone(); let resolver = provider_credentials .as_ref() .and_then(ProviderCredentialState::resolver); let dynamic_credentials = provider_credentials.as_ref().map(|state| { Arc::new(std::sync::RwLock::new( state.snapshot().dynamic_credentials.clone(), )) }); let dtx = denial_tx.clone(); let atx = activity_tx.clone(); tokio::spawn(async move { #[allow(clippy::large_futures)] if let Err(err) = handle_tcp_connection( stream, opa, cache, spid, tls, inf, policy_local, gw, resolver, dynamic_credentials, dtx, atx, ) .await { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .message(format!("Proxy connection error: {err}")) .build(); ocsf_emit!(event); } }); } Err(err) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .message(format!("Proxy accept error: {err}")) .build(); ocsf_emit!(event); break; } } } }); Ok(Self { http_addr: Some(local_addr), join, }) } #[allow(dead_code)] pub const fn http_addr(&self) -> Option { self.http_addr } } impl Drop for ProxyHandle { fn drop(&mut self) { self.join.abort(); } } fn emit_activity(tx: &Option, denied: bool, deny_group: &'static str) { if let Some(tx) = tx { let _ = try_record_activity(tx, denied, deny_group); } } fn l7_inspection_active(l7_route: Option<&L7RouteSnapshot>) -> bool { l7_route.is_some_and(|route| !route.configs.is_empty()) } #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum TunnelProtocol { Tls, Http1, H2cPriorKnowledge, Unsupported, } fn classify_tunnel_protocol(peek: &[u8]) -> TunnelProtocol { if crate::l7::tls::looks_like_tls(peek) { return TunnelProtocol::Tls; } if crate::l7::rest::looks_like_http(peek) { return TunnelProtocol::Http1; } if crate::l7::rest::looks_like_http2_prior_knowledge(peek) { return TunnelProtocol::H2cPriorKnowledge; } TunnelProtocol::Unsupported } fn could_be_tls_prefix(peek: &[u8]) -> bool { matches!(peek, [0x16] | [0x16, 0x03]) } fn could_be_supported_tunnel_protocol_prefix(peek: &[u8]) -> bool { could_be_tls_prefix(peek) || crate::l7::rest::could_be_http_request_prefix(peek) || crate::l7::rest::could_be_http2_prior_knowledge_prefix(peek) } /// Why tunnel payload inspection is mandatory for a connection, in message /// precedence order: an L7-configured endpoint owns the wording even when a /// fail-closed middleware chain also matches. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum InspectionRequirement { None, L7Route, RequiredMiddleware, } fn inspection_requirement( should_inspect_l7: bool, middleware_gate: crate::l7::middleware::UninspectableTrafficGate, ) -> InspectionRequirement { if should_inspect_l7 { InspectionRequirement::L7Route } else if middleware_gate == crate::l7::middleware::UninspectableTrafficGate::Deny { InspectionRequirement::RequiredMiddleware } else { InspectionRequirement::None } } fn unsupported_l7_tunnel_protocol_detail( tunnel_protocol: TunnelProtocol, requirement: InspectionRequirement, ) -> Option<&'static str> { match (tunnel_protocol, requirement) { (_, InspectionRequirement::None) | (TunnelProtocol::Tls | TunnelProtocol::Http1, _) => None, (TunnelProtocol::H2cPriorKnowledge, InspectionRequirement::L7Route) => { Some("HTTP/2 prior-knowledge (h2c) is not supported for L7-inspected endpoints") } (TunnelProtocol::H2cPriorKnowledge, InspectionRequirement::RequiredMiddleware) => { Some("HTTP/2 prior-knowledge (h2c) cannot be inspected by required middleware") } (TunnelProtocol::Unsupported, InspectionRequirement::L7Route) => { Some("Unsupported tunnel protocol for L7-inspected endpoint") } (TunnelProtocol::Unsupported, InspectionRequirement::RequiredMiddleware) => { Some("Unsupported tunnel protocol cannot be inspected by required middleware") } } } /// Gate for traffic that would bypass L7 inspection entirely: query the /// middleware chain matching this destination and process identity, and /// decide whether raw relay is allowed. Uninspectable traffic is denied when /// any matching entry is `fail_closed`; an all-`fail_open` chain passes it /// through with a bypass detection finding. fn middleware_uninspectable_gate( opa_engine: &OpaEngine, ctx: &crate::l7::relay::L7EvalContext, ) -> Result { let input = crate::l7::middleware::middleware_network_input(ctx); let (chain, _generation) = opa_engine.query_middleware_chain_with_generation(&input)?; Ok(crate::l7::middleware::uninspectable_traffic_gate(&chain)) } async fn peek_tunnel_protocol(client: &TcpStream) -> Result> { let mut peek_buf = [0u8; TUNNEL_PROTOCOL_PEEK_BYTES]; let deadline = tokio::time::Instant::now() + TUNNEL_PROTOCOL_PEEK_TIMEOUT; loop { let n = client.peek(&mut peek_buf).await.into_diagnostic()?; if n == 0 { return Ok(None); } let peek = &peek_buf[..n]; let protocol = classify_tunnel_protocol(peek); if protocol != TunnelProtocol::Unsupported || !could_be_supported_tunnel_protocol_prefix(peek) || n == peek_buf.len() || tokio::time::Instant::now() >= deadline { return Ok(Some(protocol)); } tokio::time::sleep(TUNNEL_PROTOCOL_PEEK_POLL).await; } } fn emit_connect_activity_if_l4_only( tx: &Option, l7_route: Option<&L7RouteSnapshot>, ) { if !l7_inspection_active(l7_route) { emit_activity(tx, false, "unknown"); } } fn emit_activity_simple(tx: Option<&ActivitySender>, denied: bool, deny_group: &'static str) { if let Some(tx) = tx { let _ = try_record_activity(tx, denied, deny_group); } } fn emit_forward_success_activity(tx: Option<&ActivitySender>, l7_activity_pending: bool) { emit_activity_simple( tx, false, if l7_activity_pending { "l7_policy" } else { "unknown" }, ); } /// Body-aware policy state carried from forward L7 admission to middleware /// execution. The selected route and its captured policy engine stay paired so /// middleware cannot be invoked with only half of the re-evaluation context. struct ForwardL7Reevaluation<'a> { config: &'a crate::l7::L7EndpointConfig, engine: &'a crate::opa::TunnelPolicyEngine, request_info: &'a crate::l7::L7RequestInfo, } /// Executes the middleware portion of the forward HTTP pipeline with an /// explicit transformed-body policy. struct ForwardMiddlewarePipeline<'a> { ctx: &'a crate::l7::relay::L7EvalContext, scheme: &'a str, runner: &'a openshell_supervisor_middleware::ChainRunner, generation_guard: &'a PolicyGenerationGuard, l7_reevaluation: Option>, } impl ForwardMiddlewarePipeline<'_> { #[allow( clippy::option_if_let_else, reason = "the Some branch must keep a borrowed evaluator alive across the async call" )] async fn apply( &self, request: crate::l7::provider::L7Request, client: &mut C, chain: Vec, ) -> Result where C: TokioAsyncRead + TokioAsyncWrite + Unpin + Send, { let validate; let transformed_body_policy = match &self.l7_reevaluation { Some(l7) => { validate = crate::l7::relay::transformed_body_validator( l7.config, l7.engine, self.ctx, l7.request_info, ); openshell_supervisor_middleware::TransformedBodyPolicy::Reevaluate(&validate) } None => openshell_supervisor_middleware::TransformedBodyPolicy::NotPolicyRelevant, }; crate::l7::middleware::apply_middleware_chain_for_scheme( request, client, self.ctx, self.scheme, chain, self.runner, self.generation_guard, transformed_body_policy, ) .await } } /// Emit a denial event to the aggregator channel (if configured). /// Used by `handle_tcp_connection` which owns `Option`. fn emit_denial( tx: &Option>, host: &str, port: u16, binary: &str, decision: &ConnectDecision, reason: &str, stage: &str, ) { if let Some(tx) = tx { let _ = tx.send(DenialEvent { host: host.to_string(), port, binary: binary.to_string(), ancestors: decision .ancestors .iter() .map(|p| p.display().to_string()) .collect(), deny_reason: reason.to_string(), denial_stage: stage.to_string(), l7_method: None, l7_path: None, }); } } /// Emit a denial event from a borrowed sender reference. /// Used by `handle_forward_proxy` which borrows `Option<&Sender>`. fn emit_denial_simple( tx: Option<&mpsc::UnboundedSender>, host: &str, port: u16, binary: &str, decision: &ConnectDecision, reason: &str, stage: &str, ) { if let Some(tx) = tx { let _ = tx.send(DenialEvent { host: host.to_string(), port, binary: binary.to_string(), ancestors: decision .ancestors .iter() .map(|p| p.display().to_string()) .collect(), deny_reason: reason.to_string(), denial_stage: stage.to_string(), l7_method: None, l7_path: None, }); } } // Many distinct, non-related context parameters are required for a CONNECT // dispatch; bundling them into a struct would just shift the noise into call // sites. #[allow(clippy::too_many_arguments)] async fn handle_tcp_connection( mut client: TcpStream, opa_engine: Arc, identity_cache: Arc, entrypoint_pid: Arc, tls_state: Option>, inference_ctx: Option>, policy_local_ctx: Option>, trusted_host_gateway: Arc>, secret_resolver: Option>, dynamic_credentials: Option< Arc< std::sync::RwLock< std::collections::HashMap, >, >, >, denial_tx: Option>, activity_tx: Option, ) -> Result<()> { let mut buf = vec![0u8; MAX_HEADER_BYTES]; let mut used = 0usize; loop { if used == buf.len() { respond( &mut client, b"HTTP/1.1 431 Request Header Fields Too Large\r\n\r\n", ) .await?; return Ok(()); } let n = client.read(&mut buf[used..]).await.into_diagnostic()?; if n == 0 { return Ok(()); } used += n; if buf[..used].windows(4).any(|win| win == b"\r\n\r\n") { break; } } let header_end = buf[..used] .windows(4) .position(|window| window == b"\r\n\r\n") .expect("header terminator was observed") + 4; if crate::l7::rest::validate_http_request_header_block(&buf[..header_end]).is_err() { respond(&mut client, b"HTTP/1.1 400 Bad Request\r\n\r\n").await?; return Ok(()); } let request = std::str::from_utf8(&buf[..header_end]).expect("validated HTTP request headers are UTF-8"); if crate::l7::rest::parse_body_length(request).is_err() { respond(&mut client, b"HTTP/1.1 400 Bad Request\r\n\r\n").await?; return Ok(()); } let mut lines = request.split("\r\n"); let request_line = lines.next().unwrap_or(""); let mut parts = request_line.split_whitespace(); let method = parts.next().unwrap_or(""); let target = parts.next().unwrap_or(""); if method != "CONNECT" { return handle_forward_proxy( method, target, &buf[..], used, &mut client, opa_engine, identity_cache, entrypoint_pid, policy_local_ctx, trusted_host_gateway, secret_resolver, dynamic_credentials, denial_tx.as_ref(), activity_tx.as_ref(), ) .await; } let (host, port) = parse_target(target)?; let host_lc = host.to_ascii_lowercase(); if host_lc == INFERENCE_LOCAL_HOST && port == INFERENCE_LOCAL_PORT { respond(&mut client, b"HTTP/1.1 200 Connection Established\r\n\r\n").await?; let outcome = handle_inference_interception( client, INFERENCE_LOCAL_HOST, port, tls_state.as_ref(), inference_ctx.as_ref(), ) .await?; if let InferenceOutcome::Denied { reason } = outcome { emit_activity(&activity_tx, true, "forward_policy"); let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, port)) .message(format!("Inference interception denied: {reason}")) .status_detail(&reason) .build(); ocsf_emit!(event); } return Ok(()); } let peer_addr = client.peer_addr().into_diagnostic()?; let _local_addr = client.local_addr().into_diagnostic()?; // Evaluate OPA policy with process-identity binding. // Wrapped in spawn_blocking because identity resolution does heavy sync I/O: // /proc scanning + SHA256 hashing of binaries (e.g. node at 124MB). let opa_clone = opa_engine.clone(); let cache_clone = identity_cache.clone(); let pid_clone = entrypoint_pid.clone(); let host_clone = host_lc.clone(); let decision = tokio::task::spawn_blocking(move || { evaluate_opa_tcp( peer_addr, &opa_clone, &cache_clone, &pid_clone, &host_clone, port, ) }) .await .map_err(|e| miette::miette!("identity resolution task panicked: {e}"))?; // Extract action string and matched policy for logging let (matched_policy, deny_reason) = match &decision.action { NetworkAction::Allow { matched_policy } => (matched_policy.clone(), String::new()), NetworkAction::Deny { reason } => (None, reason.clone()), }; // Build log context fields (shared by deny log below and deferred allow log after L7 check) let binary_str = decision .binary .as_ref() .map_or_else(|| "-".to_string(), |p| p.display().to_string()); let pid_str = decision .binary_pid .map_or_else(|| "-".to_string(), |p| p.to_string()); let ancestors_str = if decision.ancestors.is_empty() { "-".to_string() } else { decision .ancestors .iter() .map(|p| p.display().to_string()) .collect::>() .join(" -> ") }; let cmdline_str = if decision.cmdline_paths.is_empty() { "-".to_string() } else { decision .cmdline_paths .iter() .map(|p| p.display().to_string()) .collect::>() .join(", ") }; let policy_str = matched_policy.as_deref().unwrap_or("-"); // Log denied connections immediately — they never reach L7. // Allowed connections are logged after the L7 config check (below) // so we can distinguish CONNECT (L4-only) from CONNECT_L7 (L7 follows). if matches!(decision.action, NetworkAction::Deny { .. }) { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "opa") .message(format!("CONNECT denied {host_lc}:{port}")) .status_detail(&deny_reason) .build(); ocsf_emit!(event); emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, &deny_reason, "connect", ); emit_activity(&activity_tx, true, "connect_policy"); respond( &mut client, &build_json_error_response_with_reason( 403, "Forbidden", "policy_denied", &format!("CONNECT {host_lc}:{port} not permitted by policy"), &deny_reason, ), ) .await?; return Ok(()); } // Resolve the route's TLS treatment up front. `query_tls_mode` reads only // the policy decision + host/port (no peeked bytes), so it is valid before // the `200`. The fail-closed refusal that consumes it runs after the SSRF/ // allowed_ips validation below — so an internal-address CONNECT still gets // the SSRF 403 and telemetry in degraded state — but before the upstream // connect and before `200 Connection Established`. let effective_tls_skip = query_tls_mode(&opa_engine, &decision, &host_lc, port) == crate::l7::TlsMode::Skip; let sandbox_entrypoint_pid = entrypoint_pid.load(Ordering::Acquire); // Query allowed_ips from the matched endpoint config (if any). // When present, the SSRF check validates resolved IPs against this // allowlist instead of blanket-blocking all private IPs. // When the policy host is already a literal IP address, treat it as // implicitly allowed — the user explicitly declared the destination. // Exact declared hostnames also skip the private-IP blanket block below, // while keeping loopback/link-local/unspecified addresses denied. let mut raw_allowed_ips = query_allowed_ips(&opa_engine, &decision, &host_lc, port); if raw_allowed_ips.is_empty() { raw_allowed_ips = implicit_allowed_ips_for_ip_host(&host); } let exact_declared_endpoint_host = query_exact_declared_endpoint_host(&opa_engine, &decision, &host_lc, port); // Defense-in-depth: resolve DNS and reject connections to internal IPs. let dns_connect_start = std::time::Instant::now(); // The "non-empty" branch is the explicit-allowlist path; reading it first // matches the policy decision narrative. #[allow(clippy::if_not_else)] let validated_addrs = if is_host_gateway_alias(&host_lc) && let Some(gw) = *trusted_host_gateway { // Trusted host-gateway path. The compute driver injected this hostname // into /etc/hosts pointing at a known IP (read at proxy startup before // user code runs). Bypass the normal SSRF tiers so link-local gateway // addresses (used by rootless Podman with pasta) are not hard-blocked. // Cloud metadata IPs and control-plane ports are still rejected. match resolve_and_check_trusted_gateway(&host, port, gw, sandbox_entrypoint_pid).await { Ok(addrs) => addrs, Err(reason) => { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "ssrf") .message(format!( "CONNECT blocked: trusted-gateway check failed for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity(&activity_tx, true, "ssrf"); respond( &mut client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!("CONNECT {host_lc}:{port} blocked: trusted-gateway check failed"), ), ) .await?; return Ok(()); } } } else if !raw_allowed_ips.is_empty() { // allowed_ips mode: validate resolved IPs against CIDR allowlist. // Loopback and link-local are still always blocked. match parse_allowed_ips(&raw_allowed_ips) { Ok(nets) => { match resolve_and_check_allowed_ips(&host, port, &nets, sandbox_entrypoint_pid) .await { Ok(addrs) => addrs, Err(reason) => { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "ssrf") .message(format!( "CONNECT blocked: allowed_ips check failed for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity(&activity_tx, true, "ssrf"); respond( &mut client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!( "CONNECT {host_lc}:{port} blocked: allowed_ips check failed" ), ), ) .await?; return Ok(()); } } } Err(reason) => { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "ssrf") .message(format!( "CONNECT blocked: invalid allowed_ips in policy for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity(&activity_tx, true, "ssrf"); respond( &mut client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!("CONNECT {host_lc}:{port} blocked: invalid allowed_ips in policy"), ), ) .await?; return Ok(()); } } } else if exact_declared_endpoint_host { // Exact declared hostname mode: the operator explicitly allowed this // host:port, so private IP resolution is permitted without duplicating // the resolved IP in allowed_ips. Always-blocked addresses and // control-plane ports remain denied. match resolve_and_check_declared_endpoint(&host, port, sandbox_entrypoint_pid).await { Ok(addrs) => addrs, Err(reason) => { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "ssrf") .message(format!( "CONNECT blocked: declared endpoint check failed for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); respond( &mut client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!( "CONNECT {host_lc}:{port} blocked: declared endpoint check failed" ), ), ) .await?; return Ok(()); } } } else { // Default: reject all internal IPs (loopback, RFC 1918, link-local). match resolve_and_reject_internal(&host, port, sandbox_entrypoint_pid).await { Ok(addrs) => addrs, Err(reason) => { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "ssrf") .message(format!( "CONNECT blocked: internal address {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity(&activity_tx, true, "ssrf"); respond( &mut client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!("CONNECT {host_lc}:{port} blocked: internal address"), ), ) .await?; return Ok(()); } } }; // Fail closed AFTER SSRF/allowed_ips validation (an internal-address CONNECT // has already returned the SSRF 403 above) but BEFORE connecting upstream // and BEFORE `200 Connection Established`. A terminating route with no TLS // termination state cannot rewrite credential placeholders; the 503 must be // the first bytes on the socket rather than a post-200 in-tunnel write the // client would misread as a TLS error. No "allowed CONNECT" event is emitted // for this path — that log lives after the `200` below. if refuse_connect_when_tls_unavailable(&mut client, tls_state.is_some(), effective_tls_skip) .await? { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::High) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "tls") .message(format!( "CONNECT refused for {host_lc}:{port}: {TLS_TERMINATION_UNAVAILABLE_DETAIL}" )) .status_detail(TLS_TERMINATION_UNAVAILABLE_DETAIL) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx.as_ref(), true, "tls_termination_unavailable"); emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, TLS_TERMINATION_UNAVAILABLE_DETAIL, "connect-tls-termination-unavailable", ); return Ok(()); } let mut upstream = TcpStream::connect(validated_addrs.as_slice()) .await .into_diagnostic()?; debug!( "handle_tcp_connection dns_resolve_and_tcp_connect: {}ms host={host_lc}", dns_connect_start.elapsed().as_millis() ); respond(&mut client, b"HTTP/1.1 200 Connection Established\r\n\r\n").await?; // Check if endpoint has L7 config for protocol-aware inspection, and // retain the generation for HTTP passthrough keep-alive tunnels. let l7_route = query_l7_route_snapshot(&opa_engine, &decision, &host_lc, port); let should_inspect_l7 = l7_inspection_active(l7_route.as_ref()); // Log the allowed CONNECT — use CONNECT_L7 when L7 inspection follows, // so log consumers can distinguish L4-only decisions from tunnel lifecycle events. let connect_msg = if should_inspect_l7 { "CONNECT_L7" } else { "CONNECT" }; { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Allowed) .disposition(DispositionId::Allowed) .severity(SeverityId::Informational) .status(StatusId::Success) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "opa") .message(format!("{connect_msg} allowed {host_lc}:{port}")) .build(); ocsf_emit!(event); } emit_connect_activity_if_l4_only(&activity_tx, l7_route.as_ref()); // `effective_tls_skip` was resolved before the `200` above (the fail-closed // gate needs it) and drives the raw-tunnel branch below. // Build L7 eval context (shared by TLS-terminated and plaintext paths). let ctx = crate::l7::relay::L7EvalContext { host: host_lc.clone(), port, policy_name: matched_policy.clone().unwrap_or_default(), binary_path: decision .binary .as_ref() .map(|p| p.to_string_lossy().into_owned()) .unwrap_or_default(), ancestors: decision .ancestors .iter() .map(|p| p.to_string_lossy().into_owned()) .collect(), cmdline_paths: decision .cmdline_paths .iter() .map(|p| p.to_string_lossy().into_owned()) .collect(), secret_resolver: secret_resolver.clone(), activity_tx: activity_tx.clone(), dynamic_credentials: dynamic_credentials.clone(), token_grant_resolver: dynamic_credentials .as_ref() .map(|_| crate::l7::token_grant_injection::default_resolver()), }; if effective_tls_skip { // Policy validation rejects fail-closed middleware overlapping // `tls: skip` endpoints; this runtime gate is defense in depth. match middleware_uninspectable_gate(&opa_engine, &ctx)? { crate::l7::middleware::UninspectableTrafficGate::Deny => { crate::l7::middleware::emit_middleware_uninspectable(&ctx, "tls-skip tunnel", true); respond( &mut client, &build_json_error_response( 403, "Forbidden", "middleware_required", "tls: skip tunnel cannot be inspected by required middleware", ), ) .await?; return Ok(()); } crate::l7::middleware::UninspectableTrafficGate::BypassWithFinding => { crate::l7::middleware::emit_middleware_uninspectable( &ctx, "tls-skip tunnel", false, ); } crate::l7::middleware::UninspectableTrafficGate::Unrestricted => {} } // tls: skip — raw tunnel, no termination, no credential injection. debug!( host = %host_lc, port = port, "tls: skip — bypassing TLS auto-detection, raw tunnel" ); let _ = tokio::io::copy_bidirectional(&mut client, &mut upstream) .await .into_diagnostic()?; return Ok(()); } // Auto-detect the tunnel payload. L7-configured endpoints must only // enter relays that can enforce their configured protocol; unsupported // bytes fail closed below instead of falling through to raw relay. let Some(tunnel_protocol) = peek_tunnel_protocol(&client).await? else { return Ok(()); }; if tunnel_protocol == TunnelProtocol::Tls { // TLS detected — terminate unconditionally. if let Some(ref tls) = tls_state { let tls_result = async { let mut tls_client = crate::l7::tls::tls_terminate_client(client, tls, &host_lc).await?; let mut tls_upstream = crate::l7::tls::tls_connect_upstream(upstream, &host_lc, tls.upstream_config()) .await?; if let Some(route) = l7_route.as_ref().filter(|route| !route.configs.is_empty()) { // L7 inspection on terminated TLS traffic. let tunnel_engine = match opa_engine.clone_engine_for_tunnel(route.generation) { Ok(engine) => engine, Err(e) => { emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); return Ok(()); } }; if route.configs.len() == 1 { crate::l7::relay::relay_with_inspection( &route.configs[0].config, tunnel_engine, &mut tls_client, &mut tls_upstream, &ctx, ) .await } else { let configs: Vec = route .configs .iter() .map(|snapshot| snapshot.config.clone()) .collect(); crate::l7::relay::relay_with_route_selection( &configs, tunnel_engine, &mut tls_client, &mut tls_upstream, &ctx, ) .await } } else { // No L7 config — relay with credential injection only. let generation = l7_route .as_ref() .map_or(decision.generation, |route| route.generation); let generation_guard = match opa_engine.generation_guard(generation) { Ok(guard) => guard, Err(e) => { emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); return Ok(()); } }; crate::l7::relay::relay_passthrough_with_credentials( &mut tls_client, &mut tls_upstream, &ctx, &generation_guard, Some(&opa_engine), ) .await } }; if let Err(e) = tls_result.await { if is_benign_relay_error(&e) { debug!( host = %host_lc, port = port, error = %e, "TLS connection closed" ); } else { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!("TLS relay error: {e}")) .build(); ocsf_emit!(event); } } } else { // Defense in depth; unreachable in normal operation. The pre-200 // fail-closed gate already refuses terminating routes when no TLS // termination state exists, and `tls: skip` routes raw-tunnel // before this peek. Reaching here means a future refactor bypassed // that gate. The `200 Connection Established` was already sent, so // the tunnel is live: any HTTP bytes now would be decoded as a TLS // protocol error, so we fail closed by DROPPING the connection // rather than raw-tunneling (which would forward the client's TLS // stream upstream and leak any `openshell:resolve:env:*` // placeholder verbatim). const DETAIL: &str = "TLS termination unavailable after tunnel establishment; \ closing connection - credential rewrite would be bypassed"; let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::High) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "tls") .message(format!("CONNECT refused for {host_lc}:{port}: {DETAIL}")) .status_detail(DETAIL) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx.as_ref(), true, "tls_termination_unavailable"); emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, DETAIL, "connect-tls-termination-unavailable", ); // No HTTP response: the tunnel is already established, so writing // bytes here would corrupt the client's TLS handshake. Drop instead. return Ok(()); } } else if tunnel_protocol == TunnelProtocol::Http1 { // Plaintext HTTP detected. if let Some(route) = l7_route.as_ref().filter(|route| !route.configs.is_empty()) { let tunnel_engine = match opa_engine.clone_engine_for_tunnel(route.generation) { Ok(engine) => engine, Err(e) => { emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); return Ok(()); } }; let relay_result = if route.configs.len() == 1 { crate::l7::relay::relay_with_inspection( &route.configs[0].config, tunnel_engine, &mut client, &mut upstream, &ctx, ) .await } else { let configs: Vec = route .configs .iter() .map(|snapshot| snapshot.config.clone()) .collect(); crate::l7::relay::relay_with_route_selection( &configs, tunnel_engine, &mut client, &mut upstream, &ctx, ) .await }; if let Err(e) = relay_result { if is_benign_relay_error(&e) { debug!(host = %host_lc, port = port, error = %e, "L7 connection closed"); } else { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!("L7 relay error: {e}")) .build(); ocsf_emit!(event); } } } else { // Plaintext HTTP, no L7 config — relay with credential injection. let generation = l7_route .as_ref() .map_or(decision.generation, |route| route.generation); let generation_guard = match opa_engine.generation_guard(generation) { Ok(guard) => guard, Err(e) => { emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); return Ok(()); } }; if let Err(e) = crate::l7::relay::relay_passthrough_with_credentials( &mut client, &mut upstream, &ctx, &generation_guard, Some(&opa_engine), ) .await { if is_benign_relay_error(&e) { debug!(host = %host_lc, port = port, error = %e, "HTTP relay closed"); } else { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!("HTTP relay error: {e}")) .build(); ocsf_emit!(event); } } } } else { let middleware_gate = middleware_uninspectable_gate(&opa_engine, &ctx)?; let requirement = inspection_requirement(should_inspect_l7, middleware_gate); if let Some(protocol_detail) = unsupported_l7_tunnel_protocol_detail(tunnel_protocol, requirement) { if requirement == InspectionRequirement::RequiredMiddleware { crate::l7::middleware::emit_middleware_uninspectable(&ctx, protocol_detail, true); } let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "l7") .message(format!( "CONNECT_L7 blocked unsupported tunnel protocol for {host_lc}:{port}" )) .status_detail(protocol_detail) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx.as_ref(), true, "l7_parse_rejection"); emit_denial( &denial_tx, &host_lc, port, &binary_str, &decision, protocol_detail, "connect-l7-parse-rejection", ); respond( &mut client, &build_json_error_response( 403, "Forbidden", "unsupported_l7_protocol", protocol_detail, ), ) .await?; return Ok(()); } if middleware_gate == crate::l7::middleware::UninspectableTrafficGate::BypassWithFinding { crate::l7::middleware::emit_middleware_uninspectable(&ctx, "non-http tcp", false); } // Neither TLS nor HTTP — raw binary relay. debug!( host = %host_lc, port = port, "Non-TLS non-HTTP traffic detected, raw tunnel" ); let _ = tokio::io::copy_bidirectional(&mut client, &mut upstream) .await .into_diagnostic()?; } Ok(()) } /// Resolved process identity for a TCP peer: binary path, PID, ancestor chain, /// cmdline paths, and the TOFU-verified binary hash. /// /// Produced by [`resolve_process_identity`]; consumed by [`evaluate_opa_tcp`] /// and by the identity-chain regression tests. #[cfg(target_os = "linux")] struct ResolvedIdentity { bin_path: PathBuf, binary_pid: u32, ancestors: Vec, cmdline_paths: Vec, bin_hash: String, } #[cfg(target_os = "linux")] #[derive(Debug, Eq, PartialEq)] struct PolicyIdentityKey { bin_path: PathBuf, ancestors: Vec, cmdline_paths: Vec, bin_hash: String, } #[cfg(target_os = "linux")] impl ResolvedIdentity { fn policy_key(&self) -> PolicyIdentityKey { PolicyIdentityKey { bin_path: self.bin_path.clone(), ancestors: self.ancestors.clone(), cmdline_paths: self.cmdline_paths.clone(), bin_hash: self.bin_hash.clone(), } } } /// Error from [`resolve_process_identity`]. Carries the deny reason and /// whatever partial identity data was resolved before the failure so the /// caller can include it in the [`ConnectDecision`] and OCSF event. #[cfg(target_os = "linux")] struct IdentityError { reason: String, binary: Option, binary_pid: Option, ancestors: Vec, } #[cfg(target_os = "linux")] fn resolve_owner_identity( owner_pid: u32, entrypoint_pid: u32, identity_cache: &BinaryIdentityCache, ) -> std::result::Result { let bin_path = crate::procfs::binary_path(owner_pid.cast_signed()).map_err(|e| IdentityError { reason: format!("failed to resolve peer binary for PID {owner_pid}: {e}"), binary: None, binary_pid: Some(owner_pid), ancestors: vec![], })?; let bin_hash = identity_cache .verify_or_cache_process_exe(&bin_path, owner_pid) .map_err(|e| IdentityError { reason: format!("binary integrity check failed: {e}"), binary: Some(bin_path.clone()), binary_pid: Some(owner_pid), ancestors: vec![], })?; let ancestor_identities = collect_ancestor_identities(owner_pid, entrypoint_pid); let ancestors: Vec = ancestor_identities .iter() .map(|(_, path)| path.clone()) .collect(); for (ancestor_pid, ancestor) in &ancestor_identities { identity_cache .verify_or_cache_process_exe(ancestor, *ancestor_pid) .map_err(|e| IdentityError { reason: format!( "ancestor integrity check failed for {}: {e}", ancestor.display() ), binary: Some(bin_path.clone()), binary_pid: Some(owner_pid), ancestors: ancestors.clone(), })?; } let mut exclude = ancestors.clone(); exclude.push(bin_path.clone()); let cmdline_paths = crate::procfs::collect_cmdline_paths(owner_pid, entrypoint_pid, &exclude); Ok(ResolvedIdentity { bin_path, binary_pid: owner_pid, ancestors, cmdline_paths, bin_hash, }) } #[cfg(target_os = "linux")] fn collect_ancestor_identities(start_pid: u32, stop_pid: u32) -> Vec<(u32, PathBuf)> { const MAX_DEPTH: usize = 64; // When the socket owner IS the entrypoint, there are no intermediate // ancestors to verify — the chain is empty by definition. if start_pid == stop_pid { return vec![]; } let mut ancestors = Vec::new(); let mut current = start_pid; for _ in 0..MAX_DEPTH { let parent_pid = match crate::procfs::read_ppid(current) { Some(parent) if parent > 0 && parent != current => parent, _ => break, }; if let Ok(path) = crate::procfs::binary_path(parent_pid.cast_signed()) { ancestors.push((parent_pid, path)); } if parent_pid == stop_pid || parent_pid == 1 { break; } current = parent_pid; } ancestors } /// Resolve the identity of the process owning a TCP peer connection. /// /// Walks `/proc//net/tcp` to find the socket inode, locates /// every owning PID, reads `/proc//exe`, TOFU-verifies each binary hash, /// walks each ancestor chain verifying every ancestor, and collects /// cmdline-derived absolute paths for script detection. /// /// This is the identity-resolution block of [`evaluate_opa_tcp`] extracted /// into a standalone helper so it can be exercised by Linux-only regression /// tests without a full OPA engine. The key hot-swap invariant under test is /// that display paths are stripped for policy/logging, while integrity hashing /// reads the live executable via `/proc//exe` instead of the replacement /// file that now exists at the display path. #[cfg(target_os = "linux")] fn resolve_process_identity( entrypoint_pid: u32, peer_port: u16, identity_cache: &BinaryIdentityCache, ) -> std::result::Result { let socket_owners = crate::procfs::resolve_tcp_peer_socket_owners(entrypoint_pid, peer_port) .map_err(|e| IdentityError { reason: format!("failed to resolve peer binary: {e}"), binary: None, binary_pid: None, ancestors: vec![], })?; let mut identities = Vec::with_capacity(socket_owners.owners.len()); for owner in &socket_owners.owners { identities.push(resolve_owner_identity( owner.pid, entrypoint_pid, identity_cache, )?); } let Some(first_identity) = identities.first() else { return Err(IdentityError { reason: format!( "failed to resolve peer binary: no process found owning socket inode {}", socket_owners.inode ), binary: None, binary_pid: None, ancestors: vec![], }); }; let first_key = first_identity.policy_key(); if identities .iter() .skip(1) .any(|identity| identity.policy_key() != first_key) { let mut pids: Vec = identities .iter() .map(|identity| identity.binary_pid) .collect(); pids.sort_unstable(); return Err(IdentityError { reason: format!( "ambiguous shared socket ownership: inode {} is held by PIDs [{}] with different policy identities", socket_owners.inode, pids.iter() .map(u32::to_string) .collect::>() .join(", ") ), binary: None, binary_pid: None, ancestors: vec![], }); } let mut identity = identities.swap_remove(0); if let Some(lowest_pid) = socket_owners.owners.iter().map(|owner| owner.pid).min() { identity.binary_pid = lowest_pid; } Ok(identity) } /// Evaluate OPA policy for a TCP connection with identity binding via /proc/net/tcp. #[cfg(target_os = "linux")] fn evaluate_opa_tcp( peer_addr: SocketAddr, engine: &OpaEngine, identity_cache: &BinaryIdentityCache, entrypoint_pid: &AtomicU32, host: &str, port: u16, ) -> ConnectDecision { use crate::opa::NetworkInput; use std::sync::atomic::Ordering; let deny = |reason: String, binary: Option, binary_pid: Option, ancestors: Vec, cmdline_paths: Vec| -> ConnectDecision { ConnectDecision { action: NetworkAction::Deny { reason }, generation: engine.current_generation(), binary, binary_pid, ancestors, cmdline_paths, } }; if !crate::opa::network_binary_identity_required() { let result = evaluate_endpoint_only_opa(engine, host, port); debug!( "evaluate_opa_tcp endpoint-only: host={host} port={port} action={:?}", result.action ); return result; } let entrypoint_pid = entrypoint_pid.load(Ordering::Acquire); let Some(proc_net_anchor_pid) = proc_net_anchor_pid(entrypoint_pid) else { return deny( "entrypoint process not yet spawned".into(), None, None, vec![], vec![], ); }; let total_start = std::time::Instant::now(); let peer_port = peer_addr.port(); let identity = match resolve_process_identity(proc_net_anchor_pid, peer_port, identity_cache) { Ok(id) => id, Err(err) => { return deny( err.reason, err.binary, err.binary_pid, err.ancestors, vec![], ); } }; let ResolvedIdentity { bin_path, binary_pid, ancestors, cmdline_paths, bin_hash, } = identity; let input = NetworkInput { host: host.to_string(), port, binary_path: bin_path.clone(), binary_sha256: bin_hash, ancestors: ancestors.clone(), cmdline_paths: cmdline_paths.clone(), }; let result = match engine.evaluate_network_action_with_generation(&input) { Ok((action, generation)) => ConnectDecision { action, generation, binary: Some(bin_path), binary_pid: Some(binary_pid), ancestors, cmdline_paths, }, Err(e) => deny( format!("policy evaluation error: {e}"), Some(bin_path), Some(binary_pid), ancestors, cmdline_paths, ), }; debug!( "evaluate_opa_tcp TOTAL: {}ms host={host} port={port}", total_start.elapsed().as_millis() ); result } #[cfg(target_os = "linux")] fn proc_net_anchor_pid(entrypoint_pid: u32) -> Option { if entrypoint_pid != 0 { return Some(entrypoint_pid); } sidecar_topology_enabled().then(std::process::id) } #[cfg(target_os = "linux")] fn sidecar_topology_enabled() -> bool { std::env::var(openshell_core::sandbox_env::SUPERVISOR_TOPOLOGY) .is_ok_and(|value| value == SIDECAR_SUPERVISOR_TOPOLOGY) } fn evaluate_endpoint_only_opa(engine: &OpaEngine, host: &str, port: u16) -> ConnectDecision { let input = crate::opa::NetworkInput { host: host.to_string(), port, binary_path: PathBuf::new(), binary_sha256: String::new(), ancestors: vec![], cmdline_paths: vec![], }; match engine.evaluate_network_action_with_generation(&input) { Ok((action, generation)) => ConnectDecision { action, generation, binary: None, binary_pid: None, ancestors: vec![], cmdline_paths: vec![], }, Err(e) => ConnectDecision { action: NetworkAction::Deny { reason: format!("policy evaluation error: {e}"), }, generation: engine.current_generation(), binary: None, binary_pid: None, ancestors: vec![], cmdline_paths: vec![], }, } } /// Non-Linux stub: OPA identity binding requires /proc. #[cfg(not(target_os = "linux"))] fn evaluate_opa_tcp( _peer_addr: SocketAddr, engine: &OpaEngine, _identity_cache: &BinaryIdentityCache, _entrypoint_pid: &AtomicU32, host: &str, port: u16, ) -> ConnectDecision { if !crate::opa::network_binary_identity_required() { return evaluate_endpoint_only_opa(engine, host, port); } ConnectDecision { action: NetworkAction::Deny { reason: "identity binding unavailable on this platform".into(), }, generation: engine.current_generation(), binary: None, binary_pid: None, ancestors: vec![], cmdline_paths: vec![], } } /// Maximum buffer size for inference request parsing (10 MiB). const MAX_INFERENCE_BUF: usize = 10 * 1024 * 1024; /// Initial buffer size for inference request parsing (64 KiB). const INITIAL_INFERENCE_BUF: usize = 65536; /// Handle an intercepted connection for inference routing. /// /// TLS-terminates the client connection, parses HTTP requests, and executes /// inference API calls locally via `openshell-router`. /// Non-inference requests are denied with 403. /// /// Returns [`InferenceOutcome::Routed`] if at least one request was successfully /// routed, or [`InferenceOutcome::Denied`] with a reason for all denial cases. async fn handle_inference_interception( client: TcpStream, host: &str, port: u16, tls_state: Option<&Arc>, inference_ctx: Option<&Arc>, ) -> Result { let Some(ctx) = inference_ctx else { return Ok(InferenceOutcome::Denied { reason: "cluster inference context not configured".to_string(), }); }; let Some(tls) = tls_state else { return Ok(InferenceOutcome::Denied { reason: "missing TLS state".to_string(), }); }; // TLS-terminate the client side (present a cert for the target host) let mut tls_client = match crate::l7::tls::tls_terminate_client(client, tls, host).await { Ok(c) => c, Err(e) => { return Ok(InferenceOutcome::Denied { reason: format!("TLS handshake failed: {e}"), }); } }; process_inference_keepalive(&mut tls_client, ctx, port).await } /// Read and process HTTP requests from a TLS-terminated inference connection. /// /// Each request is matched against inference patterns and routed locally. /// Any non-inference request is immediately denied and the connection is closed, /// even if previous requests on the same keep-alive connection were routed /// successfully. async fn process_inference_keepalive( stream: &mut S, ctx: &InferenceContext, port: u16, ) -> Result { use crate::l7::inference::{ParseResult, format_http_response, try_parse_http_request}; let mut buf = vec![0u8; INITIAL_INFERENCE_BUF]; let mut used = 0usize; let mut routed_any = false; loop { let n = match stream.read(&mut buf[used..]).await { Ok(n) => n, Err(e) => { if routed_any { break; } return Ok(InferenceOutcome::Denied { reason: format!("I/O error: {e}"), }); } }; if n == 0 { if routed_any { break; } return Ok(InferenceOutcome::Denied { reason: "client closed connection".to_string(), }); } used += n; // Try to parse a complete HTTP request match try_parse_http_request(&buf[..used]) { ParseResult::Complete(request, consumed) => { let was_routed = route_inference_request(&request, ctx, stream).await?; if was_routed { routed_any = true; } else { // Deny and close: a non-inference request must not be silently // ignored on a keep-alive connection that previously routed // inference traffic. return Ok(InferenceOutcome::Denied { reason: "connection not allowed by policy".to_string(), }); } // Shift buffer for next request buf.copy_within(consumed..used, 0); used -= consumed; } ParseResult::Incomplete => { // Need more data — grow buffer if full if used == buf.len() { if buf.len() >= MAX_INFERENCE_BUF { let response = format_http_response(413, &[], b"Payload Too Large"); write_all(stream, &response).await?; if routed_any { break; } return Ok(InferenceOutcome::Denied { reason: "payload too large".to_string(), }); } buf.resize((buf.len() * 2).min(MAX_INFERENCE_BUF), 0); } } ParseResult::Invalid(reason) => { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Refuse) .action(ActionId::Denied) .disposition(DispositionId::Rejected) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, port)) .message(format!("Rejecting malformed inference request: {reason}")) .status_detail(&reason) .build(); ocsf_emit!(event); } let response = format_http_response(400, &[], b"Bad Request"); write_all(stream, &response).await?; return Ok(InferenceOutcome::Denied { reason }); } } } Ok(InferenceOutcome::Routed) } /// Route a parsed inference request locally via the sandbox router, or deny it. /// /// Returns `Ok(true)` if the request was routed to an inference backend, /// `Ok(false)` if it was denied as a non-inference request. async fn route_inference_request( request: &crate::l7::inference::ParsedHttpRequest, ctx: &InferenceContext, tls_client: &mut (impl tokio::io::AsyncWrite + Unpin), ) -> Result { use crate::l7::inference::{detect_inference_pattern, format_http_response}; let normalized_path = normalize_inference_path(&request.path); if let Some(pattern) = detect_inference_pattern(&request.method, &normalized_path, &ctx.patterns) { { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Allowed) .disposition(DispositionId::Detected) .severity(SeverityId::Informational) .status(StatusId::Success) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, 443)) .message(format!( "Intercepted inference request, routing locally: {} {} (protocol={}, kind={})", request.method, normalized_path, pattern.protocol, pattern.kind )) .build(); ocsf_emit!(event); } let routes = ctx.routes.read().await; if routes.is_empty() { let body = serde_json::json!({ "error": "cluster inference is not configured", "hint": "run: openshell cluster inference set --help" }); let body_bytes = body.to_string(); let response = format_http_response( 503, &[("content-type".to_string(), "application/json".to_string())], body_bytes.as_bytes(), ); write_all(tls_client, &response).await?; return Ok(true); } // Buffered protocols (embeddings, model discovery) return a single JSON // object, not an SSE token stream. Serve them buffered with an accurate // Content-Length: the streaming path would append an SSE error frame to // the body on a size-cap or idle-timeout truncation, corrupting a // payload the client parses as one JSON object. Framing is declared per // protocol on the matched pattern. if pattern.is_buffered() { match ctx .router .proxy_with_candidates( &pattern.protocol, &request.method, &normalized_path, request.headers.clone(), bytes::Bytes::from(request.body.clone()), &routes, ) .await { Ok(resp) => { let resp_headers = sanitize_inference_response_headers(resp.headers); let response = format_http_response(resp.status, &resp_headers, &resp.body); write_all(tls_client, &response).await?; } Err(e) => write_inference_router_error(tls_client, &e).await?, } return Ok(true); } match ctx .router .proxy_with_candidates_streaming( &pattern.protocol, &request.method, &normalized_path, request.headers.clone(), bytes::Bytes::from(request.body.clone()), &routes, ) .await { Ok(mut resp) => { use crate::l7::inference::{ format_chunk, format_chunk_terminator, format_http_response_header, format_sse_error, }; let resp_headers = sanitize_inference_response_headers( std::mem::take(&mut resp.headers).into_iter().collect(), ); // Write response headers immediately (chunked TE). let header_bytes = format_http_response_header(resp.status, &resp_headers); write_all(tls_client, &header_bytes).await?; // Stream body chunks with byte cap and idle timeout. // // Each upstream chunk is wrapped in HTTP chunked framing and // flushed immediately so SSE events reach the client without // delay. Unlike the previous per-byte write_all+flush, we // coalesce the framing header + data + trailer into a single // write_all call, reducing the number of TLS records per chunk // from 3 to 1 while preserving incremental delivery. let mut total_bytes: usize = 0; loop { match tokio::time::timeout(CHUNK_IDLE_TIMEOUT, resp.next_chunk()).await { Ok(Ok(Some(chunk))) => { total_bytes += chunk.len(); if total_bytes > MAX_STREAMING_BODY { warn!( total_bytes = total_bytes, limit = MAX_STREAMING_BODY, "streaming response exceeded byte limit, truncating" ); let err = format_sse_error( "response truncated: exceeded maximum streaming body size", ); let _ = write_all(tls_client, &format_chunk(&err)).await; break; } let encoded = format_chunk(&chunk); write_all(tls_client, &encoded).await?; } Ok(Ok(None)) => break, Ok(Err(e)) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, 443)) .message(format!( "error reading upstream response chunk after \ {total_bytes} bytes: {e}" )) .build(); ocsf_emit!(event); let err = format_sse_error("response truncated: upstream read error"); let _ = write_all(tls_client, &format_chunk(&err)).await; break; } Err(_) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, 443)) .message(format!( "streaming response chunk idle timeout after \ {total_bytes} bytes, closing" )) .build(); ocsf_emit!(event); let err = format_sse_error("response truncated: chunk idle timeout exceeded"); let _ = write_all(tls_client, &format_chunk(&err)).await; break; } } } // Terminate the chunked stream. write_all(tls_client, format_chunk_terminator()).await?; } Err(e) => write_inference_router_error(tls_client, &e).await?, } Ok(true) } else { // Not an inference request — deny { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, 443)) .message(format!( "connection not allowed by policy: {} {}", request.method, normalized_path )) .build(); ocsf_emit!(event); } let body = serde_json::json!({"error": "connection not allowed by policy"}); let body_bytes = body.to_string(); let response = format_http_response( 403, &[("content-type".to_string(), "application/json".to_string())], body_bytes.as_bytes(), ); write_all(tls_client, &response).await?; Ok(false) } } /// Emit an OCSF failure event and write a buffered JSON error response for a /// router error hit while proxying an inference request. /// /// Shared by the streaming and buffered routing paths so both surface upstream /// failures with the same status mapping and the same audit record. async fn write_inference_router_error( tls_client: &mut (impl tokio::io::AsyncWrite + Unpin), err: &openshell_router::RouterError, ) -> Result<()> { use crate::l7::inference::format_http_response; let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(INFERENCE_LOCAL_HOST, 443)) .message(format!( "inference endpoint detected but upstream service failed: {err}" )) .build(); ocsf_emit!(event); let (status, msg) = router_error_to_http(err); let body = serde_json::json!({ "error": msg }).to_string(); let response = format_http_response( status, &[("content-type".to_string(), "application/json".to_string())], body.as_bytes(), ); write_all(tls_client, &response).await } /// Map router errors to HTTP status codes and sanitized messages. /// /// Returns generic, client-safe messages instead of verbatim internal details; /// the full error is recorded in the OCSF failure event by the caller. fn router_error_to_http(err: &openshell_router::RouterError) -> (u16, String) { use openshell_router::RouterError; match err { RouterError::RouteNotFound(_) => (400, "no inference route configured".to_string()), RouterError::NoCompatibleRoute(_) => { (400, "no compatible inference route available".to_string()) } RouterError::Unauthorized(_) => (401, "unauthorized".to_string()), RouterError::UpstreamUnavailable(_) => (503, "inference service unavailable".to_string()), RouterError::UpstreamProtocol(_) | RouterError::Internal(_) => { (502, "inference service error".to_string()) } } } fn sanitize_inference_response_headers(headers: Vec<(String, String)>) -> Vec<(String, String)> { headers .into_iter() .filter(|(name, _)| !should_strip_response_header(name)) .collect() } fn should_strip_response_header(name: &str) -> bool { let name_lc = name.to_ascii_lowercase(); matches!(name_lc.as_str(), "content-length") || is_hop_by_hop_header(&name_lc) } fn is_hop_by_hop_header(name: &str) -> bool { matches!( name, "connection" | "keep-alive" | "proxy-authenticate" | "proxy-authorization" | "proxy-connection" | "te" | "trailer" | "transfer-encoding" | "upgrade" ) } /// Write all bytes to an async writer. async fn write_all(writer: &mut (impl tokio::io::AsyncWrite + Unpin), data: &[u8]) -> Result<()> { use tokio::io::AsyncWriteExt; writer.write_all(data).await.into_diagnostic()?; writer.flush().await.into_diagnostic()?; Ok(()) } #[derive(Debug, Clone)] struct L7ConfigSnapshot { config: crate::l7::L7EndpointConfig, } #[derive(Debug, Clone)] struct L7RouteSnapshot { configs: Vec, generation: u64, } fn emit_l7_tunnel_close_after_policy_change(host: &str, port: u16, error: miette::Report) { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Open) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(host, port)) .message(format!( "L7 tunnel closed before inspection because policy changed: {error}" )) .build(); ocsf_emit!(event); } /// Query L7 endpoint config from the OPA engine for a matched CONNECT decision. /// /// Returns `Some(L7EndpointConfig)` if the matched endpoint has L7 config (protocol field), /// `None` for L4-only endpoints. fn query_l7_route_snapshot( engine: &OpaEngine, decision: &ConnectDecision, host: &str, port: u16, ) -> Option { // Only query if action is Allow (not Deny) let has_policy = match &decision.action { NetworkAction::Allow { matched_policy } => matched_policy.is_some(), NetworkAction::Deny { .. } => false, }; if !has_policy { return None; } let input = crate::opa::NetworkInput { host: host.to_string(), port, binary_path: decision.binary.clone().unwrap_or_default(), binary_sha256: String::new(), ancestors: decision.ancestors.clone(), cmdline_paths: decision.cmdline_paths.clone(), }; match engine.query_endpoint_configs_with_generation(&input) { Ok((vals, generation)) => { let configs: Vec<_> = vals .into_iter() .filter_map(|val| crate::l7::parse_l7_config(&val)) .map(|config| L7ConfigSnapshot { config }) .collect(); debug!( host, port, generation, config_count = configs.len(), "Forward proxy L7 route lookup complete" ); Some(L7RouteSnapshot { configs, generation, }) } Err(e) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(host, port)) .message(format!("Failed to query L7 endpoint config: {e}")) .build(); ocsf_emit!(event); None } } } fn select_l7_config_for_path<'a>( configs: &'a [L7ConfigSnapshot], path: &str, ) -> Option<&'a L7ConfigSnapshot> { configs .iter() .filter(|snapshot| snapshot.config.matches_path(path)) .max_by_key(|snapshot| snapshot.config.path_specificity()) } /// Query the TLS mode for an endpoint, independent of L7 config. /// /// This extracts `tls: skip` from the endpoint even when no `protocol` is set. fn query_tls_mode( engine: &OpaEngine, decision: &ConnectDecision, host: &str, port: u16, ) -> crate::l7::TlsMode { let has_policy = match &decision.action { NetworkAction::Allow { matched_policy } => matched_policy.is_some(), NetworkAction::Deny { .. } => false, }; if !has_policy { return crate::l7::TlsMode::Auto; } let input = crate::opa::NetworkInput { host: host.to_string(), port, binary_path: decision.binary.clone().unwrap_or_default(), binary_sha256: String::new(), ancestors: decision.ancestors.clone(), cmdline_paths: decision.cmdline_paths.clone(), }; match engine.query_endpoint_config(&input) { Ok(Some(val)) => crate::l7::parse_tls_mode(&val), _ => crate::l7::TlsMode::Auto, } } /// When the policy endpoint host is a literal IP address, the user has /// explicitly declared intent to allow that destination. Synthesize an /// `allowed_ips` entry so the existing allowlist-validation path is used /// instead of the blanket internal-IP rejection. /// /// Always-blocked addresses (loopback, link-local, unspecified) are skipped /// — synthesizing an `allowed_ips` entry for them would be silently /// un-enforceable at runtime. fn implicit_allowed_ips_for_ip_host(host: &str) -> Vec { let lookup_host = normalize_host_lookup_key(host); if let Ok(ip) = lookup_host.parse::() { if is_always_blocked_ip(ip) { warn!( host, "Policy host is an always-blocked address; \ implicit allowed_ips skipped — SSRF hardening prevents \ traffic to this destination regardless of policy" ); return vec![]; } vec![lookup_host.to_string()] } else { vec![] } } fn normalize_host_lookup_key(host: &str) -> &str { host.strip_prefix('[') .and_then(|trimmed| trimmed.strip_suffix(']')) .unwrap_or(host) } /// Returns `true` if `host` is one of the well-known driver-injected aliases /// for the host machine (e.g. `host.openshell.internal`). fn is_host_gateway_alias(host: &str) -> bool { let h = normalize_host_lookup_key(host); HOST_GATEWAY_ALIASES .iter() .any(|alias| alias.eq_ignore_ascii_case(h)) } /// Returns `true` if `ip` is a known cloud instance metadata endpoint that /// must never be exempted from SSRF blocking. /// /// IPv4-mapped IPv6 addresses (e.g. `::ffff:169.254.169.254`) are normalized /// to their embedded IPv4 representation before comparison, so the invariant /// holds regardless of how the address is represented. fn is_cloud_metadata_ip(ip: IpAddr) -> bool { match ip { IpAddr::V4(_) => CLOUD_METADATA_IPS.contains(&ip), IpAddr::V6(v6) => v6 .to_ipv4_mapped() .is_some_and(|v4| CLOUD_METADATA_IPS.contains(&IpAddr::V4(v4))), } } /// Read the proxy's own `/etc/hosts` at startup and return the IP mapped to /// `host.openshell.internal`, if present and safe. /// /// This is called once before user code runs, so the returned value is immune /// to later `/etc/hosts` tampering by sandbox workloads. Returns `None` if no /// entry exists, the entry cannot be parsed, or the mapped IP is a cloud /// metadata address. #[cfg(any(target_os = "linux", test))] fn detect_trusted_host_gateway() -> Option { let contents = std::fs::read_to_string("/etc/hosts").ok()?; let ips = parse_hosts_file_for_host(&contents, "host.openshell.internal"); // Multiple distinct IPs for the alias is unexpected — compute drivers // always inject exactly one. Warn loudly so operators can diagnose the // inconsistency; we still proceed with the first entry rather than // disabling the exemption entirely, because the mismatch guard in // resolve_and_check_trusted_gateway() will reject any runtime resolution // that returns a different IP. if ips.len() > 1 { warn!( ips = ?ips, "host.openshell.internal has {} distinct IPs in /etc/hosts; \ expected exactly one. Using first entry. \ Connections resolving to any other IP will be rejected.", ips.len() ); } let ip = ips.into_iter().next()?; if is_cloud_metadata_ip(ip) { warn!( %ip, "host.openshell.internal resolves to a cloud metadata IP; \ trusted-gateway SSRF exemption disabled" ); return None; } // The exemption exists solely for link-local IPs used by rootless Podman // with pasta. Private RFC 1918 addresses (e.g. Docker bridge 172.17.0.1, // Kubernetes node 192.168.x.x), loopback, unspecified, and all other // non-link-local addresses are never legitimate candidates for the // link-local SSRF exemption — they must fall through to the normal // allowed_ips / resolve_and_reject_internal() enforcement path. if !is_link_local_ip(ip) { warn!( %ip, "host.openshell.internal maps to a non-link-local IP; \ trusted-gateway SSRF exemption disabled" ); return None; } Some(ip) } #[cfg(not(any(target_os = "linux", test)))] fn detect_trusted_host_gateway() -> Option { None } /// Resolve `host:port` and validate that every resolved address matches the /// trusted host gateway IP. /// /// This bypasses the normal SSRF tiers (always-blocked and internal-IP) for /// driver-injected host-gateway aliases, allowing link-local addresses used /// by rootless Podman with pasta without opening up arbitrary link-local or /// cloud metadata access. /// /// Rejects: /// - Any resolved IP that is a cloud metadata address (defense-in-depth) /// - Any resolved IP that does not match `trusted_gw` (prevents /etc/hosts tampering) /// - Control-plane ports (etcd, K8s API, kubelet) regardless of IP async fn resolve_and_check_trusted_gateway( host: &str, port: u16, trusted_gw: IpAddr, entrypoint_pid: u32, ) -> std::result::Result, String> { if BLOCKED_CONTROL_PLANE_PORTS.contains(&port) { return Err(format!( "port {port} is a blocked control-plane port, connection rejected" )); } let addrs = resolve_socket_addrs(host, port, entrypoint_pid).await?; if addrs.is_empty() { return Err(format!( "DNS resolution returned no addresses for {}", normalize_host_lookup_key(host) )); } for addr in &addrs { if is_cloud_metadata_ip(addr.ip()) { return Err(format!( "{host} resolves to cloud metadata address {}, connection rejected", addr.ip() )); } if addr.ip() != trusted_gw { return Err(format!( "{host} resolves to {} which does not match trusted host gateway \ {trusted_gw}, connection rejected", addr.ip() )); } // Defense-in-depth: even if the resolved IP matches trusted_gw, reject // any non-link-local address. detect_trusted_host_gateway() already // enforces this at startup, but we re-check here to guard against any // unanticipated code path that might admit a private or loopback IP. if !is_link_local_ip(addr.ip()) { return Err(format!( "{host} resolves to non-link-local address {}, \ connection rejected", addr.ip() )); } } Ok(addrs) } fn resolve_ip_literal(host: &str, port: u16) -> Option> { normalize_host_lookup_key(host) .parse::() .ok() .map(|ip| vec![SocketAddr::new(ip, port)]) } #[cfg(any(target_os = "linux", test))] fn parse_hosts_file_for_host(contents: &str, host: &str) -> Vec { let lookup_host = normalize_host_lookup_key(host); let mut addrs = Vec::new(); for raw_line in contents.lines() { let line = raw_line.split('#').next().unwrap_or("").trim(); if line.is_empty() { continue; } let mut fields = line.split_whitespace(); let Some(ip_str) = fields.next() else { continue; }; let Ok(ip) = ip_str.parse::() else { continue; }; if fields.any(|alias| alias.eq_ignore_ascii_case(lookup_host)) && !addrs.contains(&ip) { addrs.push(ip); } } addrs } #[cfg(any(target_os = "linux", test))] fn resolve_from_hosts_file_contents(contents: &str, host: &str, port: u16) -> Vec { parse_hosts_file_for_host(contents, host) .into_iter() .map(|ip| SocketAddr::new(ip, port)) .collect() } #[cfg(target_os = "linux")] async fn resolve_from_sandbox_hosts( host: &str, port: u16, entrypoint_pid: u32, ) -> Option> { if entrypoint_pid == 0 { return None; } let hosts_path = format!("/proc/{entrypoint_pid}/root/etc/hosts"); let contents = match tokio::fs::read_to_string(&hosts_path).await { Ok(contents) => contents, Err(error) => { debug!( pid = entrypoint_pid, path = %hosts_path, host, "Falling back to DNS; failed to read sandbox hosts file: {error}" ); return None; } }; let addrs = resolve_from_hosts_file_contents(&contents, host, port); if addrs.is_empty() { None } else { Some(addrs) } } // Mirrors the Linux signature so call sites can `.await` uniformly across // platforms; the non-Linux path has nothing to await. #[cfg(not(target_os = "linux"))] #[allow(clippy::unused_async)] async fn resolve_from_sandbox_hosts( _host: &str, _port: u16, _entrypoint_pid: u32, ) -> Option> { None } async fn resolve_socket_addrs( host: &str, port: u16, entrypoint_pid: u32, ) -> std::result::Result, String> { if let Some(addrs) = resolve_ip_literal(host, port) { return Ok(addrs); } if let Some(addrs) = resolve_from_sandbox_hosts(host, port, entrypoint_pid).await { return Ok(addrs); } let lookup_host = normalize_host_lookup_key(host); let addrs: Vec = tokio::net::lookup_host((lookup_host, port)) .await .map_err(|e| format!("DNS resolution failed for {lookup_host}:{port}: {e}"))? .collect(); if addrs.is_empty() { return Err(format!( "DNS resolution returned no addresses for {lookup_host}:{port}" )); } Ok(addrs) } fn reject_internal_resolved_addrs( host: &str, addrs: &[SocketAddr], ) -> std::result::Result<(), String> { if addrs.is_empty() { return Err(format!( "DNS resolution returned no addresses for {}", normalize_host_lookup_key(host) )); } for addr in addrs { if is_internal_ip(addr.ip()) { return Err(format!( "{host} resolves to internal address {}, connection rejected", addr.ip() )); } } Ok(()) } fn validate_allowed_ips_for_resolved_addrs( host: &str, port: u16, addrs: &[SocketAddr], allowed_ips: &[ipnet::IpNet], ) -> std::result::Result<(), String> { if addrs.is_empty() { return Err(format!( "DNS resolution returned no addresses for {}", normalize_host_lookup_key(host) )); } // Block control-plane ports regardless of IP match. if BLOCKED_CONTROL_PLANE_PORTS.contains(&port) { return Err(format!( "port {port} is a blocked control-plane port, connection rejected" )); } for addr in addrs { // Always block loopback and link-local if is_always_blocked_ip(addr.ip()) { return Err(format!( "{host} resolves to always-blocked address {}, connection rejected", addr.ip() )); } // Check resolved IP against the allowlist let ip_allowed = allowed_ips.iter().any(|net| net.contains(&addr.ip())); if !ip_allowed { return Err(format!( "{host} resolves to {} which is not in allowed_ips, connection rejected", addr.ip() )); } } Ok(()) } fn validate_declared_endpoint_resolved_addrs( host: &str, port: u16, addrs: &[SocketAddr], ) -> std::result::Result<(), String> { if addrs.is_empty() { return Err(format!( "DNS resolution returned no addresses for {}", normalize_host_lookup_key(host) )); } if BLOCKED_CONTROL_PLANE_PORTS.contains(&port) { return Err(format!( "port {port} is a blocked control-plane port, connection rejected" )); } for addr in addrs { if is_always_blocked_ip(addr.ip()) { return Err(format!( "{host} resolves to always-blocked address {}, connection rejected", addr.ip() )); } } Ok(()) } /// Resolve a host:port using sandbox `/etc/hosts` first (when available), then /// reject if any resolved address is internal. /// /// Returns the resolved `SocketAddr` list on success. Returns an error string /// if any resolved IP is in an internal range or if DNS resolution fails. async fn resolve_and_reject_internal( host: &str, port: u16, entrypoint_pid: u32, ) -> std::result::Result, String> { let addrs = resolve_socket_addrs(host, port, entrypoint_pid).await?; reject_internal_resolved_addrs(host, &addrs)?; Ok(addrs) } /// Resolve a host:port using sandbox `/etc/hosts` first (when available), then /// validate resolved addresses against a CIDR/IP allowlist. /// /// Rejects loopback and link-local unconditionally. For all other resolved /// addresses, checks that each one matches at least one entry in `allowed_ips`. /// Entries can be CIDR notation ("10.0.5.0/24") or exact IPs ("10.0.5.20"). /// /// Returns the resolved `SocketAddr` list on success. async fn resolve_and_check_allowed_ips( host: &str, port: u16, allowed_ips: &[ipnet::IpNet], entrypoint_pid: u32, ) -> std::result::Result, String> { let addrs = resolve_socket_addrs(host, port, entrypoint_pid).await?; validate_allowed_ips_for_resolved_addrs(host, port, &addrs, allowed_ips)?; Ok(addrs) } /// Resolve a host:port that was explicitly declared by hostname in policy. /// /// Exact declared hostnames are the operator's trust signal, so RFC1918 and /// other private ranges are allowed without a duplicated `allowed_ips` entry. /// Loopback, link-local, unspecified, and control-plane ports remain blocked. async fn resolve_and_check_declared_endpoint( host: &str, port: u16, entrypoint_pid: u32, ) -> std::result::Result, String> { let addrs = resolve_socket_addrs(host, port, entrypoint_pid).await?; validate_declared_endpoint_resolved_addrs(host, port, &addrs)?; Ok(addrs) } /// Minimum CIDR prefix length before logging a breadth warning. /// CIDRs broader than /16 (65,536+ addresses) may unintentionally expose /// control-plane services on the same network. const MIN_SAFE_PREFIX_LEN: u8 = 16; /// Ports that are always blocked in `resolve_and_check_allowed_ips`, even /// when the resolved IP matches an `allowed_ips` entry. These ports belong /// to control-plane services that should never be reachable from a sandbox. const BLOCKED_CONTROL_PLANE_PORTS: &[u16] = &[ 2379, // etcd client 2380, // etcd peer 6443, // Kubernetes API server 10250, // kubelet API 10255, // kubelet read-only ]; /// Parse CIDR/IP strings into `IpNet` values, rejecting invalid entries and /// entries that overlap always-blocked ranges (loopback, link-local, /// unspecified). /// /// Returns parsed networks on success, or an error describing which entries /// are invalid or always-blocked. Logs a warning for overly broad CIDRs /// that are not outright blocked. fn parse_allowed_ips(raw: &[String]) -> std::result::Result, String> { use openshell_core::net::is_always_blocked_net; let mut nets = Vec::with_capacity(raw.len()); let mut errors = Vec::new(); for entry in raw { // Try as CIDR first, then as bare IP (convert to /32 or /128) let parsed = entry.parse::().or_else(|_| { entry .parse::() .map(|ip| match ip { IpAddr::V4(v4) => ipnet::IpNet::V4(ipnet::Ipv4Net::from(v4)), IpAddr::V6(v6) => ipnet::IpNet::V6(ipnet::Ipv6Net::from(v6)), }) .map_err(|_| ()) }); match parsed { Ok(n) => { // Reject entries that overlap always-blocked ranges — these // would be silently denied at runtime by is_always_blocked_ip // and cause confusing UX (accepted in policy, never works). if is_always_blocked_net(n) { errors.push(format!( "allowed_ips entry {entry} falls within always-blocked range \ (loopback/link-local/unspecified); remove this entry — \ SSRF hardening prevents traffic to these destinations \ regardless of policy" )); continue; } if n.prefix_len() < MIN_SAFE_PREFIX_LEN { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .severity(SeverityId::Medium) .message(format!( "allowed_ips entry has a very broad CIDR {n} (/{}) < /{MIN_SAFE_PREFIX_LEN}; \ this may expose control-plane services on the same network", n.prefix_len() )) .build(); ocsf_emit!(event); } nets.push(n); } Err(()) => errors.push(format!("invalid CIDR/IP in allowed_ips: {entry}")), } } if errors.is_empty() { Ok(nets) } else { Err(errors.join("; ")) } } /// Query `allowed_ips` from the matched endpoint config for a CONNECT decision. fn query_allowed_ips( engine: &OpaEngine, decision: &ConnectDecision, host: &str, port: u16, ) -> Vec { // Only query if action is Allow with a matched policy let has_policy = match &decision.action { NetworkAction::Allow { matched_policy } => matched_policy.is_some(), NetworkAction::Deny { .. } => false, }; if !has_policy { return vec![]; } let input = crate::opa::NetworkInput { host: host.to_string(), port, binary_path: decision.binary.clone().unwrap_or_default(), binary_sha256: String::new(), ancestors: decision.ancestors.clone(), cmdline_paths: decision.cmdline_paths.clone(), }; match engine.query_allowed_ips(&input) { Ok(ips) => ips, Err(e) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(host, port)) .message(format!( "Failed to query allowed_ips from endpoint config: {e}" )) .build(); ocsf_emit!(event); vec![] } } } /// Query whether the matched endpoint was declared as this exact hostname. fn query_exact_declared_endpoint_host( engine: &OpaEngine, decision: &ConnectDecision, host: &str, port: u16, ) -> bool { let has_policy = match &decision.action { NetworkAction::Allow { matched_policy } => matched_policy.is_some(), NetworkAction::Deny { .. } => false, }; if !has_policy { return false; } let input = crate::opa::NetworkInput { host: host.to_string(), port, binary_path: decision.binary.clone().unwrap_or_default(), binary_sha256: String::new(), ancestors: decision.ancestors.clone(), cmdline_paths: decision.cmdline_paths.clone(), }; match engine.query_exact_declared_endpoint_host(&input) { Ok(is_exact_declared) => is_exact_declared, Err(e) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(host, port)) .message(format!("Failed to query exact declared endpoint host: {e}")) .build(); ocsf_emit!(event); false } } } /// Canonicalize the request-target for inference pattern detection. /// /// Falls back to the raw path on canonicalization error: the request is then /// routed through the normal forward path, where `rest.rs::parse_http_request` /// will reject it properly. Returning the raw path here prevents a crafted /// target from bypassing inference routing without our detection logic having /// to implement a second, duplicate error-response surface. fn normalize_inference_path(path: &str) -> String { match crate::l7::path::canonicalize_request_target( path, &crate::l7::path::CanonicalizeOptions::default(), ) { Ok((canon, _)) => canon.path, Err(_) => path.to_string(), } } /// Extract the hostname from an absolute-form URI used in plain HTTP proxy requests. /// /// For example, `"http://example.com/path"` yields `"example.com"` and /// `"http://example.com:8080/path"` yields `"example.com"`. Returns `"unknown"` /// if the URI cannot be parsed. #[cfg(test)] fn extract_host_from_uri(uri: &str) -> String { // Absolute-form URIs look like "http://host[:port]/path" // Strip the scheme prefix, then extract the authority (host[:port]) before the first '/'. let after_scheme = uri.find("://").map_or(uri, |i| &uri[i + 3..]); let authority = after_scheme.split('/').next().unwrap_or(after_scheme); // Strip port if present (handle IPv6 bracket notation) let host = if authority.starts_with('[') { // IPv6: [::1]:port authority.find(']').map_or(authority, |i| &authority[..=i]) } else { authority.split(':').next().unwrap_or(authority) }; if host.is_empty() { "unknown".to_string() } else { host.to_string() } } /// Parse an absolute-form proxy request URI into its components. /// /// For example, `"http://10.86.8.223:8000/screenshot/"` yields /// `("http", "10.86.8.223", 8000, "/screenshot/")`. /// /// Handles: /// - Default port 80 for `http`, 443 for `https` /// - IPv6 bracket notation (`[::1]`) /// - Missing path (defaults to `/`) /// - Query strings (preserved in path) fn parse_proxy_uri(uri: &str) -> Result<(String, String, u16, String)> { // Extract scheme let (scheme, rest) = uri .split_once("://") .ok_or_else(|| miette::miette!("Missing scheme in proxy URI: {uri}"))?; let scheme = scheme.to_ascii_lowercase(); // Split authority from path let (authority, path) = if rest.starts_with('[') { // IPv6: [::1]:port/path let bracket_end = rest .find(']') .ok_or_else(|| miette::miette!("Unclosed IPv6 bracket in URI: {uri}"))?; let after_bracket = &rest[bracket_end + 1..]; after_bracket.find('/').map_or((rest, "/"), |slash_pos| { ( &rest[..=bracket_end + slash_pos], &after_bracket[slash_pos..], ) }) } else if let Some(slash_pos) = rest.find('/') { (&rest[..slash_pos], &rest[slash_pos..]) } else { (rest, "/") }; // Parse host and port from authority let (host, port) = if authority.starts_with('[') { // IPv6: [::1]:port or [::1] let bracket_end = authority .find(']') .ok_or_else(|| miette::miette!("Unclosed IPv6 bracket: {uri}"))?; let host = &authority[1..bracket_end]; // strip brackets let port_str = &authority[bracket_end + 1..]; let port = if let Some(port_str) = port_str.strip_prefix(':') { port_str .parse::() .map_err(|_| miette::miette!("Invalid port in URI: {uri}"))? } else { match scheme.as_str() { "https" => 443, _ => 80, } }; (host.to_string(), port) } else if let Some((h, p)) = authority.rsplit_once(':') { let port = p .parse::() .map_err(|_| miette::miette!("Invalid port in URI: {uri}"))?; (h.to_string(), port) } else { let port = match scheme.as_str() { "https" => 443, _ => 80, }; (authority.to_string(), port) }; if host.is_empty() { return Err(miette::miette!("Empty host in URI: {uri}")); } let path = if path.is_empty() { "/" } else { path }; Ok((scheme, host, port, path.to_string())) } /// Build the HTTP/1.1 `Host` value for a plain-HTTP absolute-form target. /// /// Forward proxy requests are restricted to `http`, so port 80 is omitted as /// the default port. IPv6 literals regain the brackets removed by /// `parse_proxy_uri` before they are written as an authority. fn canonical_forward_authority(host: &str, port: u16) -> String { let host = host.to_ascii_lowercase(); let host = if host.contains(':') { format!("[{host}]") } else { host }; if port == 80 { host } else { format!("{host}:{port}") } } /// Replace every received `Host` field with the authority selected from the /// absolute-form request-target, preserving any body bytes already read. /// /// RFC 9112 section 3.2.2 requires a proxy to ignore the received `Host` field /// and generate a new value from the absolute request-target. Doing this before /// L7 and middleware processing also keeps every buffered representation tied /// to the same authority used for policy selection. fn canonicalize_forward_host_header(raw: &[u8], authority: &str) -> Result> { let header_end = raw .windows(4) .position(|window| window == b"\r\n\r\n") .ok_or_else(|| miette::miette!("HTTP request headers are missing the CRLF terminator"))? + 4; crate::l7::rest::validate_http_request_header_block(&raw[..header_end])?; let header_block = std::str::from_utf8(&raw[..header_end]) .map_err(|_| miette::miette!("HTTP headers contain invalid UTF-8"))? .strip_suffix("\r\n\r\n") .expect("validated header block has terminator"); let mut lines = header_block.split("\r\n"); let request_line = lines .next() .expect("validated header block contains a request line"); let mut output = Vec::with_capacity(raw.len() + authority.len() + 8); output.extend_from_slice(request_line.as_bytes()); output.extend_from_slice(b"\r\nHost: "); output.extend_from_slice(authority.as_bytes()); output.extend_from_slice(b"\r\n"); for line in lines { let (field_name, _) = line .split_once(':') .expect("validated header field contains colon"); if field_name.eq_ignore_ascii_case("host") { continue; } output.extend_from_slice(line.as_bytes()); output.extend_from_slice(b"\r\n"); } output.extend_from_slice(b"\r\n"); output.extend_from_slice(&raw[header_end..]); Ok(output) } /// Rewrite an absolute-form HTTP proxy request to origin-form for upstream. /// /// Transforms `GET http://host:port/path HTTP/1.1` into `GET /path HTTP/1.1`, /// strips proxy hop-by-hop headers, injects `Connection: close` and `Via`. /// /// Returns the rewritten request bytes (headers + any overflow body bytes). fn rewrite_forward_request( raw: &[u8], used: usize, path: &str, canonical_authority: &str, secret_resolver: Option<&SecretResolver>, request_body_credential_rewrite: bool, ) -> Result, secrets::UnresolvedPlaceholderError> { let header_end = raw[..used] .windows(4) .position(|w| w == b"\r\n\r\n") .map_or(used, |p| p + 4); let websocket_upgrade = crate::l7::rest::request_is_websocket_upgrade(&raw[..header_end]); let upstream_path = match secret_resolver { Some(resolver) => secrets::rewrite_target_for_eval(path, resolver)?.resolved, None => path.to_string(), }; let header_str = String::from_utf8_lossy(&raw[..header_end]); let lines = header_str.split("\r\n").collect::>(); let connection_nominated: std::collections::HashSet = lines .iter() .skip(1) .filter_map(|line| line.split_once(':')) .filter(|(name, _)| name.eq_ignore_ascii_case("connection")) .flat_map(|(_, value)| value.split(',')) .map(|token| token.trim().to_ascii_lowercase()) .filter(|token| !token.is_empty()) .collect(); // Rebuild headers, stripping hop-by-hop and adding proxy headers let mut output = Vec::with_capacity(header_end + 128); let mut has_via = false; for (i, line) in lines.iter().enumerate() { if i == 0 { // Rewrite request line: METHOD absolute-uri HTTP/1.1 → METHOD path HTTP/1.1 let parts: Vec<&str> = line.splitn(3, ' ').collect(); if parts.len() == 3 { output.extend_from_slice(parts[0].as_bytes()); output.push(b' '); output.extend_from_slice(upstream_path.as_bytes()); output.push(b' '); output.extend_from_slice(parts[2].as_bytes()); } else { output.extend_from_slice(line.as_bytes()); } output.extend_from_slice(b"\r\n"); continue; } if line.is_empty() { // End of headers break; } let (field_name, _) = line .split_once(':') .expect("forward request passed strict ingress header validation"); let field_name = field_name.to_ascii_lowercase(); // RFC 9112 section 3.2.2 requires proxies to replace every received // Host field with one generated from the absolute request-target. if field_name == "host" { continue; } // Strip proxy hop-by-hop headers if matches!( field_name.as_str(), "proxy-connection" | "proxy-authorization" | "proxy-authenticate" ) { continue; } if connection_nominated.contains(&field_name) { continue; } // Reconstruct hop-by-hop upgrade fields after processing the originals. if field_name == "connection" || field_name == "upgrade" { continue; } let rewritten_line = match secret_resolver { Some(resolver) => rewrite_header_line_checked(line, resolver)?, None => line.to_string(), }; output.extend_from_slice(rewritten_line.as_bytes()); output.extend_from_slice(b"\r\n"); if field_name == "via" { has_via = true; } } // Generate the only Host field from the absolute request-target authority. output.extend_from_slice(b"Host: "); output.extend_from_slice(canonical_authority.as_bytes()); output.extend_from_slice(b"\r\n"); // Inject missing headers if websocket_upgrade { output.extend_from_slice(b"Connection: Upgrade\r\n"); output.extend_from_slice(b"Upgrade: websocket\r\n"); } else { output.extend_from_slice(b"Connection: close\r\n"); } if !has_via { output.extend_from_slice(b"Via: 1.1 openshell-sandbox\r\n"); } // End of headers output.extend_from_slice(b"\r\n"); let rewritten_header_end = output.len(); // Append any overflow body bytes from the original buffer if header_end < used { output.extend_from_slice(&raw[header_end..used]); } // Fail-closed: scan for any remaining unresolved placeholders if secret_resolver.is_some() { let scan_end = if request_body_credential_rewrite { rewritten_header_end } else { output.len() }; let output_str = String::from_utf8_lossy(&output[..scan_end]); if output_str.contains(secrets::PLACEHOLDER_PREFIX_PUBLIC) || output_str.contains(secrets::PROVIDER_ALIAS_MARKER_PUBLIC) { return Err(secrets::UnresolvedPlaceholderError { location: "header" }); } } Ok(output) } struct ForwardRelayOptions<'a> { generation_guard: &'a PolicyGenerationGuard, websocket_extensions: crate::l7::rest::WebSocketExtensionMode, secret_resolver: Option<&'a SecretResolver>, request_body_credential_rewrite: bool, } async fn relay_rewritten_forward_request( method: &str, path: &str, rewritten: Vec, client: &mut C, upstream: &mut U, options: ForwardRelayOptions<'_>, ) -> Result where C: TokioAsyncRead + TokioAsyncWrite + Unpin, U: TokioAsyncRead + TokioAsyncWrite + Unpin, { let header_end = rewritten .windows(4) .position(|w| w == b"\r\n\r\n") .map_or(rewritten.len(), |p| p + 4); let header_str = String::from_utf8_lossy(&rewritten[..header_end]); let body_length = crate::l7::rest::parse_body_length(&header_str)?; let (_, query_params) = crate::l7::rest::parse_target_query(path)?; let req = crate::l7::provider::L7Request { action: method.to_string(), target: path.to_string(), query_params, raw_header: rewritten, body_length, }; crate::l7::rest::relay_http_request_with_options_guarded( &req, client, upstream, crate::l7::rest::RelayRequestOptions { resolver: options.secret_resolver, generation_guard: Some(options.generation_guard), websocket_extensions: options.websocket_extensions, request_body_credential_rewrite: options.request_body_credential_rewrite, credential_signing: crate::l7::CredentialSigning::None, signing_service: "", signing_region: "", host: "", port: 0, }, ) .await } async fn inject_token_grant_for_forward_request( method: &str, upstream_target: &str, forward_request_bytes: Vec, l7_ctx: &crate::l7::relay::L7EvalContext, ) -> Result> { let header_end = forward_request_bytes .windows(4) .position(|w| w == b"\r\n\r\n") .map_or(forward_request_bytes.len(), |p| p + 4); let header_str = std::str::from_utf8(&forward_request_bytes[..header_end]) .into_diagnostic() .map_err(|_| miette::miette!("Forward HTTP headers contain invalid UTF-8"))?; let body_length = crate::l7::rest::parse_body_length(header_str)?; let forward_request_for_token_grant = crate::l7::provider::L7Request { action: method.to_string(), target: upstream_target.to_string(), query_params: std::collections::HashMap::new(), raw_header: forward_request_bytes, body_length, }; crate::l7::token_grant_injection::inject_if_needed(forward_request_for_token_grant, l7_ctx) .await .map(|req| req.raw_header) } /// Handle a plain HTTP forward proxy request (non-CONNECT). /// /// Public IPs are allowed through when the endpoint passes OPA evaluation. /// Private IPs require explicit `allowed_ips` on the endpoint config (SSRF /// override). Rewrites the absolute-form request to origin-form, connects /// upstream, and relays the request/response using the guarded HTTP relay. // Many distinct, non-related context parameters are required for forward proxy // dispatch; bundling them into a struct would just shift the noise into call sites. #[allow(clippy::too_many_arguments)] async fn handle_forward_proxy( method: &str, target_uri: &str, buf: &[u8], used: usize, client: &mut TcpStream, opa_engine: Arc, identity_cache: Arc, entrypoint_pid: Arc, policy_local_ctx: Option>, trusted_host_gateway: Arc>, secret_resolver: Option>, dynamic_credentials: Option< Arc< std::sync::RwLock< std::collections::HashMap, >, >, >, denial_tx: Option<&mpsc::UnboundedSender>, activity_tx: Option<&ActivitySender>, ) -> Result<()> { // 1. Parse the absolute-form URI. `path` is marked `mut` so that, when an // L7 config applies, the canonicalized form produced below replaces it // in-place — keeping OPA evaluation and the bytes written onto the wire // in sync. See the L7 block below. let (scheme, host, port, mut path) = match parse_proxy_uri(target_uri) { Ok(parsed) => parsed, Err(e) => { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .message(format!("FORWARD parse error for {target_uri}: {e}")) .build(); ocsf_emit!(event); respond(client, b"HTTP/1.1 400 Bad Request\r\n\r\n").await?; return Ok(()); } }; let host_lc = host.to_ascii_lowercase(); if host_lc == POLICY_LOCAL_HOST { if scheme != "http" || port != 80 { respond( client, &build_json_error_response( 400, "Bad Request", "invalid_policy_local_scheme", "Use http://policy.local only", ), ) .await?; return Ok(()); } if let Some(ctx) = policy_local_ctx { return crate::policy_local::handle_forward_request( &ctx, method, &path, &buf[..used], client, ) .await; } respond( client, b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 31\r\n\r\npolicy.local is not configured", ) .await?; return Ok(()); } if scheme != "http" { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Refuse) .action(ActionId::Denied) .disposition(DispositionId::Rejected) .severity(SeverityId::Informational) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!( "FORWARD rejected: unsupported scheme {scheme} for {host_lc}:{port}" )) .build(); ocsf_emit!(event); if scheme == "https" { respond( client, b"HTTP/1.1 400 Bad Request\r\nContent-Length: 27\r\n\r\nUse CONNECT for HTTPS URLs", ) .await?; } else { respond( client, &build_json_error_response( 400, "Bad Request", "unsupported_proxy_scheme", "Forward proxy requests must use http", ), ) .await?; } return Ok(()); } let canonical_authority = canonical_forward_authority(&host_lc, port); let mut forward_request_bytes = canonicalize_forward_host_header(&buf[..used], &canonical_authority)?; // 2. Evaluate OPA policy (same identity binding as CONNECT) let peer_addr = client.peer_addr().into_diagnostic()?; let _local_addr = client.local_addr().into_diagnostic()?; let opa_clone = opa_engine.clone(); let cache_clone = identity_cache.clone(); let pid_clone = entrypoint_pid.clone(); let host_clone = host_lc.clone(); let decision = tokio::task::spawn_blocking(move || { evaluate_opa_tcp( peer_addr, &opa_clone, &cache_clone, &pid_clone, &host_clone, port, ) }) .await .map_err(|e| miette::miette!("identity resolution task panicked: {e}"))?; // Build log context let binary_str = decision .binary .as_ref() .map_or_else(|| "-".to_string(), |p| p.display().to_string()); let pid_str = decision .binary_pid .map_or_else(|| "-".to_string(), |p| p.to_string()); let ancestors_str = if decision.ancestors.is_empty() { "-".to_string() } else { decision .ancestors .iter() .map(|p| p.display().to_string()) .collect::>() .join(" -> ") }; let cmdline_str = if decision.cmdline_paths.is_empty() { "-".to_string() } else { decision .cmdline_paths .iter() .map(|p| p.display().to_string()) .collect::>() .join(", ") }; // 4. Only proceed on explicit Allow — reject Deny let matched_policy = match &decision.action { NetworkAction::Allow { matched_policy } => matched_policy.clone(), NetworkAction::Deny { reason } => { { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule("-", "opa") .message(format!("FORWARD denied {method} {host_lc}:{port}{path}")) .build(); ocsf_emit!(event); } emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, reason, "forward", ); emit_activity_simple(activity_tx, true, "forward_policy"); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } }; let policy_str = matched_policy.as_deref().unwrap_or("-"); debug!( host = %host_lc, port, binary = %binary_str, binary_pid = %pid_str, matched_policy = %policy_str, decision_generation = decision.generation, current_generation = opa_engine.current_generation(), action = ?decision.action, "Forward proxy L4 policy decision" ); let sandbox_entrypoint_pid = entrypoint_pid.load(Ordering::Acquire); let forward_generation_guard = match opa_engine.generation_guard(decision.generation) { Ok(guard) => guard, Err(e) => { warn!( host = %host_lc, port, decision_generation = decision.generation, current_generation = opa_engine.current_generation(), error = %e, "Forward proxy rejected request because policy generation changed after L4 decision" ); emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } }; let mut upstream_target = path.clone(); let mut websocket_extensions = crate::l7::rest::WebSocketExtensionMode::Preserve; let mut forward_tunnel_engine: Option = None; // L7 endpoint config and evaluated request info, carried past the L7 // block so a middleware-transformed body can be re-evaluated against the // same policy inputs before it is forwarded. let mut forward_l7_reeval: Option<(crate::l7::L7EndpointConfig, crate::l7::L7RequestInfo)> = None; let mut forward_upgrade_config: Option = None; let mut forward_upgrade_target = String::new(); let mut forward_upgrade_query_params = std::collections::HashMap::new(); let mut forward_websocket_request = crate::l7::rest::request_is_websocket_upgrade(&forward_request_bytes); let mut request_body_credential_rewrite = false; let l7_ctx = crate::l7::relay::L7EvalContext { host: host_lc.clone(), port, policy_name: matched_policy.clone().unwrap_or_default(), binary_path: decision .binary .as_ref() .map(|p| p.to_string_lossy().into_owned()) .unwrap_or_default(), ancestors: decision .ancestors .iter() .map(|p| p.to_string_lossy().into_owned()) .collect(), cmdline_paths: decision .cmdline_paths .iter() .map(|p| p.to_string_lossy().into_owned()) .collect(), secret_resolver: secret_resolver.clone(), activity_tx: activity_tx.cloned(), dynamic_credentials: dynamic_credentials.clone(), token_grant_resolver: dynamic_credentials .as_ref() .map(|_| crate::l7::token_grant_injection::default_resolver()), }; let mut l7_activity_pending = false; // 4b. If the endpoint has L7 config, evaluate the request against // L7 policy. The forward proxy handles exactly one request per // connection (Connection: close), so a single evaluation suffices. if let Some(route) = query_l7_route_snapshot(&opa_engine, &decision, &host_lc, port) && !route.configs.is_empty() { if route.generation != forward_generation_guard.captured_generation() { warn!( host = %host_lc, port, decision_generation = decision.generation, guard_generation = forward_generation_guard.captured_generation(), route_generation = route.generation, current_generation = opa_engine.current_generation(), "Forward proxy rejected request because L7 route lookup used a different policy generation" ); emit_l7_tunnel_close_after_policy_change( &host_lc, port, miette::miette!( "policy changed before forward L7 evaluation [expected_generation:{} current_generation:{}]", forward_generation_guard.captured_generation(), route.generation, ), ); emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } let tunnel_engine = match opa_engine.clone_engine_for_tunnel(route.generation) { Ok(engine) => engine, Err(e) => { warn!( host = %host_lc, port, route_generation = route.generation, current_generation = opa_engine.current_generation(), error = %e, "Forward proxy rejected request because L7 tunnel engine could not be cloned" ); emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } }; // Canonicalize the request-target. The canonical form is fed to OPA // AND reassigned to the outer `path` variable so the later call to // `rewrite_forward_request` writes canonical bytes to the upstream. // This closes the policy/upstream parser-differential at this site; // without this reassignment, OPA would evaluate the canonical form // while the upstream re-normalizes the raw input and dispatches on a // potentially different path. let canonicalize_options = crate::l7::path::CanonicalizeOptions { allow_encoded_slash: route .configs .iter() .any(|snapshot| snapshot.config.allow_encoded_slash), ..Default::default() }; let query_params = match crate::l7::path::canonicalize_request_target(&path, &canonicalize_options) { Ok((canon, query)) => { upstream_target = match query.as_deref() { Some(raw_query) if !raw_query.is_empty() => { format!("{}?{raw_query}", canon.path) } _ => canon.path.clone(), }; let params = query .as_deref() .map_or_else(std::collections::HashMap::new, |q| { crate::l7::rest::parse_query_params(q).unwrap_or_default() }); path = canon.path; params } Err(e) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!( "FORWARD_L7 rejecting non-canonical request-target: {e}" )) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx, true, "l7_parse_rejection"); respond( client, &build_json_error_response( 400, "Bad Request", "invalid_request_target", "request-target must be canonical", ), ) .await?; return Ok(()); } }; let Some(l7_config) = select_l7_config_for_path(&route.configs, &path) else { emit_activity_simple(activity_tx, true, "l7_policy"); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} did not match an L7 endpoint path"), ), ) .await?; return Ok(()); }; if crate::l7::rest::request_is_h2c_upgrade(&forward_request_bytes) { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "l7") .message(format!( "FORWARD_L7 denied unsupported h2c upgrade for {method} {host_lc}:{port}{path}" )) .status_detail(crate::l7::rest::UNSUPPORTED_H2C_UPGRADE_DETAIL) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx, true, "l7_parse_rejection"); emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, crate::l7::rest::UNSUPPORTED_H2C_UPGRADE_DETAIL, "forward-l7-parse-rejection", ); respond( client, &build_json_error_response( 403, "Forbidden", "unsupported_l7_protocol", crate::l7::rest::UNSUPPORTED_H2C_UPGRADE_DETAIL, ), ) .await?; return Ok(()); } forward_websocket_request = crate::l7::rest::request_is_websocket_upgrade(&forward_request_bytes); websocket_extensions = crate::l7::relay::websocket_extension_mode(&l7_config.config); request_body_credential_rewrite = l7_config.config.protocol == crate::l7::L7Protocol::Rest && l7_config.config.request_body_credential_rewrite; forward_upgrade_config = Some(l7_config.config.clone()); forward_upgrade_target = path.clone(); forward_upgrade_query_params = query_params.clone(); let graphql = if l7_config.config.protocol == crate::l7::L7Protocol::Graphql { let header_end = forward_request_bytes .windows(4) .position(|w| w == b"\r\n\r\n") .map_or(forward_request_bytes.len(), |p| p + 4); let header_str = std::str::from_utf8(&forward_request_bytes[..header_end]) .map_err(|_| miette::miette!("Forward GraphQL headers contain invalid UTF-8"))?; let body_length = crate::l7::rest::parse_body_length(header_str)?; let mut graphql_request = crate::l7::provider::L7Request { action: method.to_string(), target: path.clone(), query_params: query_params.clone(), raw_header: forward_request_bytes, body_length, }; let info = match crate::l7::graphql::inspect_graphql_request( client, &mut graphql_request, l7_config.config.graphql_max_body_bytes, ) .await { Ok(info) => info, Err(e) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!("FORWARD_GRAPHQL_L7 request rejected: {e}")) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx, true, "l7_parse_rejection"); respond( client, &build_json_error_response( 400, "Bad Request", "invalid_graphql_request", &format!("GraphQL request rejected before policy evaluation: {e}"), ), ) .await?; return Ok(()); } }; forward_request_bytes = graphql_request.raw_header; Some(info) } else { None }; let jsonrpc = if l7_config.config.protocol.is_jsonrpc_family() { let header_end = forward_request_bytes .windows(4) .position(|w| w == b"\r\n\r\n") .map_or(forward_request_bytes.len(), |p| p + 4); let header_str = std::str::from_utf8(&forward_request_bytes[..header_end]) .map_err(|_| miette::miette!("Forward JSON-RPC headers contain invalid UTF-8"))?; let body_length = crate::l7::rest::parse_body_length(header_str)?; let mut jsonrpc_request = crate::l7::provider::L7Request { action: method.to_string(), target: path.clone(), query_params: query_params.clone(), raw_header: forward_request_bytes, body_length, }; if crate::l7::jsonrpc::jsonrpc_receive_stream_request(&jsonrpc_request) { forward_request_bytes = jsonrpc_request.raw_header; Some(crate::l7::jsonrpc::JsonRpcRequestInfo::receive_stream()) } else { let body = match crate::l7::http::read_body_for_inspection( client, &mut jsonrpc_request, l7_config.config.json_rpc_max_body_bytes, ) .await { Ok(body) => body, Err(e) => { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!("FORWARD_JSONRPC_L7 request rejected: {e}")) .build(); ocsf_emit!(event); emit_activity_simple(activity_tx, true, "l7_parse_rejection"); respond( client, &build_json_error_response( 400, "Bad Request", "invalid_jsonrpc_request", &format!("JSON-RPC request rejected before policy evaluation: {e}"), ), ) .await?; return Ok(()); } }; forward_request_bytes = jsonrpc_request.raw_header; Some(crate::l7::jsonrpc::parse_jsonrpc_body_with_options( &body, crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(&l7_config.config), )) } } else { None }; let request_info = crate::l7::L7RequestInfo { action: method.to_string(), target: path.clone(), query_params, graphql, jsonrpc, }; let hard_deny_reason = crate::l7::relay::l7_request_hard_deny_reason(l7_config.config.protocol, &request_info); let force_deny = hard_deny_reason.is_some(); let (allowed, reason) = hard_deny_reason.map_or_else( || { crate::l7::relay::evaluate_l7_request(&tunnel_engine, &l7_ctx, &request_info) .unwrap_or_else(|e| { let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .message(format!("L7 eval failed, denying request: {e}")) .build(); ocsf_emit!(event); (false, format!("L7 evaluation error: {e}")) }) }, |reason| (false, reason), ); let decision_str = match (allowed, l7_config.config.enforcement) { (_, _) if force_deny => "deny", (true, _) => "allow", (false, crate::l7::EnforcementMode::Audit) => "audit", (false, crate::l7::EnforcementMode::Enforce) => "deny", }; { let (action_id, disposition_id, severity) = match decision_str { "deny" => (ActionId::Denied, DispositionId::Blocked, SeverityId::Medium), "allow" | "audit" => ( ActionId::Allowed, DispositionId::Allowed, SeverityId::Informational, ), _ => ( ActionId::Other, DispositionId::Other, SeverityId::Informational, ), }; let engine_type = match l7_config.config.protocol { crate::l7::L7Protocol::Graphql => "l7-graphql", crate::l7::L7Protocol::JsonRpc => "l7-jsonrpc", crate::l7::L7Protocol::Mcp => "l7-mcp", _ => "l7", }; let log_message = request_info.jsonrpc.as_ref().map_or_else( || { let message_prefix = if l7_config.config.protocol == crate::l7::L7Protocol::Graphql { "FORWARD_GRAPHQL_L7" } else { "FORWARD_L7" }; format!( "{message_prefix} {decision_str} {method} {host_lc}:{port}{path} reason={reason}" ) }, |jsonrpc_info| { let endpoint = format!("{host_lc}:{port}{path}"); crate::l7::relay::jsonrpc_log_message( decision_str, method, &endpoint, jsonrpc_info, tunnel_engine.captured_generation(), &reason, ) }, ); let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(action_id) .disposition(disposition_id) .severity(severity) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, engine_type) .message(log_message) .build(); ocsf_emit!(event); } let effectively_denied = force_deny || (!allowed && l7_config.config.enforcement == crate::l7::EnforcementMode::Enforce); if effectively_denied { emit_activity_simple(activity_tx, true, "l7_policy"); emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, &reason, "forward-l7-deny", ); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} denied by L7 policy: {reason}"), ), ) .await?; return Ok(()); } l7_activity_pending = true; forward_tunnel_engine = Some(tunnel_engine); forward_l7_reeval = Some((l7_config.config.clone(), request_info)); } // 5. DNS resolution + SSRF defence (mirrors the CONNECT path logic). // - If the host is a driver-injected host-gateway alias: bypass SSRF // tiers and validate only against the trusted gateway IP. // - If allowed_ips is set: validate resolved IPs against the allowlist // (this is the SSRF override for private IP destinations). // - If the endpoint is an exact declared hostname: allow private IPs, // but still reject always-blocked addresses and control-plane ports. // - Otherwise: reject internal IPs, allow public IPs through. // When the policy host is already a literal IP address, treat it as // implicitly allowed — the user explicitly declared the destination. let mut raw_allowed_ips = query_allowed_ips(&opa_engine, &decision, &host_lc, port); if raw_allowed_ips.is_empty() { raw_allowed_ips = implicit_allowed_ips_for_ip_host(&host); } let exact_declared_endpoint_host = query_exact_declared_endpoint_host(&opa_engine, &decision, &host_lc, port); // The trusted-gateway branch is the first path; reading it before the // allowed_ips and default branches matches the policy decision narrative. #[allow(clippy::if_not_else)] let addrs = if is_host_gateway_alias(&host_lc) && let Some(gw) = *trusted_host_gateway { // Trusted host-gateway path. Mirrors the CONNECT path logic. match resolve_and_check_trusted_gateway(&host, port, gw, sandbox_entrypoint_pid).await { Ok(addrs) => addrs, Err(reason) => { { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "ssrf") .message(format!( "FORWARD blocked: trusted-gateway check failed for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity_simple(activity_tx, true, "ssrf"); respond( client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!("{method} {host_lc}:{port} blocked: trusted-gateway check failed"), ), ) .await?; return Ok(()); } } } else if !raw_allowed_ips.is_empty() { // allowed_ips mode: validate resolved IPs against CIDR allowlist. match parse_allowed_ips(&raw_allowed_ips) { Ok(nets) => { match resolve_and_check_allowed_ips(&host, port, &nets, sandbox_entrypoint_pid) .await { Ok(addrs) => addrs, Err(reason) => { { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "ssrf") .message(format!( "FORWARD blocked: allowed_ips check failed for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity_simple(activity_tx, true, "ssrf"); respond( client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!( "{method} {host_lc}:{port} blocked: allowed_ips check failed" ), ), ) .await?; return Ok(()); } } } Err(reason) => { { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "ssrf") .message(format!( "FORWARD blocked: invalid allowed_ips in policy for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity_simple(activity_tx, true, "ssrf"); respond( client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!( "{method} {host_lc}:{port} blocked: invalid allowed_ips in policy" ), ), ) .await?; return Ok(()); } } } else if exact_declared_endpoint_host { // Exact declared hostname mode mirrors CONNECT: private resolved // addresses are allowed for this operator-declared host:port, while // always-blocked addresses and control-plane ports remain denied. match resolve_and_check_declared_endpoint(&host, port, sandbox_entrypoint_pid).await { Ok(addrs) => addrs, Err(reason) => { { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "ssrf") .message(format!( "FORWARD blocked: declared endpoint check failed for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); respond( client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!( "{method} {host_lc}:{port} blocked: declared endpoint check failed" ), ), ) .await?; return Ok(()); } } } else { // No allowed_ips: reject internal IPs, allow public IPs through. match resolve_and_reject_internal(&host, port, sandbox_entrypoint_pid).await { Ok(addrs) => addrs, Err(reason) => { { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Denied) .disposition(DispositionId::Blocked) .severity(SeverityId::Medium) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "ssrf") .message(format!( "FORWARD blocked: internal IP without allowed_ips for {host_lc}:{port}" )) .status_detail(&reason) .build(); ocsf_emit!(event); } emit_denial_simple( denial_tx, &host_lc, port, &binary_str, &decision, &reason, "ssrf", ); emit_activity_simple(activity_tx, true, "ssrf"); respond( client, &build_json_error_response( 403, "Forbidden", "ssrf_denied", &format!("{method} {host_lc}:{port} blocked: internal address"), ), ) .await?; return Ok(()); } } }; if let Err(e) = forward_generation_guard.ensure_current() { warn!( host = %host_lc, port, captured_generation = forward_generation_guard.captured_generation(), current_generation = forward_generation_guard.current_generation(), error = %e, "Forward proxy rejected request because policy changed before upstream connect" ); emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); emit_activity_simple(activity_tx, true, "policy_stale"); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } // 6. Connect upstream let mut upstream = match TcpStream::connect(addrs.as_slice()).await { Ok(s) => s, Err(e) => { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .message(format!( "FORWARD upstream connect failed for {host_lc}:{port}: {e}" )) .build(); ocsf_emit!(event); respond( client, &build_json_error_response( 502, "Bad Gateway", "upstream_unreachable", &format!("connection to {host_lc}:{port} failed"), ), ) .await?; return Ok(()); } }; let middleware_path = path.split_once('?').map_or(path.as_str(), |(path, _)| path); let middleware_input = crate::opa::NetworkInput { host: host_lc.clone(), port, binary_path: decision.binary.clone().unwrap_or_default(), binary_sha256: String::new(), ancestors: decision.ancestors.clone(), cmdline_paths: decision.cmdline_paths.clone(), }; let (chain, generation) = opa_engine.query_middleware_chain_with_generation(&middleware_input)?; if generation != forward_generation_guard.captured_generation() { emit_l7_tunnel_close_after_policy_change( &host_lc, port, miette::miette!( "policy changed before forward middleware evaluation [expected_generation:{} current_generation:{}]", forward_generation_guard.captured_generation(), generation, ), ); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } if !chain.is_empty() { let middleware_runner = opa_engine.middleware_runner()?; let request = crate::l7::rest::request_from_buffered_http( method, middleware_path, &upstream_target, forward_request_bytes, )?; let l7_reevaluation = match (forward_l7_reeval.as_ref(), forward_tunnel_engine.as_ref()) { (Some((config, request_info)), Some(engine)) => Some(ForwardL7Reevaluation { config, engine, request_info, }), _ => None, }; let pipeline = ForwardMiddlewarePipeline { ctx: &l7_ctx, scheme: &scheme, runner: &middleware_runner, generation_guard: &forward_generation_guard, l7_reevaluation, }; forward_request_bytes = match pipeline.apply(request, client, chain).await? { crate::l7::middleware::MiddlewareApplyResult::Allowed(request) => request.raw_header, crate::l7::middleware::MiddlewareApplyResult::Denied { denial, .. } => { emit_activity_simple(activity_tx, true, "middleware"); let response = denial.as_ref().map_or_else( || build_middleware_failure_response(&l7_ctx.policy_name), |denial| build_middleware_deny_response(&l7_ctx.policy_name, denial), ); respond(client, &response).await?; return Ok(()); } }; } forward_request_bytes = match inject_token_grant_for_forward_request( method, &upstream_target, forward_request_bytes, &l7_ctx, ) .await { Ok(bytes) => bytes, Err(e) => { warn!( dst_host = %host_lc, dst_port = port, error = %e, "token grant failed in forward proxy" ); respond( client, &build_json_error_response( 502, "Bad Gateway", "token_grant_failed", "dynamic token grant failed", ), ) .await?; return Ok(()); } }; // 9. Rewrite request and forward to upstream let rewritten = match rewrite_forward_request( &forward_request_bytes, forward_request_bytes.len(), &upstream_target, &canonical_authority, secret_resolver.as_deref(), request_body_credential_rewrite, ) { Ok(bytes) => bytes, Err(e) => { warn!( dst_host = %host_lc, dst_port = port, error = %e, "credential injection failed in forward proxy" ); respond( client, &build_json_error_response( 500, "Internal Server Error", "credential_injection_failed", "unresolved credential placeholder in request", ), ) .await?; return Ok(()); } }; if let Err(e) = forward_generation_guard.ensure_current() { warn!( host = %host_lc, port, captured_generation = forward_generation_guard.captured_generation(), current_generation = forward_generation_guard.current_generation(), error = %e, "Forward proxy rejected request because policy changed before relay" ); emit_l7_tunnel_close_after_policy_change(&host_lc, port, e); respond( client, &build_json_error_response( 403, "Forbidden", "policy_denied", &format!("{method} {host_lc}:{port}{path} not permitted by policy"), ), ) .await?; return Ok(()); } let outcome = relay_rewritten_forward_request( method, &path, rewritten, client, &mut upstream, ForwardRelayOptions { generation_guard: &forward_generation_guard, websocket_extensions, secret_resolver: secret_resolver.as_deref(), request_body_credential_rewrite, }, ) .await?; // The request has now survived middleware, token grant, credential // rewriting, generation checks, and the HTTP relay. Only now record the // final allowed outcome. { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Other) .action(ActionId::Allowed) .disposition(DispositionId::Allowed) .severity(SeverityId::Informational) .status(StatusId::Success) .http_request(HttpRequest::new( method, OcsfUrl::new("http", &host_lc, &path, port), )) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) .src_endpoint(Endpoint::from_ip(peer_addr.ip(), peer_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), ) .firewall_rule(policy_str, "opa") .message(format!("FORWARD allowed {method} {host_lc}:{port}{path}")) .build(); ocsf_emit!(event); } emit_forward_success_activity(activity_tx, l7_activity_pending); if let crate::l7::provider::RelayOutcome::Upgraded { overflow, websocket_permessage_deflate, } = outcome { let mut upgrade_options = if let (Some(config), Some(engine)) = ( forward_upgrade_config.as_ref(), forward_tunnel_engine.as_ref(), ) { crate::l7::relay::upgrade_options( config, &l7_ctx, forward_websocket_request, &forward_upgrade_target, &forward_upgrade_query_params, Some(engine), ) } else { crate::l7::relay::UpgradeRelayOptions { websocket_request: forward_websocket_request, ..Default::default() } }; upgrade_options.websocket.permessage_deflate = websocket_permessage_deflate; crate::l7::relay::handle_upgrade( client, &mut upstream, overflow, &host_lc, port, upgrade_options, ) .await?; } Ok(()) } fn parse_target(target: &str) -> Result<(String, u16)> { let (host, port_str) = target .split_once(':') .ok_or_else(|| miette::miette!("CONNECT target missing port: {target}"))?; let port: u16 = port_str .parse() .map_err(|_| miette::miette!("Invalid port in CONNECT target: {target}"))?; Ok((host.to_string(), port)) } async fn respond(client: &mut TcpStream, bytes: &[u8]) -> Result<()> { client.write_all(bytes).await.into_diagnostic()?; Ok(()) } /// Build an HTTP error response with a JSON body. /// /// 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, }); 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() } fn build_json_error_response_with_reason( status: u16, status_text: &str, error: &str, detail: &str, reason: &str, ) -> Vec { let mut body = serde_json::json!({ "error": error, "detail": detail, }); if !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() } fn build_middleware_deny_response( policy_name: &str, denial: &openshell_supervisor_middleware::MiddlewareDenial, ) -> Vec { let mut body = serde_json::Map::new(); body.insert("error".to_string(), serde_json::json!("middleware_denied")); body.insert( "detail".to_string(), serde_json::json!("Request rejected by configured middleware"), ); body.insert("policy".to_string(), serde_json::json!(policy_name)); body.insert( "middleware".to_string(), serde_json::json!(denial.config_name), ); if let Some(reason_code) = &denial.reason_code { body.insert("reason_code".to_string(), serde_json::json!(reason_code)); } let body_str = serde_json::Value::Object(body).to_string(); format!( "HTTP/1.1 403 Forbidden\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() } fn build_middleware_failure_response(policy_name: &str) -> Vec { let body = serde_json::json!({ "error": "middleware_failed", "detail": "Request could not be processed by configured middleware", "policy": policy_name, }); let body_str = body.to_string(); format!( "HTTP/1.1 403 Forbidden\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() } /// Detail shared by the fail-closed 503 body, the OCSF denial event, and the /// denial notification when a terminating CONNECT route has no TLS termination /// state available. const TLS_TERMINATION_UNAVAILABLE_DETAIL: &str = "TLS termination unavailable (CA initialization failed); \ refusing to tunnel — credential rewrite would be bypassed"; /// Fail-closed gate evaluated BEFORE `200 Connection Established`. /// /// A CONNECT route that terminates TLS (`tls: skip` is exempt) relies on the /// ephemeral CA to rewrite credential placeholders. When no TLS termination /// state exists — the CA failed to generate or write at startup (`run.rs`); /// `mode != Proxy` never starts this handler — the proxy cannot rewrite those /// placeholders, so raw-tunneling would forward the client's TLS stream /// straight upstream and leak any `openshell:resolve:env:*` placeholder /// verbatim. /// /// The refusal is written here, as the first bytes on the socket, because a /// CONNECT client only sends its TLS `ClientHello` after reading the `200`. /// Refusing after the `200` would land the 503 inside the established tunnel, /// where the client decodes it as a TLS protocol error rather than a readable /// HTTP status (the flaw this replaces). Returns `true` when the connection was /// refused (the caller must stop) and `false` when the caller should proceed to /// establish the tunnel. async fn refuse_connect_when_tls_unavailable( client: &mut TcpStream, tls_state_present: bool, effective_tls_skip: bool, ) -> Result { if tls_state_present || effective_tls_skip { return Ok(false); } respond( client, &build_json_error_response( 503, "Service Unavailable", "tls_termination_unavailable", TLS_TERMINATION_UNAVAILABLE_DETAIL, ), ) .await?; Ok(true) } /// Check if a miette error represents a benign connection close. /// /// TLS handshake EOF, missing `close_notify`, connection resets, and broken /// pipes are all normal lifecycle events for proxied connections — not worth /// a WARN that interrupts the user's terminal. fn is_benign_relay_error(err: &miette::Report) -> bool { const BENIGN: &[&str] = &[ "close_notify", "tls handshake eof", "connection reset", "broken pipe", "unexpected eof", ]; let msg = err.to_string().to_ascii_lowercase(); BENIGN.iter().any(|pat| msg.contains(pat)) } #[cfg(test)] #[allow( clippy::needless_raw_string_hashes, clippy::iter_on_single_items, clippy::needless_continue, reason = "Test code: test fixtures and explicit control-flow markers are idiomatic in tests." )] mod tests { use super::*; use std::future::Future; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; async fn drive_raw_request_through_handler(raw: Vec) -> Vec { let policy = include_str!("../data/sandbox-policy.rego"); let data = r#" network_middlewares: guard: middleware: openshell/regex endpoints: include: ["api.example.com"] network_policies: {} "#; let engine = Arc::new(OpaEngine::from_strings(policy, data).expect("load policy")); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let client = tokio::spawn(async move { let mut socket = TcpStream::connect(address).await.unwrap(); socket.write_all(&raw).await.unwrap(); let mut response = Vec::new(); socket.read_to_end(&mut response).await.unwrap(); response }); let (server, _) = listener.accept().await.unwrap(); Box::pin(handle_tcp_connection( server, engine, Arc::new(BinaryIdentityCache::new()), Arc::new(AtomicU32::new(std::process::id())), None, None, None, Arc::new(None), None, None, None, None, )) .await .expect("malformed request should be handled"); client.await.unwrap() } #[tokio::test] async fn malformed_forward_headers_are_rejected_before_route_or_middleware_dispatch() { for host in ["api.example.com", "unmatched.example.com"] { let raw = format!( "GET http://{host}/ HTTP/1.1\r\nHost: {host}\r\nX-Guard: before\0after\r\n\r\n" ) .into_bytes(); let response = Box::pin(drive_raw_request_through_handler(raw)).await; assert!( response.starts_with(b"HTTP/1.1 400 Bad Request"), "malformed request for {host} must fail at ingress" ); } } #[tokio::test] async fn malformed_request_lines_are_rejected_before_connect_or_forward_dispatch() { for host in ["api.example.com", "unmatched.example.com"] { for request_line in [ format!("GET http://{host}/ HTTP/1.1 extra"), format!("CONNECT {host}:443 HTTP/1.1 extra"), ] { let raw = format!("{request_line}\r\nHost: {host}\r\n\r\n").into_bytes(); let response = Box::pin(drive_raw_request_through_handler(raw)).await; assert!( response.starts_with(b"HTTP/1.1 400 Bad Request"), "malformed request for {host} must fail before dispatch" ); } } } #[tokio::test] async fn unsupported_forward_scheme_is_rejected_before_policy_or_middleware_dispatch() { let raw = b"GET ftp://api.example.com/resource HTTP/1.1\r\nHost: api.example.com\r\n\r\n" .to_vec(); let response = Box::pin(drive_raw_request_through_handler(raw)).await; assert!(response.starts_with(b"HTTP/1.1 400 Bad Request")); assert!( response .windows(b"unsupported_proxy_scheme".len()) .any(|window| window == b"unsupported_proxy_scheme") ); } #[tokio::test] async fn policy_local_preserves_specific_non_http_scheme_error() { for scheme in ["https", "ftp"] { let raw = format!( "GET {scheme}://policy.local/resource HTTP/1.1\r\nHost: policy.local\r\n\r\n" ) .into_bytes(); let response = Box::pin(drive_raw_request_through_handler(raw)).await; assert!(response.starts_with(b"HTTP/1.1 400 Bad Request")); assert!( response .windows(b"invalid_policy_local_scheme".len()) .any(|window| window == b"invalid_policy_local_scheme") ); } } #[test] fn middleware_failure_response_uses_platform_text_without_policy_guidance() { let response = build_middleware_failure_response("api-policy"); 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"], "middleware_failed"); assert_eq!( body["detail"], "Request could not be processed by configured middleware" ); assert_eq!(body["policy"], "api-policy"); assert!(body.get("rule_missing").is_none()); assert!(body.get("next_steps").is_none()); 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", "CONNECT api.example.com:443 not permitted by policy", "binary '/usr/bin/node' not allowed in policy 'allow_api' (ancestors: [/usr/local/bin/claude])", ); 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"); assert_eq!( body["detail"], "CONNECT api.example.com:443 not permitted by policy" ); assert_eq!( body["reason"], "binary '/usr/bin/node' not allowed in policy 'allow_api' (ancestors: [/usr/local/bin/claude])" ); } #[test] fn policy_deny_response_omits_empty_reason() { let response = build_json_error_response_with_reason( 403, "Forbidden", "policy_denied", "CONNECT api.example.com:443 not permitted by policy", "", ); 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("reason").is_none()); } #[test] fn endpoint_only_opa_allows_declared_endpoint_without_process_identity() { let policy = include_str!("../data/sandbox-policy.rego"); let data = r#" version: 1 network_policies: test_l7: name: test_l7 endpoints: - host: host.k3d.internal port: 56123 protocol: rest enforcement: enforce rules: - allow: method: GET path: /allowed binaries: - path: /usr/bin/curl "#; let engine = OpaEngine::from_strings_with_binary_identity_required(policy, data, false) .expect("relaxed engine"); let decision = evaluate_endpoint_only_opa(&engine, "host.k3d.internal", 56123); assert_eq!( decision.action, NetworkAction::Allow { matched_policy: Some("test_l7".to_string()), } ); assert!(decision.binary.is_none()); assert!(decision.ancestors.is_empty()); let denied = evaluate_endpoint_only_opa(&engine, "api.example.com", 443); assert!( matches!(denied.action, NetworkAction::Deny { .. }), "endpoint-only mode must still deny undeclared endpoints" ); } fn websocket_l7_config( protocol: crate::l7::L7Protocol, websocket_credential_rewrite: bool, ) -> crate::l7::L7EndpointConfig { crate::l7::L7EndpointConfig { protocol, path: "/**".to_string(), tls: crate::l7::TlsMode::Auto, enforcement: crate::l7::EnforcementMode::Enforce, graphql_max_body_bytes: crate::l7::graphql::DEFAULT_MAX_BODY_BYTES, json_rpc_max_body_bytes: crate::l7::jsonrpc::DEFAULT_MAX_BODY_BYTES, mcp_strict_tool_names: true, allow_encoded_slash: false, websocket_credential_rewrite, request_body_credential_rewrite: false, websocket_graphql_policy: false, credential_signing: crate::l7::CredentialSigning::None, signing_service: String::new(), signing_region: String::new(), } } #[test] fn tunnel_protocol_classification_detects_supported_protocols() { assert_eq!( classify_tunnel_protocol(&[0x16, 0x03, 0x03, 0x00]), TunnelProtocol::Tls ); assert_eq!( classify_tunnel_protocol(b"GET / HTTP/1.1\r\n"), TunnelProtocol::Http1 ); assert_eq!( classify_tunnel_protocol(&crate::l7::rest::HTTP2_PRIOR_KNOWLEDGE_PREFACE[..8]), TunnelProtocol::H2cPriorKnowledge ); assert_eq!( classify_tunnel_protocol(b"SSH-2.0-OpenSSH\r\n"), TunnelProtocol::Unsupported ); } #[test] fn tunnel_protocol_prefix_detection_waits_for_partial_supported_prefixes() { assert!(could_be_supported_tunnel_protocol_prefix(&[0x16])); assert!(could_be_supported_tunnel_protocol_prefix(b"GE")); assert!(could_be_supported_tunnel_protocol_prefix(b"PRI * H")); assert!(!could_be_supported_tunnel_protocol_prefix(b"SSH")); } #[tokio::test] async fn h2c_prior_knowledge_is_blocked_for_l7_tunnel() { use crate::l7::middleware::UninspectableTrafficGate; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let mut client = TcpStream::connect(addr).await.unwrap(); let (server, _) = listener.accept().await.unwrap(); client .write_all(crate::l7::rest::HTTP2_PRIOR_KNOWLEDGE_PREFACE) .await .unwrap(); let protocol = peek_tunnel_protocol(&server) .await .expect("peek should succeed") .expect("client sent bytes"); assert_eq!(protocol, TunnelProtocol::H2cPriorKnowledge); assert_eq!( unsupported_l7_tunnel_protocol_detail(protocol, InspectionRequirement::L7Route), Some("HTTP/2 prior-knowledge (h2c) is not supported for L7-inspected endpoints") ); // A fail-closed middleware chain makes inspection mandatory even for // an L4-only endpoint. assert_eq!( unsupported_l7_tunnel_protocol_detail( protocol, InspectionRequirement::RequiredMiddleware ), Some("HTTP/2 prior-knowledge (h2c) cannot be inspected by required middleware") ); assert_eq!( unsupported_l7_tunnel_protocol_detail( TunnelProtocol::Unsupported, InspectionRequirement::RequiredMiddleware ), Some("Unsupported tunnel protocol cannot be inspected by required middleware") ); assert_eq!( unsupported_l7_tunnel_protocol_detail(protocol, InspectionRequirement::None), None ); // The L7 route owns the wording even when middleware also requires // inspection; the middleware requirement applies only without a route. assert_eq!( inspection_requirement(true, UninspectableTrafficGate::Deny), InspectionRequirement::L7Route ); assert_eq!( inspection_requirement(false, UninspectableTrafficGate::Deny), InspectionRequirement::RequiredMiddleware ); assert_eq!( inspection_requirement(false, UninspectableTrafficGate::BypassWithFinding), InspectionRequirement::None ); } #[tokio::test] async fn h2c_upgrade_request_on_l7_relay_is_denied_without_upstream_write() { let data = r#" network_policies: rest_api: name: rest_api endpoints: - host: h2c.example.test port: 80 path: "/allowed" protocol: rest enforcement: enforce rules: - allow: method: GET path: "/allowed" binaries: - { path: /usr/bin/node } "#; let (config, tunnel_engine, ctx) = forward_websocket_policy_parts(data, "h2c.example.test", 80, "/allowed", "rest_api"); let (mut app, mut proxy_client) = tokio::io::duplex(8192); let (mut proxy_upstream, mut upstream) = tokio::io::duplex(8192); let request = b"GET /allowed HTTP/1.1\r\n\ Host: h2c.example.test\r\n\ Upgrade: h2c\r\n\ Connection: keep-alive, Upgrade\r\n\ HTTP2-Settings: AAMAAABkAAQAAP__\r\n\r\n"; app.write_all(request).await.unwrap(); app.shutdown().await.unwrap(); crate::l7::relay::relay_with_inspection( &config, tunnel_engine, &mut proxy_client, &mut proxy_upstream, &ctx, ) .await .expect("h2c upgrade should be handled as a policy denial"); drop(proxy_client); drop(proxy_upstream); let mut response = Vec::new(); app.read_to_end(&mut response).await.unwrap(); let response = String::from_utf8(response).unwrap(); assert!( response.starts_with("HTTP/1.1 403 Forbidden\r\n"), "expected h2c upgrade to be denied, got: {response}" ); assert!( response.contains(crate::l7::rest::UNSUPPORTED_H2C_UPGRADE_DETAIL), "denial should explain unsupported h2c upgrade, got: {response}" ); let mut leaked = Vec::new(); upstream.read_to_end(&mut leaked).await.unwrap(); assert!( leaked.is_empty(), "h2c upgrade request must not be written to upstream" ); } #[test] fn connect_activity_is_skipped_when_l7_will_count_the_request() { let (tx, mut rx) = mpsc::channel(4); let activity_tx = Some(tx); let l7_route = L7RouteSnapshot { configs: vec![L7ConfigSnapshot { config: websocket_l7_config(crate::l7::L7Protocol::Rest, false), }], generation: 1, }; let l4_route = L7RouteSnapshot { configs: Vec::new(), generation: 1, }; emit_connect_activity_if_l4_only(&activity_tx, Some(&l7_route)); assert!( rx.try_recv().is_err(), "L7-inspected CONNECT should not emit an extra L4 activity event" ); emit_connect_activity_if_l4_only(&activity_tx, Some(&l4_route)); let event = rx.try_recv().expect("L4-only CONNECT should emit activity"); assert!(!event.denied); assert_eq!(event.deny_group, "unknown"); emit_connect_activity_if_l4_only(&activity_tx, None); let event = rx .try_recv() .expect("CONNECT without an L7 route should emit activity"); assert!(!event.denied); assert_eq!(event.deny_group, "unknown"); } #[test] fn l7_hard_deny_reason_includes_jsonrpc_errors() { let cases: &[(&[u8], &str)] = &[ (b"{", "JSON-RPC request rejected: invalid JSON"), ( br#"{"id":1,"method":"reports.list"}"#, "JSON-RPC request rejected: missing or non-string 'jsonrpc' field", ), ]; for &(body, expected_reason) in cases { let request_info = crate::l7::L7RequestInfo { action: "POST".to_string(), target: "/rpc".to_string(), query_params: std::collections::HashMap::new(), graphql: None, jsonrpc: Some(crate::l7::jsonrpc::parse_jsonrpc_body( body, crate::l7::jsonrpc::JsonRpcInspectionMode::JsonRpc, )), }; let reason = crate::l7::relay::l7_request_hard_deny_reason( crate::l7::L7Protocol::JsonRpc, &request_info, ) .expect("JSON-RPC parse error"); assert_eq!(reason, expected_reason); } } #[test] fn l7_hard_deny_reason_includes_jsonrpc_response_frames() { let request_info = crate::l7::L7RequestInfo { action: "POST".to_string(), target: "/rpc".to_string(), query_params: std::collections::HashMap::new(), graphql: None, jsonrpc: Some(crate::l7::jsonrpc::JsonRpcRequestInfo { calls: Vec::new(), is_batch: false, receive_stream: false, has_response: true, error: None, }), }; let reason = crate::l7::relay::l7_request_hard_deny_reason( crate::l7::L7Protocol::JsonRpc, &request_info, ) .expect("JSON-RPC response hard deny"); assert_eq!(reason, crate::l7::relay::JSONRPC_RESPONSE_FRAME_DENY_REASON); assert!( crate::l7::relay::l7_request_hard_deny_reason( crate::l7::L7Protocol::Mcp, &request_info, ) .is_none(), "MCP response frames are evaluated by policy instead of hard-denied" ); } #[tokio::test] async fn forward_middleware_pipeline_denies_policy_invalid_transformation() { const TEST_POLICY: &str = include_str!("../data/sandbox-policy.rego"); let data = r#" network_policies: jsonrpc_api: name: jsonrpc_api endpoints: - host: api.example.test port: 80 path: /rpc protocol: json-rpc enforcement: enforce rules: - allow: method: sk-ABCDEFGHIJKLMNOP binaries: - { path: /usr/bin/node } "#; let engine = OpaEngine::from_strings(TEST_POLICY, data).unwrap(); let input = crate::opa::NetworkInput { host: "api.example.test".into(), port: 80, binary_path: PathBuf::from("/usr/bin/node"), binary_sha256: "unused".into(), ancestors: vec![], cmdline_paths: vec![], }; let (endpoint, generation) = engine .query_endpoint_config_with_generation(&input) .expect("endpoint config"); let config = crate::l7::parse_l7_config(&endpoint.expect("JSON-RPC endpoint")) .expect("parse JSON-RPC config"); let tunnel_engine = engine.clone_engine_for_tunnel(generation).unwrap(); let body = br#"{"jsonrpc":"2.0","id":1,"method":"sk-ABCDEFGHIJKLMNOP"}"#; let request_info = crate::l7::L7RequestInfo { action: "POST".into(), target: "/rpc".into(), query_params: std::collections::HashMap::new(), graphql: None, jsonrpc: Some(crate::l7::jsonrpc::parse_jsonrpc_body_with_options( body, crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(&config), )), }; let raw = format!( "POST /rpc HTTP/1.1\r\nHost: api.example.test\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), std::str::from_utf8(body).unwrap() ) .into_bytes(); let request = crate::l7::rest::request_from_buffered_http("POST", "/rpc", "/rpc", raw).unwrap(); let ctx = crate::l7::relay::L7EvalContext { host: "api.example.test".into(), port: 80, policy_name: "jsonrpc_api".into(), binary_path: "/usr/bin/node".into(), ancestors: vec![], cmdline_paths: vec![], secret_resolver: None, activity_tx: None, dynamic_credentials: None, token_grant_resolver: None, }; let runner = openshell_supervisor_middleware::ChainRunner::new( openshell_supervisor_middleware_builtins::services() .into_iter() .next() .expect("built-in middleware service"), ); let pipeline = ForwardMiddlewarePipeline { ctx: &ctx, scheme: "http", runner: &runner, generation_guard: tunnel_engine.generation_guard(), l7_reevaluation: Some(ForwardL7Reevaluation { config: &config, engine: &tunnel_engine, request_info: &request_info, }), }; let chain = vec![openshell_supervisor_middleware::ChainEntry { name: "redactor".into(), implementation: openshell_supervisor_middleware_builtins::BUILTIN_REGEX.into(), order: 0, config: prost_types::Struct::default(), on_error: openshell_supervisor_middleware::OnError::FailClosed, }]; let (_app, mut client) = tokio::io::duplex(8192); let outcome = pipeline .apply(request, &mut client, chain) .await .expect("forward middleware pipeline"); match outcome { crate::l7::middleware::MiddlewareApplyResult::Denied { denial } => { assert!(denial.is_none()); } crate::l7::middleware::MiddlewareApplyResult::Allowed(_) => { panic!("policy-invalid transformed request must be denied") } } } #[test] fn forward_l7_allowed_activity_is_deferred_until_after_ssrf() { let (tx, mut rx) = mpsc::channel(4); let activity_tx = Some(tx); let l7_activity_pending = true; assert!( rx.try_recv().is_err(), "allowed L7 evaluation must not emit activity before SSRF succeeds" ); emit_activity_simple(activity_tx.as_ref(), true, "ssrf"); let event = rx .try_recv() .expect("SSRF denial should emit the request activity"); assert!(event.denied); assert_eq!(event.deny_group, "ssrf"); assert!( rx.try_recv().is_err(), "SSRF-denied forward request must not also emit allowed L7 activity" ); emit_forward_success_activity(activity_tx.as_ref(), l7_activity_pending); let event = rx .try_recv() .expect("L7 activity should emit after SSRF succeeds"); assert!(!event.denied); assert_eq!(event.deny_group, "l7_policy"); } #[test] fn forward_middleware_denial_does_not_emit_success_activity() { let (tx, mut rx) = mpsc::channel(4); let activity_tx = Some(tx); emit_activity_simple(activity_tx.as_ref(), true, "middleware"); let event = rx .try_recv() .expect("middleware denial should emit the request activity"); assert!(event.denied); assert_eq!(event.deny_group, "middleware"); assert!( rx.try_recv().is_err(), "middleware-denied forward request must not emit success activity" ); } #[test] fn forward_success_activity_uses_unknown_without_l7() { let (tx, mut rx) = mpsc::channel(4); let activity_tx = Some(tx); emit_forward_success_activity(activity_tx.as_ref(), false); let event = rx .try_recv() .expect("non-L7 forward success should emit activity"); assert!(!event.denied); assert_eq!(event.deny_group, "unknown"); } fn forward_test_guard() -> PolicyGenerationGuard { let policy = include_str!("../data/sandbox-policy.rego"); let policy_data = "network_policies: {}\n"; let engine = OpaEngine::from_strings(policy, policy_data).unwrap(); engine .generation_guard(engine.current_generation()) .unwrap() } async fn relay_forward_request_and_capture( method: &str, path: &str, raw: &[u8], resolver: Option<&SecretResolver>, request_body_credential_rewrite: bool, ) -> Result { let guard = forward_test_guard(); let target_uri = std::str::from_utf8(raw) .expect("forward test request is UTF-8") .lines() .next() .and_then(|line| line.split(' ').nth(1)) .expect("forward test request has an absolute target"); let (_, host, port, _) = parse_proxy_uri(target_uri)?; let authority = canonical_forward_authority(&host, port); let rewritten = rewrite_forward_request( raw, raw.len(), path, &authority, resolver, request_body_credential_rewrite, ) .map_err(|e| miette::miette!("{e}"))?; let (mut proxy_to_upstream, mut upstream_side) = tokio::io::duplex(8192); let (mut _app_side, mut proxy_to_client) = tokio::io::duplex(8192); let upstream_task = tokio::spawn(async move { let mut buf = vec![0u8; 8192]; let mut total = 0usize; let mut expected_total = None; loop { let n = upstream_side.read(&mut buf[total..]).await.unwrap(); if n == 0 { break; } total += n; if expected_total.is_none() && let Some(end) = buf[..total].windows(4).position(|w| w == b"\r\n\r\n") { let header_end = end + 4; let headers = String::from_utf8_lossy(&buf[..header_end]); let len = headers .lines() .find_map(|line| { let (name, value) = line.split_once(':')?; name.eq_ignore_ascii_case("content-length") .then(|| value.trim().parse::().ok()) .flatten() }) .unwrap_or(0); expected_total = Some(header_end + len); } if expected_total.is_some_and(|expected| total >= expected) { break; } } upstream_side .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") .await .unwrap(); upstream_side.flush().await.unwrap(); String::from_utf8_lossy(&buf[..total]).to_string() }); relay_rewritten_forward_request( method, path, rewritten, &mut proxy_to_client, &mut proxy_to_upstream, ForwardRelayOptions { generation_guard: &guard, websocket_extensions: crate::l7::rest::WebSocketExtensionMode::Preserve, secret_resolver: resolver, request_body_credential_rewrite, }, ) .await?; upstream_task .await .map_err(|e| miette::miette!("upstream task failed: {e}")) } fn forward_token_grant_context( resolver_response: std::result::Result<&str, &str>, ) -> ( crate::l7::relay::L7EvalContext, crate::l7::token_grant_injection::test_support::TokenGrantTestFixture, ) { let provider_key = "api.example.test\t8080\t/v1/**\tprovider:access_token"; let fixture = match resolver_response { Ok(token) => { crate::l7::token_grant_injection::test_support::TokenGrantTestFixture::success( provider_key, token, ) } Err(error) => { crate::l7::token_grant_injection::test_support::TokenGrantTestFixture::failure( provider_key, error, ) } }; let ctx = crate::l7::relay::L7EvalContext { host: "api.example.test".into(), port: 8080, policy_name: "rest_api".into(), binary_path: "/usr/bin/curl".into(), ancestors: vec![], cmdline_paths: vec![], secret_resolver: None, activity_tx: None, dynamic_credentials: Some(fixture.dynamic_credentials()), token_grant_resolver: Some(fixture.resolver()), }; (ctx, fixture) } fn authorization_header_count(headers: &str) -> usize { headers .lines() .filter(|line| { line.split_once(':') .is_some_and(|(name, _)| name.eq_ignore_ascii_case("authorization")) }) .count() } fn forward_websocket_policy_parts( data: &str, host: &str, port: u16, path: &str, policy_name: &str, ) -> ( crate::l7::L7EndpointConfig, crate::opa::TunnelPolicyEngine, crate::l7::relay::L7EvalContext, ) { let policy = include_str!("../data/sandbox-policy.rego"); let engine = OpaEngine::from_strings(policy, data).unwrap(); let decision = ConnectDecision { action: NetworkAction::Allow { matched_policy: Some(policy_name.to_string()), }, generation: engine.current_generation(), binary: Some(PathBuf::from("/usr/bin/node")), binary_pid: None, ancestors: vec![], cmdline_paths: vec![], }; let route = query_l7_route_snapshot(&engine, &decision, host, port).expect("L7 route should match"); let config = select_l7_config_for_path(&route.configs, path) .expect("path-specific L7 config should match") .config .clone(); let tunnel_engine = engine .clone_engine_for_tunnel(route.generation) .expect("tunnel engine"); let ctx = crate::l7::relay::L7EvalContext { host: host.to_string(), port, policy_name: policy_name.to_string(), binary_path: "/usr/bin/node".to_string(), ancestors: vec![], cmdline_paths: vec![], secret_resolver: None, activity_tx: None, dynamic_credentials: None, token_grant_resolver: None, }; (config, tunnel_engine, ctx) } async fn read_http_headers(reader: &mut R) -> Vec { let mut bytes = Vec::new(); let mut chunk = [0u8; 256]; loop { let n = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read(&mut chunk)) .await .expect("HTTP headers should arrive") .expect("header read should succeed"); assert!(n > 0, "stream closed before HTTP headers"); bytes.extend_from_slice(&chunk[..n]); if bytes.windows(4).any(|w| w == b"\r\n\r\n") { return bytes; } } } fn masked_text_frame(payload: &[u8]) -> Vec { let mask = [0x11, 0x22, 0x33, 0x44]; assert!( payload.len() <= 125, "test helper only supports small frames" ); let payload_len = u8::try_from(payload.len()).expect("small frame length"); let mut frame = vec![0x81, 0x80 | payload_len]; frame.extend_from_slice(&mask); frame.extend( payload .iter() .enumerate() .map(|(idx, byte)| byte ^ mask[idx % 4]), ); frame } async fn forward_websocket_denied_after_upgrade( config: crate::l7::L7EndpointConfig, tunnel_engine: crate::opa::TunnelPolicyEngine, ctx: crate::l7::relay::L7EvalContext, path: &str, payload: &str, ) -> (miette::Report, Vec) { let host = ctx.host.clone(); let port = ctx.port; let raw = format!( "GET http://{host}{path} HTTP/1.1\r\n\ Host: {host}\r\n\ Upgrade: websocket\r\n\ Connection: Upgrade\r\n\ Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\ Sec-WebSocket-Version: 13\r\n\r\n" ); let rewritten = rewrite_forward_request( raw.as_bytes(), raw.len(), path, &canonical_forward_authority(&host, port), None, false, ) .expect("forward websocket request should rewrite to origin form"); let websocket_extensions = crate::l7::relay::websocket_extension_mode(&config); let target = path.to_string(); let query_params = std::collections::HashMap::new(); let (mut proxy_to_upstream, mut upstream) = tokio::io::duplex(8192); let (mut app, mut proxy_to_client) = tokio::io::duplex(8192); let relay = tokio::spawn(async move { let guard = tunnel_engine.generation_guard(); let outcome = relay_rewritten_forward_request( "GET", &target, rewritten, &mut proxy_to_client, &mut proxy_to_upstream, ForwardRelayOptions { generation_guard: guard, websocket_extensions, secret_resolver: None, request_body_credential_rewrite: false, }, ) .await?; if let crate::l7::provider::RelayOutcome::Upgraded { overflow, websocket_permessage_deflate, } = outcome { let mut options = crate::l7::relay::upgrade_options( &config, &ctx, true, &target, &query_params, Some(&tunnel_engine), ); options.websocket.permessage_deflate = websocket_permessage_deflate; crate::l7::relay::handle_upgrade( &mut proxy_to_client, &mut proxy_to_upstream, overflow, &host, port, options, ) .await?; } Ok::<(), miette::Report>(()) }); let forwarded_headers = read_http_headers(&mut upstream).await; let forwarded_headers = String::from_utf8_lossy(&forwarded_headers); assert!(forwarded_headers.starts_with(&format!("GET {path} HTTP/1.1\r\n"))); assert!(forwarded_headers.contains("Upgrade: websocket\r\n")); upstream .write_all( b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", ) .await .unwrap(); let response = read_http_headers(&mut app).await; assert!(String::from_utf8_lossy(&response).contains("101 Switching Protocols")); app.write_all(&masked_text_frame(payload.as_bytes())) .await .unwrap(); let err = tokio::time::timeout(std::time::Duration::from_secs(1), relay) .await .expect("websocket relay should fail closed after denied frame") .expect("relay task should not panic") .expect_err("denied websocket frame should fail the forward relay"); let mut leaked = Vec::new(); tokio::time::timeout( std::time::Duration::from_secs(1), upstream.read_to_end(&mut leaked), ) .await .expect("upstream side should close") .expect("upstream read should succeed"); (err, leaked) } #[test] fn forward_websocket_upgrade_options_enable_native_policy_context() { let (_, resolver) = SecretResolver::from_provider_env( [("DISCORD_BOT_TOKEN".to_string(), "discord-real".to_string())] .into_iter() .collect(), ); let resolver = resolver.map(Arc::new); let policy = include_str!("../data/sandbox-policy.rego"); let policy_data = "network_policies: {}\n"; let engine = OpaEngine::from_strings(policy, policy_data).unwrap(); let tunnel_engine = engine .clone_engine_for_tunnel(engine.current_generation()) .unwrap(); let ctx = crate::l7::relay::L7EvalContext { host: "gateway.example.test".to_string(), port: 80, policy_name: "ws_api".to_string(), binary_path: "/usr/bin/node".to_string(), ancestors: vec![], cmdline_paths: vec![], secret_resolver: resolver, activity_tx: None, dynamic_credentials: None, token_grant_resolver: None, }; let query_params = std::collections::HashMap::new(); let extensions = crate::l7::relay::websocket_extension_mode(&websocket_l7_config( crate::l7::L7Protocol::Websocket, true, )); let options = crate::l7::relay::upgrade_options( &websocket_l7_config(crate::l7::L7Protocol::Websocket, true), &ctx, true, "/ws", &query_params, Some(&tunnel_engine), ); assert_eq!( extensions, crate::l7::rest::WebSocketExtensionMode::PermessageDeflate ); assert!(options.websocket.credential_rewrite); assert!(options.secret_resolver.is_some()); assert!(options.engine.is_some()); assert!(options.ctx.is_some()); assert!(matches!( options.websocket.message_policy, crate::l7::relay::WebSocketMessagePolicy::Transport )); } #[test] fn forward_websocket_upgrade_options_preserve_rest_without_rewrite() { let ctx = crate::l7::relay::L7EvalContext { host: "gateway.example.test".to_string(), port: 80, policy_name: "rest_api".to_string(), binary_path: "/usr/bin/node".to_string(), ancestors: vec![], cmdline_paths: vec![], secret_resolver: None, activity_tx: None, dynamic_credentials: None, token_grant_resolver: None, }; let query_params = std::collections::HashMap::new(); let config = websocket_l7_config(crate::l7::L7Protocol::Rest, false); let extensions = crate::l7::relay::websocket_extension_mode(&config); let options = crate::l7::relay::upgrade_options(&config, &ctx, true, "/ws", &query_params, None); assert_eq!( extensions, crate::l7::rest::WebSocketExtensionMode::Preserve ); assert!(!options.websocket.credential_rewrite); assert!(options.secret_resolver.is_none()); assert!(options.engine.is_none()); assert!(options.ctx.is_none()); assert!(matches!( options.websocket.message_policy, crate::l7::relay::WebSocketMessagePolicy::None )); } #[tokio::test] async fn forward_websocket_upgrade_blocks_text_frame_by_policy() { let data = r#" network_policies: ws_api: name: ws_api endpoints: - host: gateway.example.test port: 80 path: "/ws" protocol: websocket enforcement: enforce rules: - allow: method: GET path: "/ws" - allow: method: WEBSOCKET_TEXT path: "/ws" deny_rules: - method: WEBSOCKET_TEXT path: "/ws" binaries: - { path: /usr/bin/node } "#; let (config, tunnel_engine, ctx) = forward_websocket_policy_parts(data, "gateway.example.test", 80, "/ws", "ws_api"); let (err, leaked) = forward_websocket_denied_after_upgrade( config, tunnel_engine, ctx, "/ws", r#"{"type":"unsafe"}"#, ) .await; assert!(err.to_string().contains("websocket text message denied")); assert!( leaked.is_empty(), "denied forward-proxy WebSocket text frames must not reach upstream" ); } #[tokio::test] async fn forward_graphql_websocket_upgrade_blocks_unallowed_operation() { let data = r#" network_policies: graphql_ws: name: graphql_ws endpoints: - host: gateway.example.test port: 80 path: "/graphql" protocol: websocket enforcement: enforce rules: - allow: method: GET path: "/graphql" - allow: operation_type: query fields: [viewer] deny_rules: - operation_type: query fields: [admin] binaries: - { path: /usr/bin/node } "#; let (config, tunnel_engine, ctx) = forward_websocket_policy_parts( data, "gateway.example.test", 80, "/graphql", "graphql_ws", ); assert!( config.websocket_graphql_policy, "operation rules should enable GraphQL-over-WebSocket inspection" ); let (err, leaked) = forward_websocket_denied_after_upgrade( config, tunnel_engine, ctx, "/graphql", r#"{"id":"1","type":"subscribe","payload":{"query":"query { admin }"}}"#, ) .await; assert!(err.to_string().contains("websocket GraphQL message denied")); assert!( leaked.is_empty(), "denied forward-proxy GraphQL WebSocket operations must not reach upstream" ); } #[test] fn l7_route_selection_prefers_path_specific_graphql_endpoint() { let configs = vec![ L7ConfigSnapshot { config: crate::l7::L7EndpointConfig { protocol: crate::l7::L7Protocol::Rest, path: "/**".to_string(), tls: crate::l7::TlsMode::Auto, enforcement: crate::l7::EnforcementMode::Enforce, graphql_max_body_bytes: crate::l7::graphql::DEFAULT_MAX_BODY_BYTES, json_rpc_max_body_bytes: crate::l7::jsonrpc::DEFAULT_MAX_BODY_BYTES, mcp_strict_tool_names: true, allow_encoded_slash: false, websocket_credential_rewrite: false, request_body_credential_rewrite: false, websocket_graphql_policy: false, credential_signing: crate::l7::CredentialSigning::None, signing_service: String::new(), signing_region: String::new(), }, }, L7ConfigSnapshot { config: crate::l7::L7EndpointConfig { protocol: crate::l7::L7Protocol::Graphql, path: "/graphql".to_string(), tls: crate::l7::TlsMode::Auto, enforcement: crate::l7::EnforcementMode::Enforce, graphql_max_body_bytes: crate::l7::graphql::DEFAULT_MAX_BODY_BYTES, json_rpc_max_body_bytes: crate::l7::jsonrpc::DEFAULT_MAX_BODY_BYTES, mcp_strict_tool_names: true, allow_encoded_slash: false, websocket_credential_rewrite: false, request_body_credential_rewrite: false, websocket_graphql_policy: false, credential_signing: crate::l7::CredentialSigning::None, signing_service: String::new(), signing_region: String::new(), }, }, ]; let selected = select_l7_config_for_path(&configs, "/graphql").expect("expected path-specific route"); assert_eq!(selected.config.protocol, crate::l7::L7Protocol::Graphql); let selected = select_l7_config_for_path(&configs, "/repos/org/repo").expect("expected REST route"); assert_eq!(selected.config.protocol, crate::l7::L7Protocol::Rest); } // -- is_internal_ip: IPv4 -- #[test] fn test_rejects_ipv4_loopback() { assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::LOCALHOST))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2)))); } #[test] fn test_rejects_ipv4_private_10() { assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(10, 255, 255, 255)))); } #[test] fn test_rejects_ipv4_private_172_16() { assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(172, 16, 0, 1)))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(172, 31, 255, 255)))); } #[test] fn test_rejects_ipv4_private_192_168() { assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(192, 168, 0, 1)))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new( 192, 168, 255, 255 )))); } #[test] fn test_rejects_ipv4_link_local_metadata() { // Cloud metadata endpoint assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new( 169, 254, 169, 254 )))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(169, 254, 0, 1)))); } #[test] fn test_rejects_ipv4_unspecified() { assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::UNSPECIFIED))); } #[test] fn test_rejects_ipv4_cgnat() { // 100.64.0.0/10 — CGNAT / shared address space (RFC 6598) assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(100, 64, 0, 1)))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(100, 100, 50, 3)))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new( 100, 127, 255, 255 )))); // Just outside the /10 boundary assert!(!is_internal_ip(IpAddr::V4(Ipv4Addr::new(100, 128, 0, 1)))); assert!(!is_internal_ip(IpAddr::V4(Ipv4Addr::new( 100, 63, 255, 255 )))); } #[test] fn test_rejects_ipv4_special_use_ranges() { // 192.0.0.0/24 — IETF protocol assignments assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(192, 0, 0, 1)))); // 198.18.0.0/15 — benchmarking assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(198, 18, 0, 1)))); assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(198, 19, 255, 255)))); // 198.51.100.0/24 — TEST-NET-2 assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1)))); // 203.0.113.0/24 — TEST-NET-3 assert!(is_internal_ip(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)))); } #[test] fn test_rejects_ipv6_mapped_cgnat() { // ::ffff:100.64.0.1 should be caught via IPv4-mapped unwrapping let v6 = Ipv4Addr::new(100, 64, 0, 1).to_ipv6_mapped(); assert!(is_internal_ip(IpAddr::V6(v6))); } #[test] fn test_allows_ipv4_public() { assert!(!is_internal_ip(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); assert!(!is_internal_ip(IpAddr::V4(Ipv4Addr::new(1, 1, 1, 1)))); assert!(!is_internal_ip(IpAddr::V4(Ipv4Addr::new(93, 184, 216, 34)))); } #[test] fn test_allows_ipv4_non_private_172() { // 172.32.0.0 is outside the 172.16/12 private range assert!(!is_internal_ip(IpAddr::V4(Ipv4Addr::new(172, 32, 0, 1)))); } // -- is_internal_ip: IPv6 -- #[test] fn test_rejects_ipv6_loopback() { assert!(is_internal_ip(IpAddr::V6(Ipv6Addr::LOCALHOST))); } #[test] fn test_rejects_ipv6_unspecified() { assert!(is_internal_ip(IpAddr::V6(Ipv6Addr::UNSPECIFIED))); } #[test] fn test_rejects_ipv6_link_local() { // fe80::1 assert!(is_internal_ip(IpAddr::V6(Ipv6Addr::new( 0xfe80, 0, 0, 0, 0, 0, 0, 1 )))); } #[test] fn test_rejects_ipv6_unique_local_address() { // fdc4:f303:9324::254 assert!(is_internal_ip(IpAddr::V6(Ipv6Addr::new( 0xfdc4, 0xf303, 0x9324, 0, 0, 0, 0, 0x0254 )))); } #[test] fn test_rejects_ipv4_mapped_ipv6_private() { // ::ffff:10.0.0.1 let v6 = Ipv4Addr::new(10, 0, 0, 1).to_ipv6_mapped(); assert!(is_internal_ip(IpAddr::V6(v6))); } #[test] fn test_rejects_ipv4_mapped_ipv6_loopback() { // ::ffff:127.0.0.1 let v6 = Ipv4Addr::LOCALHOST.to_ipv6_mapped(); assert!(is_internal_ip(IpAddr::V6(v6))); } #[test] fn test_rejects_ipv4_mapped_ipv6_link_local() { // ::ffff:169.254.169.254 let v6 = Ipv4Addr::new(169, 254, 169, 254).to_ipv6_mapped(); assert!(is_internal_ip(IpAddr::V6(v6))); } #[test] fn test_allows_ipv6_public() { // 2001:4860:4860::8888 (Google DNS) assert!(!is_internal_ip(IpAddr::V6(Ipv6Addr::new( 0x2001, 0x4860, 0x4860, 0, 0, 0, 0, 0x8888 )))); } #[test] fn test_allows_ipv4_mapped_ipv6_public() { // ::ffff:8.8.8.8 let v6 = Ipv4Addr::new(8, 8, 8, 8).to_ipv6_mapped(); assert!(!is_internal_ip(IpAddr::V6(v6))); } // -- resolve_and_reject_internal -- #[test] fn test_parse_hosts_file_for_host_handles_comments_invalid_rows_and_case() { let contents = r#" # comment 192.168.1.105 searxng.local searxng bad-ip ignored.local 93.184.216.34 Example.Local # trailing comment ::1 loopback.local 192.168.1.105 searxng.local "#; let result = parse_hosts_file_for_host(contents, "SEARXNG.LOCAL"); assert_eq!(result, vec![IpAddr::V4(Ipv4Addr::new(192, 168, 1, 105))]); let public = parse_hosts_file_for_host(contents, "example.local"); assert_eq!(public, vec![IpAddr::V4(Ipv4Addr::new(93, 184, 216, 34))]); } #[test] fn test_resolve_from_hosts_file_contents_requires_exact_alias_match() { let contents = "192.168.1.105 searxng.local\n"; assert!( resolve_from_hosts_file_contents(contents, "searxng", 8080).is_empty(), "partial alias match should not resolve" ); let result = resolve_from_hosts_file_contents(contents, "searxng.local", 8080); assert_eq!( result, vec![SocketAddr::new( IpAddr::V4(Ipv4Addr::new(192, 168, 1, 105)), 8080 )] ); } #[test] fn test_resolve_from_hosts_file_contents_public_ip_passes_default_ssrf_check() { let addrs = resolve_from_hosts_file_contents("93.184.216.34 example.local\n", "example.local", 80); assert!(reject_internal_resolved_addrs("example.local", &addrs).is_ok()); } #[test] fn test_resolve_from_hosts_file_contents_private_ip_requires_allowed_ips() { let addrs = resolve_from_hosts_file_contents( "192.168.1.105 searxng.local\n", "searxng.local", 8080, ); let err = reject_internal_resolved_addrs("searxng.local", &addrs).unwrap_err(); assert!( err.contains("internal address"), "expected private hosts-file resolution to remain blocked: {err}" ); let nets = parse_allowed_ips(&["192.168.1.105/32".to_string()]).unwrap(); assert!( validate_allowed_ips_for_resolved_addrs("searxng.local", 8080, &addrs, &nets).is_ok() ); } #[test] fn test_declared_endpoint_private_hosts_file_resolution_allowed() { let addrs = resolve_from_hosts_file_contents( "192.168.1.105 searxng.local\n", "searxng.local", 8080, ); assert!(validate_declared_endpoint_resolved_addrs("searxng.local", 8080, &addrs).is_ok()); } #[test] fn test_declared_endpoint_loopback_stays_blocked() { let addrs = resolve_from_hosts_file_contents("127.0.0.1 loopback.local\n", "loopback.local", 80); let err = validate_declared_endpoint_resolved_addrs("loopback.local", 80, &addrs).unwrap_err(); assert!( err.contains("always-blocked"), "expected loopback to stay blocked: {err}" ); } #[test] fn test_declared_endpoint_link_local_stays_blocked() { let addrs = resolve_from_hosts_file_contents( "169.254.169.254 metadata.local\n", "metadata.local", 80, ); let err = validate_declared_endpoint_resolved_addrs("metadata.local", 80, &addrs).unwrap_err(); assert!( err.contains("always-blocked"), "expected link-local to stay blocked: {err}" ); } #[test] fn test_declared_endpoint_blocks_control_plane_ports() { let addrs = resolve_from_hosts_file_contents("10.0.0.5 kube-api.local\n", "kube-api.local", 6443); let err = validate_declared_endpoint_resolved_addrs("kube-api.local", 6443, &addrs).unwrap_err(); assert!( err.contains("blocked control-plane port"), "expected control-plane port to stay blocked: {err}" ); } #[test] fn test_resolve_from_hosts_file_contents_always_blocked_ip_stays_blocked() { let addrs = resolve_from_hosts_file_contents("127.0.0.1 loopback.local\n", "loopback.local", 80); let nets = vec!["127.0.0.0/8".parse::().unwrap()]; let err = validate_allowed_ips_for_resolved_addrs("loopback.local", 80, &addrs, &nets) .unwrap_err(); assert!( err.contains("always-blocked"), "expected always-blocked hosts-file resolution to stay blocked: {err}" ); } #[test] fn test_resolve_from_hosts_file_contents_returns_empty_without_match() { let result = resolve_from_hosts_file_contents("192.168.1.105 searxng.local\n", "missing.local", 80); assert!(result.is_empty()); } // -- is_host_gateway_alias -- #[test] fn test_is_host_gateway_alias_recognises_known_aliases() { assert!(is_host_gateway_alias("host.openshell.internal")); assert!(is_host_gateway_alias("host.containers.internal")); assert!(is_host_gateway_alias("host.docker.internal")); } #[test] fn test_is_host_gateway_alias_is_case_insensitive() { assert!(is_host_gateway_alias("HOST.OPENSHELL.INTERNAL")); assert!(is_host_gateway_alias("Host.Containers.Internal")); assert!(is_host_gateway_alias("HOST.DOCKER.INTERNAL")); } #[test] fn test_is_host_gateway_alias_rejects_unknown_hosts() { assert!(!is_host_gateway_alias("api.example.com")); assert!(!is_host_gateway_alias("host.openshell.internal.evil.com")); assert!(!is_host_gateway_alias("evil.host.openshell.internal")); assert!(!is_host_gateway_alias("openshell.internal")); assert!(!is_host_gateway_alias("")); } // -- is_cloud_metadata_ip -- #[test] fn test_is_cloud_metadata_ip_blocks_known_metadata_ip() { assert!(is_cloud_metadata_ip(IpAddr::V4(Ipv4Addr::new( 169, 254, 169, 254 )))); } #[test] fn test_is_cloud_metadata_ip_allows_other_link_local() { // The pasta gateway address on this test host — not a metadata IP. assert!(!is_cloud_metadata_ip(IpAddr::V4(Ipv4Addr::new( 169, 254, 1, 2 )))); assert!(!is_cloud_metadata_ip(IpAddr::V4(Ipv4Addr::new( 169, 254, 0, 1 )))); } #[test] fn test_is_cloud_metadata_ip_allows_private_and_public() { assert!(!is_cloud_metadata_ip(IpAddr::V4(Ipv4Addr::new( 10, 0, 0, 1 )))); assert!(!is_cloud_metadata_ip(IpAddr::V4(Ipv4Addr::new( 192, 168, 1, 1 )))); assert!(!is_cloud_metadata_ip(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); } #[test] fn test_is_cloud_metadata_ip_blocks_ipv4_mapped_metadata() { // ::ffff:169.254.169.254 is the IPv4-mapped IPv6 representation of the // AWS/GCP/Azure IMDS endpoint. is_link_local_ip() recognizes it as // link-local, so is_cloud_metadata_ip() must also catch it — otherwise // the trusted-gateway exemption would be granted to the metadata service. let mapped = Ipv4Addr::new(169, 254, 169, 254).to_ipv6_mapped(); assert!( is_cloud_metadata_ip(IpAddr::V6(mapped)), "::ffff:169.254.169.254 must be recognized as cloud metadata" ); } #[test] fn test_is_cloud_metadata_ip_allows_other_ipv4_mapped_link_local() { // Other IPv4-mapped link-local addresses are NOT metadata. let mapped = Ipv4Addr::new(169, 254, 1, 2).to_ipv6_mapped(); assert!( !is_cloud_metadata_ip(IpAddr::V6(mapped)), "::ffff:169.254.1.2 should not be flagged as cloud metadata" ); } // -- detect_trusted_host_gateway -- #[test] fn test_detect_trusted_host_gateway_returns_ip_from_hosts_content() { // We test the underlying parser directly since detect_trusted_host_gateway // reads the real /etc/hosts. The production code composes these same primitives. let contents = "169.254.1.2\thost.openshell.internal host.containers.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!(ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))]); } #[test] fn test_detect_trusted_host_gateway_ignores_cloud_metadata_ip() { // Simulate a /etc/hosts where the driver injected the cloud metadata IP — // this should be caught and suppressed. let contents = "169.254.169.254\thost.openshell.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!(ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254))]); // is_cloud_metadata_ip should flag it, preventing the exemption. assert!(is_cloud_metadata_ip(ips[0])); } #[test] fn test_detect_trusted_host_gateway_no_entry_returns_empty() { let contents = "127.0.0.1 localhost\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert!(ips.is_empty()); } #[test] fn test_detect_trusted_host_gateway_rejects_loopback() { // Loopback is not link-local — must not receive the SSRF exemption. let ip = IpAddr::V4(Ipv4Addr::LOCALHOST); assert!(!is_cloud_metadata_ip(ip)); assert!(!is_link_local_ip(ip)); // The guard: !link-local → reject. assert!(!is_link_local_ip(ip)); } #[test] fn test_detect_trusted_host_gateway_rejects_unspecified() { // Unspecified (0.0.0.0) is not link-local — must not be trusted. let ip = IpAddr::V4(Ipv4Addr::UNSPECIFIED); assert!(!is_cloud_metadata_ip(ip)); assert!(!is_link_local_ip(ip)); assert!(!is_link_local_ip(ip)); } #[test] fn test_detect_trusted_host_gateway_rejects_loopback_v6() { let ip = IpAddr::V6(Ipv6Addr::LOCALHOST); assert!(!is_cloud_metadata_ip(ip)); assert!(!is_link_local_ip(ip)); } #[test] fn test_detect_trusted_host_gateway_rejects_private_ip() { // Docker bridge (172.17.0.1) and K8s host gateway (192.168.x.x) are // RFC 1918 private addresses — not link-local. Before this fix they // slipped through the old always-blocked guard and received the SSRF // exemption. The new guard (!is_link_local_ip) rejects them, so // connections to these hosts fall through to resolve_and_reject_internal(). for ip in [ IpAddr::V4(Ipv4Addr::new(172, 17, 0, 1)), IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), ] { assert!(!is_cloud_metadata_ip(ip), "{ip} should not be metadata"); assert!(!is_link_local_ip(ip), "{ip} should not be link-local"); // Guard fires — exemption disabled. assert!(!is_link_local_ip(ip), "{ip}: guard must reject"); } } #[test] fn test_detect_trusted_host_gateway_allows_link_local_non_metadata() { // 169.254.1.2 (rootless Podman pasta gateway) IS link-local and is // not a cloud metadata IP — it is the only address class the exemption // is designed for. let ip = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)); assert!(!is_cloud_metadata_ip(ip)); assert!(is_link_local_ip(ip)); // Guard does NOT fire — this IP is eligible for the exemption. assert!(is_link_local_ip(ip)); } // -- parse_hosts_file_for_host: multi-entry / duplicate scenarios -- #[test] fn test_parse_hosts_file_single_entry() { // Normal driver-injected case: exactly one IP for the alias. let contents = "169.254.1.2\thost.openshell.internal host.containers.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!(ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))]); } #[test] fn test_parse_hosts_file_duplicate_same_ip_deduplicated() { // Same IP on two separate lines for the same alias — deduplicated to one. let contents = "169.254.1.2\thost.openshell.internal\n\ 169.254.1.2\thost.openshell.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!( ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))], "identical IPs across lines must be deduplicated" ); } #[test] fn test_parse_hosts_file_multiple_distinct_ips() { // Two distinct IPs for the same alias — both returned, first entry wins // in detect_trusted_host_gateway(), second would cause mismatch rejection // in resolve_and_check_trusted_gateway(). let contents = "169.254.1.2\thost.openshell.internal\n\ 169.254.1.3\thost.openshell.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!(ips.len(), 2, "two distinct IPs must both be returned"); assert_eq!(ips[0], IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))); assert_eq!(ips[1], IpAddr::V4(Ipv4Addr::new(169, 254, 1, 3))); } #[test] fn test_parse_hosts_file_first_entry_wins_on_ambiguity() { // detect_trusted_host_gateway() pins to the first entry via .next(). // Verify the ordering guarantee: first line wins. let contents = "169.254.1.3\thost.openshell.internal\n\ 169.254.1.2\thost.openshell.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!( ips[0], IpAddr::V4(Ipv4Addr::new(169, 254, 1, 3)), "first line must be first in the returned vec" ); } #[test] fn test_parse_hosts_file_ignores_other_aliases_on_same_line() { // An entry with multiple aliases — only the matching alias counts. let contents = "169.254.1.2\thost.containers.internal host.openshell.internal host.docker.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!(ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))]); // Non-matching aliases on the same line do not produce extra entries. let ips2 = parse_hosts_file_for_host(contents, "host.docker.internal"); assert_eq!(ips2, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))]); } #[test] fn test_parse_hosts_file_alias_not_present() { let contents = "127.0.0.1\tlocalhost\n\ ::1\t\tlocalhost\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert!(ips.is_empty()); } #[test] fn test_parse_hosts_file_comment_lines_skipped() { let contents = "# 169.254.1.2 host.openshell.internal\n\ 169.254.1.2\thost.openshell.internal\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); // Commented-out line must not produce an entry. assert_eq!(ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))]); } #[test] fn test_parse_hosts_file_inline_comment_stripped() { // Anything after '#' on a data line is treated as a comment. let contents = "169.254.1.2\thost.openshell.internal # injected by driver\n"; let ips = parse_hosts_file_for_host(contents, "host.openshell.internal"); assert_eq!(ips, vec![IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2))]); } // -- resolve_and_check_trusted_gateway -- #[tokio::test] async fn test_trusted_gateway_allows_link_local_gateway_ip() { // Simulate the rootless Podman pasta case: host.openshell.internal // points to a link-local address which is the only path to the host. let trusted_gw = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)); // We resolve via /etc/hosts (pid=0 falls back to system), so we // exercise the trusted_gw mismatch / cloud-metadata guards directly // against a known resolved address. let addrs = [SocketAddr::new(trusted_gw, 8080)]; // Validate the guard logic inline (mirrors resolve_and_check_trusted_gateway). assert!(!is_cloud_metadata_ip(trusted_gw)); assert_eq!(addrs[0].ip(), trusted_gw); } #[tokio::test] async fn test_trusted_gateway_rejects_cloud_metadata_ip() { let trusted_gw = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)); let metadata_ip = IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)); // Simulate resolution returning the metadata IP. let addrs = [SocketAddr::new(metadata_ip, 80)]; // Cloud metadata check must fire before the trusted_gw equality check. let err: Result<(), String> = if is_cloud_metadata_ip(addrs[0].ip()) { Err(format!( "host resolves to cloud metadata address {}, connection rejected", addrs[0].ip() )) } else if addrs[0].ip() != trusted_gw { Err(format!( "host resolves to {} which does not match trusted host gateway \ {trusted_gw}, connection rejected", addrs[0].ip() )) } else { Ok(()) }; assert!(err.is_err()); assert!( err.unwrap_err().contains("cloud metadata"), "expected cloud-metadata rejection" ); } #[tokio::test] async fn test_trusted_gateway_rejects_mismatched_ip() { let trusted_gw = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)); let other_ip = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)); let addrs = [SocketAddr::new(other_ip, 8080)]; let err: Result<(), String> = if is_cloud_metadata_ip(addrs[0].ip()) { Err("cloud metadata".to_string()) } else if addrs[0].ip() != trusted_gw { Err(format!( "{} does not match trusted host gateway {trusted_gw}", addrs[0].ip() )) } else { Ok(()) }; assert!(err.is_err()); assert!( err.unwrap_err() .contains("does not match trusted host gateway"), "expected mismatch rejection" ); } #[tokio::test] async fn test_trusted_gateway_rejects_control_plane_port() { // Control-plane port check runs before resolution. let result = resolve_and_check_trusted_gateway( "host.openshell.internal", 6443, IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)), 0, ) .await; assert!(result.is_err()); assert!( result.unwrap_err().contains("blocked control-plane port"), "expected control-plane port rejection" ); } #[tokio::test] async fn test_trusted_gateway_rejects_all_control_plane_ports() { let trusted_gw = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)); for &port in BLOCKED_CONTROL_PLANE_PORTS { let result = resolve_and_check_trusted_gateway("host.openshell.internal", port, trusted_gw, 0) .await; assert!( result.is_err(), "port {port} should be blocked by control-plane guard" ); assert!( result.unwrap_err().contains("blocked control-plane port"), "expected control-plane rejection for port {port}" ); } } #[tokio::test] async fn test_trusted_gateway_rejects_loopback_as_trusted_gw() { // Defense-in-depth: even if detect_trusted_host_gateway somehow admitted // a loopback IP, resolve_and_check_trusted_gateway must reject it. // Using an IP literal as the host bypasses DNS and gives a deterministic // resolved address, allowing us to exercise the actual function. let loopback = IpAddr::V4(Ipv4Addr::LOCALHOST); let result = resolve_and_check_trusted_gateway("127.0.0.1", 8080, loopback, 0).await; assert!(result.is_err(), "loopback must be rejected"); let err = result.unwrap_err(); assert!( err.contains("non-link-local"), "expected non-link-local rejection, got: {err}" ); } #[tokio::test] async fn test_trusted_gateway_rejects_unspecified_as_trusted_gw() { // Defense-in-depth: 0.0.0.0 as trusted_gw must be rejected. // IP literal resolves to 0.0.0.0 directly, bypassing DNS. let unspecified = IpAddr::V4(Ipv4Addr::UNSPECIFIED); let result = resolve_and_check_trusted_gateway("0.0.0.0", 8080, unspecified, 0).await; assert!(result.is_err(), "unspecified must be rejected"); let err = result.unwrap_err(); assert!( err.contains("non-link-local"), "expected non-link-local rejection, got: {err}" ); } #[tokio::test] async fn test_trusted_gateway_rejects_ip_literal_mismatch() { // If the requested IP literal doesn't match trusted_gw, the mismatch // guard fires. This exercises the full resolution→validation path. let trusted_gw = IpAddr::V4(Ipv4Addr::new(169, 254, 1, 2)); let other_ip = "10.0.0.1"; // RFC1918, resolves as a literal let result = resolve_and_check_trusted_gateway(other_ip, 8080, trusted_gw, 0).await; assert!(result.is_err(), "IP mismatch must be rejected"); let err = result.unwrap_err(); assert!( err.contains("does not match trusted host gateway"), "expected mismatch rejection, got: {err}" ); } #[tokio::test] async fn test_trusted_gateway_rejects_cloud_metadata_literal() { // Cloud metadata IP as a literal address — must be rejected even when // it matches trusted_gw (which detect_trusted_host_gateway prevents, // but this is the defense-in-depth layer). let metadata = IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)); let result = resolve_and_check_trusted_gateway("169.254.169.254", 80, metadata, 0).await; assert!(result.is_err(), "cloud metadata IP must be rejected"); let err = result.unwrap_err(); assert!( err.contains("cloud metadata"), "expected cloud-metadata rejection, got: {err}" ); } #[tokio::test] async fn test_trusted_gateway_rejects_private_ip_as_trusted_gw() { // Defense-in-depth: a private RFC 1918 IP (e.g. Docker bridge 172.17.0.1) // must be rejected even if it somehow matched trusted_gw. // detect_trusted_host_gateway() already blocks these via !is_link_local_ip(), // but resolve_and_check_trusted_gateway() must enforce the same invariant. let docker_bridge = IpAddr::V4(Ipv4Addr::new(172, 17, 0, 1)); let result = resolve_and_check_trusted_gateway("172.17.0.1", 8080, docker_bridge, 0).await; assert!(result.is_err(), "private RFC 1918 IP must be rejected"); let err = result.unwrap_err(); assert!( err.contains("non-link-local"), "expected non-link-local rejection for private IP, got: {err}" ); } #[tokio::test] async fn test_rejects_localhost_resolution() { let result = resolve_and_reject_internal("localhost", 80, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("internal address"), "expected 'internal address' in error: {err}" ); } #[tokio::test] async fn test_rejects_loopback_ip_literal() { let result = resolve_and_reject_internal("127.0.0.1", 443, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("internal address"), "expected 'internal address' in error: {err}" ); } #[tokio::test] async fn test_rejects_metadata_ip() { let result = resolve_and_reject_internal("169.254.169.254", 80, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("internal address"), "expected 'internal address' in error: {err}" ); } #[tokio::test] async fn test_dns_failure_returns_error() { let result = resolve_and_reject_internal("this-host-does-not-exist.invalid", 80, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("DNS resolution failed"), "expected 'DNS resolution failed' in error: {err}" ); } #[tokio::test] async fn inference_interception_applies_router_header_allowlist() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_addr = listener.local_addr().unwrap(); let upstream_task = tokio::spawn(async move { use crate::l7::inference::{ParseResult, try_parse_http_request}; let (mut upstream, _) = listener.accept().await.unwrap(); let mut buf = Vec::new(); let mut chunk = [0u8; 4096]; loop { let n = upstream.read(&mut chunk).await.unwrap(); assert!(n > 0, "upstream request closed before request completed"); buf.extend_from_slice(&chunk[..n]); match try_parse_http_request(&buf) { ParseResult::Complete(_, consumed) => { upstream .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") .await .unwrap(); return String::from_utf8_lossy(&buf[..consumed]).to_string(); } ParseResult::Incomplete => continue, ParseResult::Invalid(reason) => { panic!("forwarded request should parse cleanly: {reason}"); } } } }); let router = openshell_router::Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); let ctx = InferenceContext::new( patterns, router, vec![openshell_router::config::ResolvedRoute { name: "inference.local".to_string(), endpoint: format!("http://{upstream_addr}"), model: "meta/llama-3.1-8b-instruct".to_string(), api_key: "test-api-key".to_string(), protocols: vec!["openai_chat_completions".to_string()], auth: openshell_router::config::AuthHeader::Bearer, default_headers: vec![], passthrough_headers: vec![ "openai-organization".to_string(), "x-model-id".to_string(), ], timeout: openshell_router::config::DEFAULT_ROUTE_TIMEOUT, model_in_path: false, request_path_override: None, }], vec![], ); let body = r#"{"model":"ignored","messages":[{"role":"user","content":"hi"}]}"#; let request = format!( "POST /v1/chat/completions HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Type: application/json\r\n\ OpenAI-Organization: org_123\r\n\ Authorization: Bearer client-key\r\n\ Cookie: session=abc\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body, ); let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); client_write.write_all(request.as_bytes()).await.unwrap(); client_write.shutdown().await.unwrap(); let mut response = Vec::new(); client_read.read_to_end(&mut response).await.unwrap(); let response_text = String::from_utf8_lossy(&response); assert!(response_text.starts_with("HTTP/1.1 200")); let outcome = server_task.await.unwrap().unwrap(); assert!( matches!(outcome, InferenceOutcome::Routed), "expected Routed outcome, got: {outcome:?}" ); let forwarded = upstream_task.await.unwrap(); let forwarded_lc = forwarded.to_ascii_lowercase(); assert!(forwarded_lc.contains("openai-organization: org_123")); assert!(forwarded_lc.contains("authorization: bearer test-api-key")); assert!(!forwarded_lc.contains("authorization: bearer client-key")); assert!(!forwarded_lc.contains("cookie:")); } fn streaming_inference_route(endpoint: String) -> openshell_router::config::ResolvedRoute { openshell_router::config::ResolvedRoute { name: "inference.local".to_string(), endpoint, model: "meta/llama-3.1-8b-instruct".to_string(), api_key: "test-api-key".to_string(), protocols: vec!["openai_chat_completions".to_string()], auth: openshell_router::config::AuthHeader::Bearer, default_headers: vec![], passthrough_headers: vec![], timeout: openshell_router::config::DEFAULT_ROUTE_TIMEOUT, model_in_path: false, request_path_override: None, } } fn embeddings_inference_route(endpoint: String) -> openshell_router::config::ResolvedRoute { openshell_router::config::ResolvedRoute { name: "inference.local".to_string(), endpoint, model: "text-embedding-3-small".to_string(), api_key: "test-api-key".to_string(), protocols: vec!["openai_embeddings".to_string()], auth: openshell_router::config::AuthHeader::Bearer, default_headers: vec![], passthrough_headers: vec![], timeout: openshell_router::config::DEFAULT_ROUTE_TIMEOUT, model_in_path: false, request_path_override: None, } } /// Embeddings responses are a single buffered JSON object, not an SSE /// stream. They must be framed with `Content-Length` and must never be sent /// through the chunked streaming path, whose truncation handlers would /// append an SSE `proxy_stream_error` frame into the JSON body. #[tokio::test] async fn inference_embeddings_served_buffered_with_content_length() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_addr = listener.local_addr().unwrap(); let upstream_body = r#"{"object":"list","data":[{"object":"embedding","index":0,"embedding":[0.1,0.2]}],"model":"text-embedding-3-small"}"#; let upstream_task = tokio::spawn(async move { let (mut upstream, _) = listener.accept().await.unwrap(); read_forwarded_inference_request(&mut upstream).await; // Buffered upstream response with Content-Length (no chunked TE). let resp = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", upstream_body.len(), upstream_body, ); upstream.write_all(resp.as_bytes()).await.unwrap(); }); let router = openshell_router::Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); let ctx = InferenceContext::new( patterns, router, vec![embeddings_inference_route(format!( "http://{upstream_addr}" ))], vec![], ); let body = r#"{"model":"text-embedding-3-small","input":"hello"}"#; let request = format!( "POST /v1/embeddings HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Type: application/json\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body, ); let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); client_write.write_all(request.as_bytes()).await.unwrap(); client_write.shutdown().await.unwrap(); let mut response = Vec::new(); client_read.read_to_end(&mut response).await.unwrap(); let response = String::from_utf8(response).unwrap(); server_task.await.unwrap().unwrap(); upstream_task.await.unwrap(); assert!( response.starts_with("HTTP/1.1 200 OK\r\n"), "expected buffered 200 response, got: {response}" ); let lower = response.to_ascii_lowercase(); assert!( lower.contains("content-length:"), "embeddings response must be Content-Length framed, got: {response}" ); assert!( !lower.contains("transfer-encoding: chunked"), "embeddings response must NOT be chunked, got: {response}" ); assert!( !response.contains("proxy_stream_error"), "embeddings response must not carry an SSE error frame, got: {response}" ); assert!( response.contains(r#""object":"list""#), "embeddings JSON body must be forwarded intact, got: {response}" ); } fn model_discovery_inference_route( endpoint: String, ) -> openshell_router::config::ResolvedRoute { openshell_router::config::ResolvedRoute { name: "inference.local".to_string(), endpoint, model: "text-embedding-3-small".to_string(), api_key: "test-api-key".to_string(), protocols: vec!["model_discovery".to_string()], auth: openshell_router::config::AuthHeader::Bearer, default_headers: vec![], passthrough_headers: vec![], timeout: openshell_router::config::DEFAULT_ROUTE_TIMEOUT, model_in_path: false, request_path_override: None, } } /// `GET /v1/models` (model discovery) returns one JSON object — a model /// list — exactly like embeddings. It must be served buffered with /// `Content-Length`, never through the chunked streaming path whose /// truncation handlers would append an SSE `proxy_stream_error` frame into /// the JSON body. This guards the framing classification for the protocol. #[tokio::test] async fn inference_model_discovery_served_buffered_with_content_length() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_addr = listener.local_addr().unwrap(); let upstream_body = r#"{"object":"list","data":[{"id":"text-embedding-3-small","object":"model"}]}"#; let upstream_task = tokio::spawn(async move { let (mut upstream, _) = listener.accept().await.unwrap(); read_forwarded_inference_request(&mut upstream).await; let resp = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", upstream_body.len(), upstream_body, ); upstream.write_all(resp.as_bytes()).await.unwrap(); }); let router = openshell_router::Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); let ctx = InferenceContext::new( patterns, router, vec![model_discovery_inference_route(format!( "http://{upstream_addr}" ))], vec![], ); // GET model discovery carries no request body. let request = "GET /v1/models HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Length: 0\r\n\r\n" .to_string(); let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); client_write.write_all(request.as_bytes()).await.unwrap(); client_write.shutdown().await.unwrap(); let mut response = Vec::new(); client_read.read_to_end(&mut response).await.unwrap(); let response = String::from_utf8(response).unwrap(); server_task.await.unwrap().unwrap(); upstream_task.await.unwrap(); assert!( response.starts_with("HTTP/1.1 200 OK\r\n"), "expected buffered 200 response, got: {response}" ); let lower = response.to_ascii_lowercase(); assert!( lower.contains("content-length:"), "model discovery response must be Content-Length framed, got: {response}" ); assert!( !lower.contains("transfer-encoding: chunked"), "model discovery response must NOT be chunked, got: {response}" ); assert!( !response.contains("proxy_stream_error"), "model discovery response must not carry an SSE error frame, got: {response}" ); assert!( response.contains(r#""object":"list""#), "model discovery JSON body must be forwarded intact, got: {response}" ); } /// `GET /v1/models/{id}` (model discovery glob) must forward the model id in /// the path through the buffered path with the id intact, never streamed. #[tokio::test] async fn inference_model_discovery_glob_path_served_buffered() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_addr = listener.local_addr().unwrap(); let upstream_body = r#"{"id":"gpt-4.1","object":"model"}"#; let upstream_task = tokio::spawn(async move { let (mut upstream, _) = listener.accept().await.unwrap(); let forwarded = read_forwarded_request_line(&mut upstream).await; let resp = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", upstream_body.len(), upstream_body, ); upstream.write_all(resp.as_bytes()).await.unwrap(); forwarded }); let router = openshell_router::Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); let ctx = InferenceContext::new( patterns, router, vec![model_discovery_inference_route(format!( "http://{upstream_addr}" ))], vec![], ); let request = "GET /v1/models/gpt-4.1 HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Length: 0\r\n\r\n" .to_string(); let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); client_write.write_all(request.as_bytes()).await.unwrap(); client_write.shutdown().await.unwrap(); let mut response = Vec::new(); client_read.read_to_end(&mut response).await.unwrap(); let response = String::from_utf8(response).unwrap(); server_task.await.unwrap().unwrap(); let (method, forwarded_path) = upstream_task.await.unwrap(); assert_eq!(method, "GET"); assert_eq!( forwarded_path, "/v1/models/gpt-4.1", "the model id in the glob path must be forwarded intact" ); let lower = response.to_ascii_lowercase(); assert!( response.starts_with("HTTP/1.1 200 OK\r\n") && lower.contains("content-length:") && !lower.contains("transfer-encoding: chunked") && !response.contains("proxy_stream_error"), "glob model discovery must be buffered and Content-Length framed, got: {response}" ); } /// A failed model-discovery upstream must produce a buffered, Content-Length /// framed JSON error, never a chunked SSE `proxy_stream_error` frame. #[tokio::test] async fn inference_model_discovery_error_served_buffered() { // A port with no listener so the upstream connection is refused. let dead_addr = { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); drop(listener); addr }; let router = openshell_router::Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); let ctx = InferenceContext::new( patterns, router, vec![model_discovery_inference_route(format!( "http://{dead_addr}" ))], vec![], ); let request = "GET /v1/models HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Length: 0\r\n\r\n" .to_string(); let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); client_write.write_all(request.as_bytes()).await.unwrap(); client_write.shutdown().await.unwrap(); let mut response = Vec::new(); client_read.read_to_end(&mut response).await.unwrap(); let response = String::from_utf8(response).unwrap(); server_task.await.unwrap().unwrap(); let lower = response.to_ascii_lowercase(); assert!( response.starts_with("HTTP/1.1 5"), "a refused upstream should yield a 5xx, got: {response}" ); assert!( lower.contains("content-length:") && !lower.contains("transfer-encoding: chunked") && !response.contains("proxy_stream_error"), "buffered model-discovery error must be Content-Length framed JSON, got: {response}" ); assert!( response.contains("error"), "error response should carry a JSON error body, got: {response}" ); } async fn read_forwarded_inference_request(stream: &mut S) { use crate::l7::inference::{ParseResult, try_parse_http_request}; let mut buf = Vec::new(); let mut chunk = [0u8; 4096]; loop { let n = stream.read(&mut chunk).await.unwrap(); assert!(n > 0, "upstream request closed before completion"); buf.extend_from_slice(&chunk[..n]); match try_parse_http_request(&buf) { ParseResult::Complete(_, _) => return, ParseResult::Incomplete => continue, ParseResult::Invalid(reason) => { panic!("forwarded request should parse cleanly: {reason}"); } } } } /// Like [`read_forwarded_inference_request`] but returns the forwarded /// request line (method, path) so a test can assert the upstream URL path. async fn read_forwarded_request_line(stream: &mut S) -> (String, String) { use crate::l7::inference::{ParseResult, try_parse_http_request}; let mut buf = Vec::new(); let mut chunk = [0u8; 4096]; loop { let n = stream.read(&mut chunk).await.unwrap(); assert!(n > 0, "upstream request closed before completion"); buf.extend_from_slice(&chunk[..n]); match try_parse_http_request(&buf) { ParseResult::Complete(req, _) => return (req.method, req.path), ParseResult::Incomplete => continue, ParseResult::Invalid(reason) => { panic!("forwarded request should parse cleanly: {reason}"); } } } } async fn run_live_streaming_inference(serve_upstream: F) -> String where F: FnOnce(TcpStream) -> Fut + Send + 'static, Fut: Future + Send + 'static, { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_addr = listener.local_addr().unwrap(); let upstream_task = tokio::spawn(async move { let (mut upstream, _) = listener.accept().await.unwrap(); read_forwarded_inference_request(&mut upstream).await; serve_upstream(upstream).await; }); let router = openshell_router::Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); let ctx = InferenceContext::new( patterns, router, vec![streaming_inference_route(format!("http://{upstream_addr}"))], vec![], ); let body = r#"{"model":"ignored","messages":[{"role":"user","content":"hi"}]}"#; let request = format!( "POST /v1/chat/completions HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Type: application/json\r\n\ Accept: text/event-stream\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body, ); let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); client_write.write_all(request.as_bytes()).await.unwrap(); client_write.shutdown().await.unwrap(); let mut response = Vec::new(); client_read.read_to_end(&mut response).await.unwrap(); let outcome = server_task.await.unwrap().unwrap(); assert!( matches!(outcome, InferenceOutcome::Routed), "expected Routed outcome, got: {outcome:?}" ); upstream_task.await.unwrap(); String::from_utf8(response).unwrap() } fn assert_streaming_sse_error(response: &str, message: &str) { assert!( response.starts_with("HTTP/1.1 200 OK\r\n"), "expected successful streaming response, got: {response}" ); assert!( response .to_ascii_lowercase() .contains("transfer-encoding: chunked"), "expected chunked streaming response, got: {response}" ); assert!( response.contains("\"type\":\"proxy_stream_error\""), "expected proxy_stream_error SSE event, got: {response}" ); assert!( response.contains(&format!("\"message\":\"{message}\"")), "expected SSE message {message:?}, got: {response}" ); assert!( response.ends_with("0\r\n\r\n"), "streaming response must end with chunked terminator, got: {response}" ); } #[tokio::test] async fn inference_stream_byte_limit_injects_sse_error() { let response = run_live_streaming_inference(|mut upstream| async move { use crate::l7::inference::{format_chunk, format_chunk_terminator}; upstream .write_all( b"HTTP/1.1 200 OK\r\n\ Content-Type: text/event-stream\r\n\ Transfer-Encoding: chunked\r\n\r\n", ) .await .unwrap(); let body = vec![b'a'; MAX_STREAMING_BODY + 1]; let _ = upstream.write_all(&format_chunk(&body)).await; let _ = upstream.write_all(format_chunk_terminator()).await; }) .await; assert_streaming_sse_error( &response, "response truncated: exceeded maximum streaming body size", ); } #[tokio::test] async fn inference_stream_upstream_read_error_injects_sse_error() { let response = run_live_streaming_inference(|mut upstream| async move { upstream .write_all( b"HTTP/1.1 200 OK\r\n\ Content-Type: text/event-stream\r\n\ Content-Length: 64\r\n\r\n\ partial", ) .await .unwrap(); }) .await; assert!( response.contains("partial"), "expected initial upstream bytes before truncation, got: {response}" ); assert_streaming_sse_error(&response, "response truncated: upstream read error"); } #[tokio::test] async fn inference_stream_idle_timeout_injects_sse_error() { let response = run_live_streaming_inference(|mut upstream| async move { upstream .write_all( b"HTTP/1.1 200 OK\r\n\ Content-Type: text/event-stream\r\n\ Transfer-Encoding: chunked\r\n\r\n", ) .await .unwrap(); tokio::time::sleep(CHUNK_IDLE_TIMEOUT + std::time::Duration::from_millis(50)).await; }) .await; assert_streaming_sse_error(&response, "response truncated: chunk idle timeout exceeded"); } // -- router_error_to_http -- #[test] fn router_error_route_not_found_maps_to_400() { let err = openshell_router::RouterError::RouteNotFound("local".into()); let (status, msg) = router_error_to_http(&err); assert_eq!(status, 400); assert_eq!(msg, "no inference route configured"); // SEC-008: must NOT leak the route hint to sandboxed code assert!(!msg.contains("local")); } #[test] fn router_error_no_compatible_route_maps_to_400() { let err = openshell_router::RouterError::NoCompatibleRoute("anthropic_messages".into()); let (status, msg) = router_error_to_http(&err); assert_eq!(status, 400); assert_eq!(msg, "no compatible inference route available"); // SEC-008: must NOT leak the protocol name to sandboxed code assert!(!msg.contains("anthropic_messages")); } #[test] fn router_error_unauthorized_maps_to_401() { let err = openshell_router::RouterError::Unauthorized("bad token from 10.0.0.5:8080".into()); let (status, msg) = router_error_to_http(&err); assert_eq!(status, 401); assert_eq!(msg, "unauthorized"); // SEC-008: must NOT leak upstream details to sandboxed code assert!(!msg.contains("10.0.0.5")); } #[test] fn router_error_upstream_unavailable_maps_to_503() { let err = openshell_router::RouterError::UpstreamUnavailable( "connection refused to 10.0.0.5:8080".into(), ); let (status, msg) = router_error_to_http(&err); assert_eq!(status, 503); assert_eq!(msg, "inference service unavailable"); // SEC-008: must NOT leak upstream address to sandboxed code assert!(!msg.contains("10.0.0.5")); } #[test] fn router_error_upstream_protocol_maps_to_502() { let err = openshell_router::RouterError::UpstreamProtocol( "TLS handshake failed for nim.internal.svc:443".into(), ); let (status, msg) = router_error_to_http(&err); assert_eq!(status, 502); assert_eq!(msg, "inference service error"); // SEC-008: must NOT leak internal hostnames to sandboxed code assert!(!msg.contains("nim.internal")); } #[test] fn router_error_internal_maps_to_502() { let err = openshell_router::RouterError::Internal( "failed to read /etc/openshell/routes.json".into(), ); let (status, msg) = router_error_to_http(&err); assert_eq!(status, 502); assert_eq!(msg, "inference service error"); // SEC-008: must NOT leak file paths to sandboxed code assert!(!msg.contains("/etc/openshell")); } #[test] fn sanitize_response_headers_strips_hop_by_hop() { let headers = vec![ ("transfer-encoding".to_string(), "chunked".to_string()), ("content-length".to_string(), "128".to_string()), ("connection".to_string(), "keep-alive".to_string()), ("content-type".to_string(), "text/event-stream".to_string()), ("cache-control".to_string(), "no-cache".to_string()), ]; let kept = sanitize_inference_response_headers(headers); assert!( kept.iter() .all(|(k, _)| !k.eq_ignore_ascii_case("transfer-encoding")), "transfer-encoding should be stripped" ); assert!( kept.iter() .all(|(k, _)| !k.eq_ignore_ascii_case("content-length")), "content-length should be stripped" ); assert!( kept.iter() .all(|(k, _)| !k.eq_ignore_ascii_case("connection")), "connection should be stripped" ); assert!( kept.iter() .any(|(k, _)| k.eq_ignore_ascii_case("content-type")), "content-type should be preserved" ); assert!( kept.iter() .any(|(k, _)| k.eq_ignore_ascii_case("cache-control")), "cache-control should be preserved" ); } // -- is_always_blocked_ip -- #[test] fn test_always_blocked_loopback_v4() { assert!(is_always_blocked_ip(IpAddr::V4(Ipv4Addr::LOCALHOST))); assert!(is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new( 127, 0, 0, 2 )))); } #[test] fn test_always_blocked_link_local_v4() { assert!(is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new( 169, 254, 169, 254 )))); assert!(is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new( 169, 254, 0, 1 )))); } #[test] fn test_always_blocked_loopback_v6() { assert!(is_always_blocked_ip(IpAddr::V6(Ipv6Addr::LOCALHOST))); } #[test] fn test_always_blocked_link_local_v6() { assert!(is_always_blocked_ip(IpAddr::V6(Ipv6Addr::new( 0xfe80, 0, 0, 0, 0, 0, 0, 1 )))); } #[test] fn test_always_blocked_ipv4_unspecified() { assert!(is_always_blocked_ip(IpAddr::V4(Ipv4Addr::UNSPECIFIED))); } #[test] fn test_always_blocked_ipv6_unspecified() { assert!(is_always_blocked_ip(IpAddr::V6(Ipv6Addr::UNSPECIFIED))); } #[test] fn test_always_blocked_ipv4_mapped_v6_loopback() { let v6 = Ipv4Addr::LOCALHOST.to_ipv6_mapped(); assert!(is_always_blocked_ip(IpAddr::V6(v6))); } #[test] fn test_always_blocked_ipv4_mapped_v6_link_local() { let v6 = Ipv4Addr::new(169, 254, 169, 254).to_ipv6_mapped(); assert!(is_always_blocked_ip(IpAddr::V6(v6))); } #[test] fn test_always_blocked_allows_rfc1918() { // RFC 1918 addresses should NOT be always-blocked (they're allowed // when allowed_ips is configured) assert!(!is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new( 10, 0, 0, 1 )))); assert!(!is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new( 172, 16, 0, 1 )))); assert!(!is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new( 192, 168, 0, 1 )))); } #[test] fn test_always_blocked_allows_public() { assert!(!is_always_blocked_ip(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); assert!(!is_always_blocked_ip(IpAddr::V6(Ipv6Addr::new( 0x2001, 0x4860, 0x4860, 0, 0, 0, 0, 0x8888 )))); } // -- parse_allowed_ips -- #[test] fn test_parse_cidr_notation() { let raw = vec!["10.0.5.0/24".to_string()]; let nets = parse_allowed_ips(&raw).unwrap(); assert_eq!(nets.len(), 1); assert!(nets[0].contains(&IpAddr::V4(Ipv4Addr::new(10, 0, 5, 1)))); assert!(!nets[0].contains(&IpAddr::V4(Ipv4Addr::new(10, 0, 6, 1)))); } #[test] fn test_parse_exact_ip() { let raw = vec!["10.0.5.20".to_string()]; let nets = parse_allowed_ips(&raw).unwrap(); assert_eq!(nets.len(), 1); assert!(nets[0].contains(&IpAddr::V4(Ipv4Addr::new(10, 0, 5, 20)))); assert!(!nets[0].contains(&IpAddr::V4(Ipv4Addr::new(10, 0, 5, 21)))); } #[test] fn test_parse_multiple_entries() { let raw = vec![ "10.0.0.0/8".to_string(), "172.16.0.0/12".to_string(), "192.168.1.1".to_string(), ]; let nets = parse_allowed_ips(&raw).unwrap(); assert_eq!(nets.len(), 3); } #[test] fn test_parse_invalid_entry_errors() { let raw = vec!["not-an-ip".to_string()]; let result = parse_allowed_ips(&raw); assert!(result.is_err()); assert!(result.unwrap_err().contains("invalid CIDR/IP")); } #[test] fn test_parse_mixed_valid_invalid_errors() { let raw = vec!["10.0.5.0/24".to_string(), "garbage".to_string()]; let result = parse_allowed_ips(&raw); assert!(result.is_err()); } #[tokio::test] async fn test_resolve_check_allowed_ips_blocks_loopback() { // Construct nets directly (parse_allowed_ips now rejects always-blocked). let nets = vec!["127.0.0.0/8".parse::().unwrap()]; let result = resolve_and_check_allowed_ips("127.0.0.1", 80, &nets, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("always-blocked"), "expected 'always-blocked' in error: {err}" ); } #[tokio::test] async fn test_resolve_check_allowed_ips_blocks_metadata() { // Construct nets directly (parse_allowed_ips now rejects always-blocked). let nets = vec!["169.254.0.0/16".parse::().unwrap()]; let result = resolve_and_check_allowed_ips("169.254.169.254", 80, &nets, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("always-blocked"), "expected 'always-blocked' in error: {err}" ); } #[tokio::test] async fn test_resolve_check_allowed_ips_blocks_unspecified() { // Construct nets directly (parse_allowed_ips now rejects always-blocked). let nets = vec!["0.0.0.0/0".parse::().unwrap()]; let result = resolve_and_check_allowed_ips("0.0.0.0", 80, &nets, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("always-blocked"), "expected 'always-blocked' in error: {err}" ); } #[tokio::test] async fn test_resolve_check_allowed_ips_rejects_outside_allowlist() { // 8.8.8.8 resolves to a public IP which is NOT in 10.0.0.0/8 let nets = parse_allowed_ips(&["10.0.0.0/8".to_string()]).unwrap(); let result = resolve_and_check_allowed_ips("dns.google", 443, &nets, 0).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( err.contains("not in allowed_ips"), "expected 'not in allowed_ips' in error: {err}" ); } // --- SEC-005: CIDR breadth warning and control-plane port blocklist --- #[tokio::test] async fn test_resolve_check_allowed_ips_blocks_control_plane_ports() { // Use a public CIDR (parse_allowed_ips now rejects 0.0.0.0/0). let nets = parse_allowed_ips(&["8.8.8.0/24".to_string()]).unwrap(); // K8s API server port let result = resolve_and_check_allowed_ips("8.8.8.8", 6443, &nets, 0).await; assert!(result.is_err()); assert!(result.unwrap_err().contains("blocked control-plane port")); // etcd client port let result = resolve_and_check_allowed_ips("8.8.8.8", 2379, &nets, 0).await; assert!(result.is_err()); assert!(result.unwrap_err().contains("blocked control-plane port")); // kubelet API port let result = resolve_and_check_allowed_ips("8.8.8.8", 10250, &nets, 0).await; assert!(result.is_err()); assert!(result.unwrap_err().contains("blocked control-plane port")); } #[tokio::test] async fn test_resolve_check_allowed_ips_allows_non_control_plane_ports() { // Port 443 should not be blocked by the control-plane port list let nets = parse_allowed_ips(&["8.8.8.0/24".to_string()]).unwrap(); let result = resolve_and_check_allowed_ips("8.8.8.8", 443, &nets, 0).await; assert!(result.is_ok()); } #[test] fn test_parse_allowed_ips_broad_cidr_is_accepted() { // Broad CIDRs are accepted (just warned about) -- design trade-off let result = parse_allowed_ips(&["10.0.0.0/8".to_string()]); assert!(result.is_ok()); } // --- parse_allowed_ips: always-blocked rejection tests --- #[test] fn test_parse_allowed_ips_rejects_loopback_cidr() { let result = parse_allowed_ips(&["127.0.0.0/8".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_rejects_link_local_cidr() { let result = parse_allowed_ips(&["169.254.0.0/16".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_rejects_unspecified() { let result = parse_allowed_ips(&["0.0.0.0".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_rejects_single_loopback_ip() { let result = parse_allowed_ips(&["127.0.0.1".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_rejects_single_metadata_ip() { let result = parse_allowed_ips(&["169.254.169.254".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_rejects_wildcard_cidr() { let result = parse_allowed_ips(&["0.0.0.0/0".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_mixed_valid_and_blocked() { // A blocked entry taints the whole batch. let result = parse_allowed_ips(&["10.0.5.0/24".to_string(), "127.0.0.1".to_string()]); assert!(result.is_err()); assert!(result.unwrap_err().contains("always-blocked")); } #[test] fn test_parse_allowed_ips_accepts_rfc1918() { let result = parse_allowed_ips(&["10.0.5.0/24".to_string(), "192.168.1.0/24".to_string()]); assert!(result.is_ok()); } // --- implicit_allowed_ips_for_ip_host: always-blocked skip tests --- #[test] fn test_implicit_allowed_ips_skips_loopback() { let result = implicit_allowed_ips_for_ip_host("127.0.0.1"); assert!(result.is_empty()); } #[test] fn test_implicit_allowed_ips_skips_link_local() { let result = implicit_allowed_ips_for_ip_host("169.254.169.254"); assert!(result.is_empty()); } #[test] fn test_implicit_allowed_ips_skips_unspecified() { let result = implicit_allowed_ips_for_ip_host("0.0.0.0"); assert!(result.is_empty()); } #[test] fn test_implicit_allowed_ips_allows_rfc1918() { let result = implicit_allowed_ips_for_ip_host("10.0.5.20"); assert_eq!(result, vec!["10.0.5.20"]); } // --- extract_host_from_uri tests --- #[test] fn test_extract_host_from_http_uri() { assert_eq!( extract_host_from_uri("http://example.com/path"), "example.com" ); } #[test] fn test_extract_host_from_https_uri() { assert_eq!( extract_host_from_uri("https://api.openai.com/v1/chat/completions"), "api.openai.com" ); } #[test] fn test_extract_host_from_uri_with_port() { assert_eq!( extract_host_from_uri("http://example.com:8080/path"), "example.com" ); } #[test] fn test_extract_host_from_uri_ipv6() { assert_eq!(extract_host_from_uri("http://[::1]:8080/path"), "[::1]"); } #[test] fn test_extract_host_from_uri_no_path() { assert_eq!(extract_host_from_uri("http://example.com"), "example.com"); } #[test] fn test_extract_host_from_uri_empty() { assert_eq!(extract_host_from_uri(""), "unknown"); } #[test] fn test_extract_host_from_uri_malformed() { // Gracefully handles garbage input let result = extract_host_from_uri("not-a-uri"); assert!(!result.is_empty()); } // --- parse_proxy_uri tests --- #[test] fn test_parse_proxy_uri_standard() { let (scheme, host, port, path) = parse_proxy_uri("http://10.86.8.223:8000/screenshot/").unwrap(); assert_eq!(scheme, "http"); assert_eq!(host, "10.86.8.223"); assert_eq!(port, 8000); assert_eq!(path, "/screenshot/"); } #[test] fn test_parse_proxy_uri_default_port() { let (scheme, host, port, path) = parse_proxy_uri("http://example.com/path").unwrap(); assert_eq!(scheme, "http"); assert_eq!(host, "example.com"); assert_eq!(port, 80); assert_eq!(path, "/path"); } #[test] fn test_parse_proxy_uri_https_default_port() { let (scheme, host, port, path) = parse_proxy_uri("https://api.example.com/v1/chat").unwrap(); assert_eq!(scheme, "https"); assert_eq!(host, "api.example.com"); assert_eq!(port, 443); assert_eq!(path, "/v1/chat"); } #[test] fn test_parse_proxy_uri_missing_path() { let (_, host, port, path) = parse_proxy_uri("http://10.0.0.1:9090").unwrap(); assert_eq!(host, "10.0.0.1"); assert_eq!(port, 9090); assert_eq!(path, "/"); } #[test] fn test_parse_proxy_uri_with_query() { let (_, _, _, path) = parse_proxy_uri("http://host:80/api?key=val&foo=bar").unwrap(); assert_eq!(path, "/api?key=val&foo=bar"); } #[test] fn test_parse_proxy_uri_ipv6() { let (_, host, port, path) = parse_proxy_uri("http://[::1]:8080/test").unwrap(); assert_eq!(host, "::1"); assert_eq!(port, 8080); assert_eq!(path, "/test"); } #[test] fn test_parse_proxy_uri_ipv6_default_port() { let (_, host, port, path) = parse_proxy_uri("http://[fe80::1]/path").unwrap(); assert_eq!(host, "fe80::1"); assert_eq!(port, 80); assert_eq!(path, "/path"); } #[test] fn test_parse_proxy_uri_missing_scheme() { let result = parse_proxy_uri("example.com/path"); assert!(result.is_err()); } #[test] fn test_parse_proxy_uri_empty_host() { let result = parse_proxy_uri("http:///path"); assert!(result.is_err()); } // -- parse_target: CONNECT target parser regression tests -- #[test] fn test_parse_target_valid_baseline() { let (host, port) = parse_target("example.com:443").unwrap(); assert_eq!(host, "example.com"); assert_eq!(port, 443); } #[test] fn test_parse_target_preserves_case() { let (host, port) = parse_target("EXAMPLE.COM:443").unwrap(); assert_eq!(host, "EXAMPLE.COM", "parse_target should preserve case"); assert_eq!(port, 443); } #[test] fn test_parse_target_accepts_empty_host() { let (host, port) = parse_target(":443").unwrap(); assert!(host.is_empty(), "empty host accepted without validation"); assert_eq!(port, 443); } #[test] fn test_parse_target_nul_byte_passes_through() { let (host, _) = parse_target("evil.com\0.safe.com:443").unwrap(); assert_eq!( host, "evil.com\0.safe.com", "NUL byte not stripped or rejected" ); } #[test] fn test_parse_target_control_char_passes_through() { let (host, _) = parse_target("evil\x01.com:443").unwrap(); assert!( host.contains('\x01'), "control characters pass through without validation" ); } #[test] fn test_parse_target_percent_encoded_dot_is_literal() { let (host, _) = parse_target("evil%2ecom:443").unwrap(); assert_eq!( host, "evil%2ecom", "percent-encoded dot not decoded — literal %2e in host" ); } #[test] fn test_parse_target_percent_encoded_nul_is_literal() { let (host, _) = parse_target("evil%00.safe.com:443").unwrap(); assert_eq!( host, "evil%00.safe.com", "percent-encoded NUL not decoded — literal %00 in host" ); } #[test] fn test_parse_target_rejects_missing_port_separator() { assert!( parse_target("hostonly").is_err(), "missing colon should be rejected" ); } #[test] fn test_parse_target_rejects_non_numeric_port() { assert!( parse_target("host:notaport").is_err(), "non-numeric port should be rejected" ); } #[test] fn test_parse_target_rejects_port_overflow() { assert!( parse_target("host:65536").is_err(), "port > 65535 should be rejected by u16 parse" ); } #[test] fn test_parse_target_accepts_port_zero() { let (_, port) = parse_target("host:0").unwrap(); assert_eq!(port, 0); } #[test] fn test_parse_target_accepts_port_max() { let (_, port) = parse_target("host:65535").unwrap(); assert_eq!(port, 65535); } #[test] fn test_parse_target_bracket_chars_pass_through() { let (host, _) = parse_target("a]b[c:443").unwrap(); assert_eq!(host, "a]b[c", "brackets pass through without validation"); } #[test] fn test_parse_target_oversized_hostname_accepted() { let long_host = "a".repeat(254); let target = format!("{long_host}:443"); let (host, _) = parse_target(&target).unwrap(); assert_eq!( host.len(), 254, "hostname exceeding DNS 253-char limit not rejected" ); } #[test] fn test_parse_target_backslash_passes_through() { let (host, _) = parse_target("evil.com\\..safe.com:443").unwrap(); assert!( host.contains('\\'), "backslash passes through without validation" ); } #[test] fn test_parse_target_slash_passes_through() { let (host, _) = parse_target("evil.com/../safe.com:443").unwrap(); assert!( host.contains('/'), "forward slash passes through without validation" ); } #[test] fn test_parse_target_extra_colon_fails_port_parse() { assert!( parse_target("host:80:extra").is_err(), "trailing content after port should fail u16 parse" ); } #[test] fn test_parse_target_ipv6_bracket_notation_fails() { assert!( parse_target("[::1]:443").is_err(), "split_once splits at first colon inside brackets — port parse fails" ); } // -- parse_proxy_uri: hostname parser regression tests -- #[test] fn test_parse_proxy_uri_nul_byte_in_host() { let (_, host, port, _) = parse_proxy_uri("http://evil.com\0.safe.com:80/path").unwrap(); assert_eq!( host, "evil.com\0.safe.com", "NUL byte not stripped or rejected in forward proxy URI" ); assert_eq!(port, 80); } #[test] fn test_parse_proxy_uri_control_char_in_host() { let (_, host, _, _) = parse_proxy_uri("http://evil\x01.com:80/").unwrap(); assert!( host.contains('\x01'), "control characters pass through without validation" ); } #[test] fn test_parse_proxy_uri_percent_encoded_dot_in_host() { let (_, host, _, _) = parse_proxy_uri("http://evil%2ecom:80/").unwrap(); assert_eq!( host, "evil%2ecom", "percent-encoded dot not decoded — literal %2e in host" ); } #[test] fn test_parse_proxy_uri_oversized_hostname() { let long_host = "a".repeat(254); let uri = format!("http://{long_host}:80/"); let (_, host, _, _) = parse_proxy_uri(&uri).unwrap(); assert_eq!( host.len(), 254, "hostname exceeding DNS 253-char limit not rejected" ); } // --- rewrite_forward_request tests --- #[tokio::test] async fn forward_proxy_injects_token_grant_before_rewriting_request() { let (ctx, fixture) = forward_token_grant_context(Ok("grant-token")); let raw = b"GET http://api.example.test:8080/v1/projects HTTP/1.1\r\nHost: api.example.test:8080\r\nAuthorization: Bearer stale-token\r\nConnection: close\r\n\r\n".to_vec(); let with_token = inject_token_grant_for_forward_request("GET", "/v1/projects", raw, &ctx) .await .expect("forward token grant should inject"); let rewritten = rewrite_forward_request( &with_token, with_token.len(), "/v1/projects", "api.example.test:8080", None, false, ) .expect("forward request should rewrite"); let rewritten = String::from_utf8_lossy(&rewritten); assert!(rewritten.starts_with("GET /v1/projects HTTP/1.1\r\n")); assert!(rewritten.contains("Authorization: Bearer grant-token\r\n")); assert!(!rewritten.contains("stale-token")); assert_eq!(authorization_header_count(&rewritten), 1); fixture.assert_one_request("api.example.test\t8080\t/v1/**\tprovider:access_token"); } #[tokio::test] async fn forward_proxy_token_grant_failure_returns_error_before_rewrite() { let (ctx, fixture) = forward_token_grant_context(Err("oauth unavailable")); let raw = b"GET http://api.example.test:8080/v1/projects HTTP/1.1\r\nHost: api.example.test:8080\r\nConnection: close\r\n\r\n".to_vec(); let err = inject_token_grant_for_forward_request("GET", "/v1/projects", raw, &ctx) .await .expect_err("forward token grant failure should stop request rewriting"); assert!(err.to_string().contains("Token grant failed")); assert!(err.to_string().contains("oauth unavailable")); fixture.assert_one_request("api.example.test\t8080\t/v1/**\tprovider:access_token"); } #[test] fn test_rewrite_get_request() { let raw = b"GET http://10.0.0.1:8000/api HTTP/1.1\r\nHost: 10.0.0.1:8000\r\nAccept: */*\r\n\r\n"; let result = rewrite_forward_request(raw, raw.len(), "/api", "10.0.0.1:8000", None, false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); assert!(result_str.starts_with("GET /api HTTP/1.1\r\n")); assert!(result_str.contains("Host: 10.0.0.1:8000")); assert!(result_str.contains("Connection: close")); assert!(result_str.contains("Via: 1.1 openshell-sandbox")); } #[test] fn canonical_forward_authority_formats_ports_and_ipv6() { for (uri, expected) in [ ("http://API.EXAMPLE.TEST/path", "api.example.test"), ("http://api.example.test:8080/path", "api.example.test:8080"), ("http://[2001:DB8::1]/path", "[2001:db8::1]"), ("http://[2001:DB8::1]:8080/path", "[2001:db8::1]:8080"), ] { let (_, host, port, _) = parse_proxy_uri(uri).expect("parse absolute target"); assert_eq!(canonical_forward_authority(&host, port), expected); } } #[test] fn forward_host_header_is_replaced_from_absolute_target() { let raw = b"POST http://allowed.example.test:8080/api HTTP/1.1\r\n\ Host: disallowed.example.test\r\n\ hOsT: second.example.test\r\n\ Content-Length: 4\r\n\r\nbody"; let authority = "allowed.example.test:8080"; let canonical = canonicalize_forward_host_header(raw, authority) .expect("canonicalize received Host fields"); let canonical = String::from_utf8(canonical).expect("canonical request is UTF-8"); let host_fields: Vec<_> = canonical .split("\r\n") .skip(1) .take_while(|line| !line.is_empty()) .filter(|line| { line.split_once(':') .is_some_and(|(name, _)| name.eq_ignore_ascii_case("host")) }) .collect(); assert_eq!(host_fields, ["Host: allowed.example.test:8080"]); assert!(!canonical.contains("disallowed.example.test")); assert!(!canonical.contains("second.example.test")); assert!(canonical.ends_with("\r\n\r\nbody")); let rewritten = rewrite_forward_request(raw, raw.len(), "/api", authority, None, false) .expect("final rewrite enforces canonical Host"); let rewritten = String::from_utf8(rewritten).expect("rewritten request is UTF-8"); assert_eq!( rewritten .split("\r\n") .filter(|line| { line.split_once(':') .is_some_and(|(name, _)| name.eq_ignore_ascii_case("host")) }) .collect::>(), ["Host: allowed.example.test:8080"] ); assert!(!rewritten.contains("disallowed.example.test")); assert!(!rewritten.contains("second.example.test")); } #[test] fn forward_host_header_is_generated_when_missing() { let raw = b"GET http://allowed.example.test/api HTTP/1.1\r\nAccept: */*\r\n\r\n"; let canonical = canonicalize_forward_host_header(raw, "allowed.example.test") .expect("generate missing Host field"); let canonical = String::from_utf8(canonical).expect("canonical request is UTF-8"); assert!(canonical.starts_with( "GET http://allowed.example.test/api HTTP/1.1\r\nHost: allowed.example.test\r\n" )); } #[tokio::test] async fn middleware_selected_forward_keeps_canonical_host_on_the_wire() { let authority = "allowed.example.test"; let raw = b"GET http://allowed.example.test/api HTTP/1.1\r\n\ Host: disallowed.example.test\r\n\ HOST: second.example.test\r\n\r\n"; let raw = canonicalize_forward_host_header(raw, authority) .expect("canonicalize Host before middleware"); let request = crate::l7::rest::request_from_buffered_http("GET", "/api", "/api", raw) .expect("build middleware request"); let ctx = crate::l7::relay::L7EvalContext { host: authority.into(), port: 80, policy_name: "test".into(), binary_path: "/usr/bin/node".into(), ancestors: vec![], cmdline_paths: vec![], secret_resolver: None, activity_tx: None, dynamic_credentials: None, token_grant_resolver: None, }; let runner = openshell_supervisor_middleware::ChainRunner::new( openshell_supervisor_middleware_builtins::services() .into_iter() .next() .expect("built-in middleware service"), ); let guard = forward_test_guard(); let pipeline = ForwardMiddlewarePipeline { ctx: &ctx, scheme: "http", runner: &runner, generation_guard: &guard, l7_reevaluation: None, }; let chain = vec![openshell_supervisor_middleware::ChainEntry { name: "redactor".into(), implementation: openshell_supervisor_middleware_builtins::BUILTIN_REGEX.into(), order: 0, config: prost_types::Struct::default(), on_error: openshell_supervisor_middleware::OnError::FailClosed, }]; let (_app, mut client) = tokio::io::duplex(8192); let allowed = pipeline .apply(request, &mut client, chain) .await .expect("middleware pipeline"); let crate::l7::middleware::MiddlewareApplyResult::Allowed(request) = allowed else { panic!("middleware-selected request should be allowed"); }; let rewritten = rewrite_forward_request( &request.raw_header, request.raw_header.len(), "/api", authority, None, false, ) .expect("rewrite middleware-selected request"); let rewritten = String::from_utf8(rewritten).expect("rewritten request is UTF-8"); assert_eq!( rewritten .split("\r\n") .filter(|line| { line.split_once(':') .is_some_and(|(name, _)| name.eq_ignore_ascii_case("host")) }) .collect::>(), ["Host: allowed.example.test"] ); assert!(!rewritten.contains("disallowed.example.test")); assert!(!rewritten.contains("second.example.test")); } #[test] fn test_rewrite_strips_proxy_headers() { let raw = b"GET http://host/p HTTP/1.1\r\nHost: host\r\nProxy-Authorization: Basic abc\r\nProxy-Connection: keep-alive\r\nAccept: */*\r\n\r\n"; let result = rewrite_forward_request(raw, raw.len(), "/p", "host", None, false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); assert!( !result_str .to_ascii_lowercase() .contains("proxy-authorization") ); assert!(!result_str.to_ascii_lowercase().contains("proxy-connection")); assert!(result_str.contains("Accept: */*")); } #[test] fn test_rewrite_replaces_connection_header() { let raw = b"GET http://host/p HTTP/1.1\r\nHost: host\r\nConnection: keep-alive\r\n\r\n"; let result = rewrite_forward_request(raw, raw.len(), "/p", "host", None, false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); assert!(result_str.contains("Connection: close")); assert!(!result_str.contains("keep-alive")); } #[test] fn test_rewrite_strips_connection_nominated_headers() { let raw = b"GET http://host/p HTTP/1.1\r\nHost: host\r\nX-Guard: hidden\r\nConnection: keep-alive, x-guard\r\nKeep-Alive: timeout=5\r\nX-Visible: yes\r\n\r\n"; let result = rewrite_forward_request(raw, raw.len(), "/p", "host", None, false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); let lower = result_str.to_ascii_lowercase(); assert!(!lower.contains("x-guard:")); assert!(!lower.contains("keep-alive:")); assert!(result_str.contains("Connection: close\r\n")); assert!(result_str.contains("X-Visible: yes\r\n")); } #[test] fn test_rewrite_preserves_body_overflow() { let raw = b"POST http://host/api HTTP/1.1\r\nHost: host\r\nContent-Length: 13\r\n\r\n{\"key\":\"val\"}"; let result = rewrite_forward_request(raw, raw.len(), "/api", "host", None, false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); assert!(result_str.contains("{\"key\":\"val\"}")); assert!(result_str.contains("POST /api HTTP/1.1")); } #[test] fn test_rewrite_preserves_existing_via() { let raw = b"GET http://host/p HTTP/1.1\r\nHost: host\r\nVia: 1.0 upstream\r\n\r\n"; let result = rewrite_forward_request(raw, raw.len(), "/p", "host", None, false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); assert!(result_str.contains("Via: 1.0 upstream")); // Should not add a second Via header assert!(!result_str.contains("Via: 1.1 openshell-sandbox")); } #[test] fn test_rewrite_forward_request_uses_canonical_path_on_the_wire() { // Regression: the forward-proxy caller must canonicalize first and // then pass the canonical form to rewrite_forward_request so that // OPA's policy evaluation and the bytes dispatched to the upstream // agree. Prior to this guarantee, OPA saw the canonical form while // the upstream re-normalized the raw path independently, re-opening // the parser-differential this PR closes. let raw = b"GET http://host/public/../secret HTTP/1.1\r\nHost: host\r\n\r\n"; let (canon, _) = crate::l7::path::canonicalize_request_target( "/public/../secret", &crate::l7::path::CanonicalizeOptions::default(), ) .expect("canonicalization should succeed for the attack payload"); assert_eq!(canon.path, "/secret"); let rewritten = rewrite_forward_request(raw, raw.len(), &canon.path, "host", None, false) .expect("rewrite_forward_request should succeed"); let rewritten_str = String::from_utf8_lossy(&rewritten); assert!( rewritten_str.starts_with("GET /secret HTTP/1.1\r\n"), "outbound request line must use canonical path, got: {rewritten_str:?}" ); assert!( !rewritten_str.contains(".."), "outbound bytes must not leak the pre-canonical form, got: {rewritten_str:?}" ); } #[test] fn test_rewrite_forward_request_preserves_canonical_query_on_the_wire() { let raw = b"GET http://host/public/../graphql?query=query+Viewer+%7B+viewer+%7B+login+%7D+%7D HTTP/1.1\r\nHost: host\r\n\r\n"; let (canon, raw_query) = crate::l7::path::canonicalize_request_target( "/public/../graphql?query=query+Viewer+%7B+viewer+%7B+login+%7D+%7D", &crate::l7::path::CanonicalizeOptions::default(), ) .expect("canonicalization should preserve query separately"); let upstream_target = match raw_query.as_deref() { Some(raw_query) if !raw_query.is_empty() => format!("{}?{raw_query}", canon.path), _ => canon.path, }; let rewritten = rewrite_forward_request(raw, raw.len(), &upstream_target, "host", None, false) .expect("rewrite_forward_request should succeed"); let rewritten_str = String::from_utf8_lossy(&rewritten); assert!( rewritten_str.starts_with( "GET /graphql?query=query+Viewer+%7B+viewer+%7B+login+%7D+%7D HTTP/1.1\r\n" ), "outbound request line must preserve canonical query, got: {rewritten_str:?}" ); } #[test] fn test_rewrite_resolves_placeholder_auth_headers() { let (_, resolver) = SecretResolver::from_provider_env( [("ANTHROPIC_API_KEY".to_string(), "sk-test".to_string())] .into_iter() .collect(), ); let raw = b"GET http://host/p HTTP/1.1\r\nHost: host\r\nAuthorization: Bearer openshell:resolve:env:ANTHROPIC_API_KEY\r\n\r\n"; let result = rewrite_forward_request(raw, raw.len(), "/p", "host", resolver.as_ref(), false) .expect("should succeed"); let result_str = String::from_utf8_lossy(&result); assert!(result_str.contains("Authorization: Bearer sk-test")); assert!(!result_str.contains("openshell:resolve:env:ANTHROPIC_API_KEY")); } #[tokio::test] async fn forward_relay_rewrites_urlencoded_body_alias_from_initial_read() { let (_, resolver) = SecretResolver::from_provider_env( [("API_TOKEN".to_string(), "provider-real-token".to_string())] .into_iter() .collect(), ); let resolver = resolver.expect("resolver"); let alias = "provider-OPENSHELL-RESOLVE-ENV-API_TOKEN"; let body = format!("token={alias}&channel=C123"); let raw = format!( "POST http://api.example.com/api/messages HTTP/1.1\r\n\ Host: api.example.com\r\n\ Authorization: Bearer {alias}\r\n\ Content-Type: application/x-www-form-urlencoded\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body ); let forwarded = relay_forward_request_and_capture( "POST", "/api/messages", raw.as_bytes(), Some(&resolver), true, ) .await .expect("forward relay should rewrite credentials"); let expected_body = "token=provider-real-token&channel=C123"; assert!(forwarded.starts_with("POST /api/messages HTTP/1.1\r\n")); assert!(forwarded.contains("Authorization: Bearer provider-real-token\r\n")); assert!(forwarded.contains(&format!("Content-Length: {}\r\n", expected_body.len()))); assert!(forwarded.ends_with(expected_body)); assert!(!forwarded.contains("OPENSHELL-RESOLVE-ENV")); } #[tokio::test] async fn forward_relay_rewrites_urlencoded_canonical_body_from_initial_read() { let (_, resolver) = SecretResolver::from_provider_env( [("API_TOKEN".to_string(), "provider-real-token".to_string())] .into_iter() .collect(), ); let resolver = resolver.expect("resolver"); let alias = "provider-OPENSHELL-RESOLVE-ENV-API_TOKEN"; let body = "token=openshell%3Aresolve%3Aenv%3AAPI_TOKEN&channel=C123"; let raw = format!( "POST http://api.example.com/api/messages HTTP/1.1\r\n\ Host: api.example.com\r\n\ Authorization: Bearer {alias}\r\n\ Content-Type: application/x-www-form-urlencoded\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body ); let forwarded = relay_forward_request_and_capture( "POST", "/api/messages", raw.as_bytes(), Some(&resolver), true, ) .await .expect("forward relay should rewrite credentials"); let expected_body = "token=provider-real-token&channel=C123"; assert!(forwarded.contains("Authorization: Bearer provider-real-token\r\n")); assert!(forwarded.contains(&format!("Content-Length: {}\r\n", expected_body.len()))); assert!(forwarded.ends_with(expected_body)); assert!(!forwarded.contains("openshell%3Aresolve%3Aenv%3AAPI_TOKEN")); assert!(!forwarded.contains("openshell:resolve:env:API_TOKEN")); } #[tokio::test] async fn forward_relay_unresolved_body_placeholder_fails_before_upstream_write() { let (_, resolver) = SecretResolver::from_provider_env( [("API_TOKEN".to_string(), "provider-real-token".to_string())] .into_iter() .collect(), ); let resolver = resolver.expect("resolver"); let alias = "provider-OPENSHELL-RESOLVE-ENV-API_TOKEN"; let body = "token=provider-OPENSHELL-RESOLVE-ENV-MISSING_TOKEN"; let raw = format!( "POST http://api.example.com/api/messages HTTP/1.1\r\n\ Host: api.example.com\r\n\ Authorization: Bearer {alias}\r\n\ Content-Type: application/x-www-form-urlencoded\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body ); let guard = forward_test_guard(); let rewritten = rewrite_forward_request( raw.as_bytes(), raw.len(), "/api/messages", "api.example.com", Some(&resolver), true, ) .expect("header rewrite should defer body overflow to body rewriter"); let (mut proxy_to_upstream, mut upstream_side) = tokio::io::duplex(8192); let (mut _app_side, mut proxy_to_client) = tokio::io::duplex(8192); let err = relay_rewritten_forward_request( "POST", "/api/messages", rewritten, &mut proxy_to_client, &mut proxy_to_upstream, ForwardRelayOptions { generation_guard: &guard, websocket_extensions: crate::l7::rest::WebSocketExtensionMode::Preserve, secret_resolver: Some(&resolver), request_body_credential_rewrite: true, }, ) .await .expect_err("unresolved body placeholder should fail closed"); assert!(!err.to_string().contains("provider-real-token")); assert!(!err.to_string().contains("MISSING_TOKEN")); drop(proxy_to_upstream); let mut forwarded = Vec::new(); upstream_side.read_to_end(&mut forwarded).await.unwrap(); assert!( forwarded.is_empty(), "failed forward body rewrite must not reach upstream" ); } #[test] fn test_forward_rewrite_preserves_websocket_upgrade_connection_header() { let raw = "GET http://gateway.example.test/ws HTTP/1.1\r\n\ Host: gateway.example.test\r\n\ Upgrade: h2c\r\n\ Upgrade: h2c, websocket\r\n\ Connection: keep-alive, Upgrade\r\n\ Keep-Alive: timeout=5\r\n\ X-Guard: hidden\r\n\ Connection: x-guard\r\n\ Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\ Sec-WebSocket-Extensions: permessage-deflate; client_no_context_takeover\r\n\ Sec-WebSocket-Version: 13\r\n\r\n"; let result = rewrite_forward_request( raw.as_bytes(), raw.len(), "/ws", "gateway.example.test", None, false, ) .expect("websocket forward rewrite should succeed"); let result_str = String::from_utf8_lossy(&result); assert!(result_str.starts_with("GET /ws HTTP/1.1\r\n")); assert_eq!(result_str.matches("Connection: Upgrade\r\n").count(), 1); assert_eq!(result_str.matches("Upgrade: websocket\r\n").count(), 1); assert!(!result_str.to_ascii_lowercase().contains("upgrade: h2c")); assert!(!result_str.contains("keep-alive")); assert!(!result_str.to_ascii_lowercase().contains("x-guard:")); assert!( !result_str.contains("Connection: close\r\n"), "websocket forward proxy must not strip the upgrade token" ); } #[tokio::test] async fn test_forward_relay_guard_blocks_stale_generation_before_upstream_write() { let policy = include_str!("../data/sandbox-policy.rego"); let policy_data = "network_policies: {}\n"; let engine = OpaEngine::from_strings(policy, policy_data).unwrap(); let guard = engine .generation_guard(engine.current_generation()) .unwrap(); engine.reload(policy, policy_data).unwrap(); let raw = b"GET http://host/api HTTP/1.1\r\nHost: host\r\n\r\n"; let rewritten = rewrite_forward_request(raw, raw.len(), "/api", "host", None, false) .expect("rewrite should succeed"); let (mut proxy_to_upstream, mut upstream_side) = tokio::io::duplex(8192); let (mut _app_side, mut proxy_to_client) = tokio::io::duplex(8192); let result = relay_rewritten_forward_request( "GET", "/api", rewritten, &mut proxy_to_client, &mut proxy_to_upstream, ForwardRelayOptions { generation_guard: &guard, websocket_extensions: crate::l7::rest::WebSocketExtensionMode::Preserve, secret_resolver: None, request_body_credential_rewrite: false, }, ) .await; assert!( result.is_err(), "stale generation must stop forward relay before upstream write" ); drop(proxy_to_upstream); let mut forwarded = Vec::new(); upstream_side.read_to_end(&mut forwarded).await.unwrap(); assert!( forwarded.is_empty(), "stale forward request bytes must not reach upstream" ); } #[tokio::test] async fn test_forward_relay_rejects_cl_te_smuggling_before_upstream_write() { let policy = include_str!("../data/sandbox-policy.rego"); let policy_data = "network_policies: {}\n"; let engine = OpaEngine::from_strings(policy, policy_data).unwrap(); let guard = engine .generation_guard(engine.current_generation()) .unwrap(); let raw = b"POST http://host/api HTTP/1.1\r\nHost: host\r\nContent-Length: 4\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n"; let rewritten = rewrite_forward_request(raw, raw.len(), "/api", "host", None, false) .expect("rewrite should succeed"); let (mut proxy_to_upstream, mut upstream_side) = tokio::io::duplex(8192); let (mut _app_side, mut proxy_to_client) = tokio::io::duplex(8192); let result = relay_rewritten_forward_request( "POST", "/api", rewritten, &mut proxy_to_client, &mut proxy_to_upstream, ForwardRelayOptions { generation_guard: &guard, websocket_extensions: crate::l7::rest::WebSocketExtensionMode::Preserve, secret_resolver: None, request_body_credential_rewrite: false, }, ) .await; assert!(result.is_err(), "forward relay must reject CL/TE ambiguity"); drop(proxy_to_upstream); let mut forwarded = Vec::new(); upstream_side.read_to_end(&mut forwarded).await.unwrap(); assert!( forwarded.is_empty(), "smuggled forward request bytes must not reach upstream" ); } // --- Forward proxy SSRF defence tests --- // // The forward proxy handler uses the same SSRF logic as the CONNECT path: // - No allowed_ips: resolve_and_reject_internal blocks private IPs, allows public. // - With allowed_ips: resolve_and_check_allowed_ips validates against allowlist. // // These tests document that contract for the forward proxy path specifically. #[tokio::test] async fn test_forward_public_ip_allowed_without_allowed_ips() { // Public IPs (e.g. dns.google -> 8.8.8.8) should pass through // resolve_and_reject_internal without needing allowed_ips. let result = resolve_and_reject_internal("dns.google", 80, 0).await; assert!( result.is_ok(), "Public IP should be allowed without allowed_ips: {result:?}" ); let addrs = result.unwrap(); assert!(!addrs.is_empty(), "Should resolve to at least one address"); // All resolved addresses should be public. for addr in &addrs { assert!( !is_internal_ip(addr.ip()), "dns.google should resolve to public IPs, got {}", addr.ip() ); } } #[tokio::test] async fn test_forward_private_ip_rejected_without_allowed_ips() { // Private IP literals should be rejected by resolve_and_reject_internal. let result = resolve_and_reject_internal("10.0.0.1", 80, 0).await; assert!( result.is_err(), "Private IP should be rejected without allowed_ips" ); let err = result.unwrap_err(); assert!( err.contains("internal address"), "expected 'internal address' in error: {err}" ); } #[tokio::test] async fn test_forward_private_ip_accepted_with_allowed_ips() { // Private IP with matching allowed_ips should pass through. let nets = parse_allowed_ips(&["10.0.0.0/8".to_string()]).unwrap(); let result = resolve_and_check_allowed_ips("10.0.0.1", 80, &nets, 0).await; assert!( result.is_ok(), "Private IP with matching allowed_ips should be accepted: {result:?}" ); } #[tokio::test] async fn test_forward_private_ip_rejected_with_wrong_allowed_ips() { // Private IP not in allowed_ips should be rejected. let nets = parse_allowed_ips(&["192.168.0.0/16".to_string()]).unwrap(); let result = resolve_and_check_allowed_ips("10.0.0.1", 80, &nets, 0).await; assert!( result.is_err(), "Private IP not in allowed_ips should be rejected" ); let err = result.unwrap_err(); assert!( err.contains("not in allowed_ips"), "expected 'not in allowed_ips' in error: {err}" ); } #[tokio::test] async fn test_forward_loopback_always_blocked_even_with_allowed_ips() { // Loopback addresses are always blocked, even if in allowed_ips. // Construct nets directly (parse_allowed_ips now rejects always-blocked). let nets = vec!["127.0.0.0/8".parse::().unwrap()]; let result = resolve_and_check_allowed_ips("127.0.0.1", 80, &nets, 0).await; assert!(result.is_err(), "Loopback should be always blocked"); let err = result.unwrap_err(); assert!( err.contains("always-blocked"), "expected 'always-blocked' in error: {err}" ); } #[tokio::test] async fn test_forward_link_local_always_blocked_even_with_allowed_ips() { // Link-local / cloud metadata addresses are always blocked. // Construct nets directly (parse_allowed_ips now rejects always-blocked). let nets = vec!["169.254.0.0/16".parse::().unwrap()]; let result = resolve_and_check_allowed_ips("169.254.169.254", 80, &nets, 0).await; assert!(result.is_err(), "Link-local should be always blocked"); let err = result.unwrap_err(); assert!( err.contains("always-blocked"), "expected 'always-blocked' in error: {err}" ); } // -- SSRF: malformed hostname resolution regression tests -- #[tokio::test] async fn test_resolve_reject_internal_fails_closed_on_nul_hostname() { let result = resolve_and_reject_internal("evil.com\0.safe.com", 443, 0).await; assert!( result.is_err(), "NUL-containing hostname should fail DNS resolution (fail closed)" ); } #[tokio::test] async fn test_resolve_allowed_ips_fails_closed_on_nul_hostname() { let nets = parse_allowed_ips(&["0.0.0.0/0".to_string()]) .unwrap_or_else(|_| vec!["0.0.0.0/0".parse::().unwrap()]); let result = resolve_and_check_allowed_ips("evil.com\0.safe.com", 443, &nets, 0).await; assert!( result.is_err(), "NUL-containing hostname should fail DNS resolution (fail closed)" ); } // -- implicit_allowed_ips_for_ip_host -- #[test] fn test_implicit_allowed_ips_returns_ip_for_ipv4_literal() { let result = implicit_allowed_ips_for_ip_host("192.168.1.100"); assert_eq!(result, vec!["192.168.1.100"]); } #[test] fn test_implicit_allowed_ips_skips_ipv6_loopback() { // ::1 is always-blocked, so implicit allowed_ips should be empty. let result = implicit_allowed_ips_for_ip_host("::1"); assert!(result.is_empty()); } #[test] fn test_implicit_allowed_ips_returns_empty_for_hostname() { let result = implicit_allowed_ips_for_ip_host("api.github.com"); assert!(result.is_empty()); } #[test] fn test_implicit_allowed_ips_returns_empty_for_wildcard() { let result = implicit_allowed_ips_for_ip_host("*.example.com"); assert!(result.is_empty()); } /// Regression test: exercises the actual keep-alive interception loop to /// verify that a non-inference request is denied even after a previous /// inference request was successfully routed on the same connection. /// /// Before the fix, `handle_inference_interception` used /// `else if !routed_any` which silently dropped denials once `routed_any` /// was true, allowing non-inference HTTP requests to piggyback on a /// keep-alive connection that had previously handled inference traffic. /// Regression test: exercises the actual keep-alive interception loop to /// verify that a non-inference request is denied even after a previous /// inference request was successfully routed on the same connection. /// /// The server runs in a spawned task with empty routes (the inference /// request gets a 503 "not configured" but is still recognized as /// inference and returns Ok(true)). The client sends the inference /// request, reads the 503 response, then sends a non-inference request /// on the same connection. The server must return Denied. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_keepalive_denies_non_inference_after_routed() { use openshell_router::Router; use tokio::io::{AsyncReadExt, AsyncWriteExt}; let router = Router::new().unwrap(); let patterns = crate::l7::inference::default_patterns(); // Empty routes: inference request gets 503 but returns Ok(true). let ctx = InferenceContext::new(patterns, router, vec![], vec![]); let body = r#"{"model":"test","messages":[{"role":"user","content":"hi"}]}"#; let inference_req = format!( "POST /v1/chat/completions HTTP/1.1\r\n\ Host: inference.local\r\n\ Content-Type: application/json\r\n\ Content-Length: {}\r\n\r\n{}", body.len(), body, ); let non_inference_req = "GET /admin/config HTTP/1.1\r\nHost: inference.local\r\n\r\n"; let (client, mut server) = tokio::io::duplex(65536); let (mut client_read, mut client_write) = tokio::io::split(client); // Spawn the server task so it runs concurrently. let server_task = tokio::spawn(async move { process_inference_keepalive(&mut server, &ctx, 443).await }); // Client: send inference request, read response, send non-inference. client_write .write_all(inference_req.as_bytes()) .await .unwrap(); // Read the 503 response so the server loops back to read. let mut buf = vec![0u8; 4096]; let _ = client_read.read(&mut buf).await.unwrap(); // Send non-inference request on the same keep-alive connection. client_write .write_all(non_inference_req.as_bytes()) .await .unwrap(); drop(client_write); // Drain remaining response bytes. tokio::spawn(async move { let mut buf = vec![0u8; 4096]; loop { match client_read.read(&mut buf).await { Ok(0) | Err(_) => break, Ok(_) => continue, } } }); let outcome = server_task.await.unwrap().unwrap(); assert!( matches!(outcome, InferenceOutcome::Denied { .. }), "expected Denied after non-inference request on keep-alive, got: {outcome:?}" ); } // -- build_json_error_response -- #[test] fn test_json_error_response_403() { let resp = build_json_error_response( 403, "Forbidden", "policy_denied", "CONNECT api.example.com:443 not permitted by policy", ); let resp_str = String::from_utf8(resp).unwrap(); assert!(resp_str.starts_with("HTTP/1.1 403 Forbidden\r\n")); assert!(resp_str.contains("Content-Type: application/json\r\n")); assert!(resp_str.contains("Connection: close\r\n")); // Extract body after \r\n\r\n let body_start = resp_str.find("\r\n\r\n").unwrap() + 4; let body: serde_json::Value = serde_json::from_str(&resp_str[body_start..]).unwrap(); assert_eq!(body["error"], "policy_denied"); assert_eq!( body["detail"], "CONNECT api.example.com:443 not permitted by policy" ); } #[test] fn test_json_error_response_502() { let resp = build_json_error_response( 502, "Bad Gateway", "upstream_unreachable", "connection to api.example.com:443 failed", ); let resp_str = String::from_utf8(resp).unwrap(); assert!(resp_str.starts_with("HTTP/1.1 502 Bad Gateway\r\n")); let body_start = resp_str.find("\r\n\r\n").unwrap() + 4; let body: serde_json::Value = serde_json::from_str(&resp_str[body_start..]).unwrap(); assert_eq!(body["error"], "upstream_unreachable"); assert_eq!(body["detail"], "connection to api.example.com:443 failed"); } #[test] fn middleware_deny_response_uses_policy_config_identity() { let resp = build_middleware_deny_response( "api-policy", &openshell_supervisor_middleware::MiddlewareDenial { config_name: "prototype-content-guard".into(), reason_code: Some("content_match".into()), }, ); let resp_str = String::from_utf8(resp).unwrap(); let body_start = resp_str.find("\r\n\r\n").unwrap() + 4; let body: serde_json::Value = serde_json::from_str(&resp_str[body_start..]).unwrap(); assert_eq!(body["error"], "middleware_denied"); assert_eq!(body["detail"], "Request rejected by configured middleware"); assert_eq!(body["middleware"], "prototype-content-guard"); assert_eq!(body["reason_code"], "content_match"); assert_eq!(body["policy"], "api-policy"); assert!(body.get("rule_missing").is_none()); assert!(body.get("next_steps").is_none()); } /// Locks the fail-closed response the CONNECT handler sends when TLS is /// detected but no termination state exists (ephemeral CA setup failed). /// The proxy must refuse the connection with a 503 instead of raw-tunneling /// the TLS stream, which would bypass credential rewrite and leak /// placeholders verbatim. #[test] fn test_json_error_response_503_tls_termination_unavailable() { let detail = TLS_TERMINATION_UNAVAILABLE_DETAIL; let resp = build_json_error_response( 503, "Service Unavailable", "tls_termination_unavailable", detail, ); let resp_str = String::from_utf8(resp).unwrap(); assert!(resp_str.starts_with("HTTP/1.1 503 Service Unavailable\r\n")); assert!(resp_str.contains("Connection: close\r\n")); let body_start = resp_str.find("\r\n\r\n").unwrap() + 4; let body: serde_json::Value = serde_json::from_str(&resp_str[body_start..]).unwrap(); assert_eq!(body["error"], "tls_termination_unavailable"); assert_eq!(body["detail"], detail); } /// Connection-level regression for the pre-200 fail-closed gate. A real /// loopback client opens a connection and the proxy-side gate runs with no /// TLS termination state for a terminating (non-`tls: skip`) route. The /// FIRST bytes the client reads back must be a real `HTTP/1.1 503`, not the /// `HTTP/1.1 200 Connection Established` that would establish the tunnel and /// bury the refusal inside it as a TLS protocol error (PR #2162 gator flaw). #[tokio::test] async fn connect_without_tls_termination_refused_with_503_before_200() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let mut client = TcpStream::connect(addr).await.unwrap(); let (mut server, _) = listener.accept().await.unwrap(); // Terminating route (tls: skip = false) with no termination state. let refused = refuse_connect_when_tls_unavailable(&mut server, false, false) .await .expect("gate write should succeed"); assert!( refused, "gate must refuse when TLS termination is unavailable" ); let mut buf = vec![0u8; 512]; let n = client.read(&mut buf).await.unwrap(); let response = String::from_utf8_lossy(&buf[..n]); assert!( response.starts_with("HTTP/1.1 503 Service Unavailable\r\n"), "client must read a 503 as the first bytes, not a 200; got: {response}" ); assert!( !response.contains("200 Connection Established"), "no tunnel must be established before the refusal; got: {response}" ); let body_start = response.find("\r\n\r\n").unwrap() + 4; let body: serde_json::Value = serde_json::from_str(&response[body_start..]).unwrap(); assert_eq!(body["error"], "tls_termination_unavailable"); assert_eq!(body["detail"], TLS_TERMINATION_UNAVAILABLE_DETAIL); } /// The pre-200 gate must NOT touch the socket when the route can proceed: /// either TLS termination state is present, or the route is `tls: skip` /// (raw tunnel, no termination or credential injection). In both cases the /// gate returns `false` and writes nothing, letting the caller send its /// own `200 Connection Established`. #[tokio::test] async fn connect_with_tls_termination_or_skip_proceeds_without_writing() { for (tls_present, tls_skip) in [(true, false), (false, true), (true, true)] { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let mut client = TcpStream::connect(addr).await.unwrap(); let (mut server, _) = listener.accept().await.unwrap(); let refused = refuse_connect_when_tls_unavailable(&mut server, tls_present, tls_skip) .await .expect("gate should succeed"); assert!( !refused, "gate must proceed (tls_present={tls_present}, tls_skip={tls_skip})" ); // Nothing was written; after dropping the server the client sees a // clean EOF (0 bytes) rather than any buffered response. drop(server); let mut buf = vec![0u8; 16]; let n = client.read(&mut buf).await.unwrap(); assert_eq!( n, 0, "gate must not write to the socket when proceeding \ (tls_present={tls_present}, tls_skip={tls_skip})" ); } } /// Drives a real `CONNECT` through the full `handle_tcp_connection` entry /// point and returns `(completed, client_response_bytes, denial_stages)`. /// `completed` is `false` when the handler was still running at `budget` /// (e.g. a `tls: skip` route that proceeded past the fail-closed gate and /// blocked on the unroutable upstream connect). `denial_stages` collects the /// `denial_stage` of every `DenialEvent` the handler emitted — empty means /// the connection was allowed and not blocked/refused at any stage. /// /// The client is an in-process `TcpStream` (no child process), so the client /// socket is owned solely by this test process. Identity resolution finds /// that owner in the descendant scan (which includes the entrypoint PID /// itself), binds to `current_exe()`, and never falls through to the /// whole-`/proc` scan — the environment-sensitive path that made a forked /// child flaky under a busy CI `/proc`. Callers gate on Linux; /// `evaluate_opa_tcp` denies unconditionally without `/proc`. async fn drive_connect_through_handler( endpoint_yaml: &str, connect_target: &str, budget: std::time::Duration, ) -> (bool, Vec, Vec) { const POLICY_REGO: &str = include_str!("../data/sandbox-policy.rego"); // Allow the current test binary — the in-process client's // `/proc//exe` — to reach the endpoint. Matching is by path. let exe = std::env::current_exe().expect("current_exe"); let data = format!( r#"network_policies: test_allow: name: test_allow endpoints: {endpoint_yaml} binaries: - {{ path: "{exe}" }} "#, exe = exe.display(), ); let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, &data).expect("load policy")); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let proxy_port = listener.local_addr().unwrap().port(); // In-process client: connect, send the CONNECT request, read the reply to // EOF. The proxy closes the socket after writing a 503/403; a proceeding // `tls: skip` route stalls on the upstream connect until the handler is // dropped at `budget`, which closes the socket and ends the read. let target = connect_target.to_string(); let client = tokio::spawn(async move { let mut sock = TcpStream::connect(("127.0.0.1", proxy_port)).await.unwrap(); let req = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n"); sock.write_all(req.as_bytes()).await.unwrap(); let mut buf = Vec::new(); let _ = sock.read_to_end(&mut buf).await; buf }); let (server, _peer) = listener.accept().await.unwrap(); let entrypoint_pid = Arc::new(AtomicU32::new(std::process::id())); let cache = Arc::new(BinaryIdentityCache::new()); let (denial_tx, mut denial_rx) = mpsc::unbounded_channel(); let completed = tokio::time::timeout( budget, Box::pin(handle_tcp_connection( server, engine, cache, entrypoint_pid, None, // tls_state — ephemeral CA unavailable None, // inference_ctx None, // policy_local_ctx Arc::new(None), // trusted_host_gateway None, // secret_resolver None, // dynamic_credentials Some(denial_tx), // denial_tx — positive allow/deny signal None, // activity_tx )), ) .await .is_ok(); let stdout = client.await.expect("client task"); // The handler future is now finished or dropped, so its sender half is // gone; drain every denial it emitted. let mut denial_stages = Vec::new(); while let Ok(event) = denial_rx.try_recv() { denial_stages.push(event.denial_stage); } (completed, stdout, denial_stages) } /// End-to-end regression for the gator finding on PR #2162: with no TLS /// termination state, a terminating `CONNECT` must have its 503 written as /// the FIRST bytes on the socket — never after a `200 Connection /// Established` that would bury the refusal inside the tunnel as a TLS error. #[tokio::test] async fn connect_handler_refuses_terminating_route_with_503_before_200() { if !cfg!(target_os = "linux") { eprintln!("skipping: handler identity binding requires /proc (Linux)"); return; } // Public documentation IP (RFC 5737 TEST-NET-3): passes the SSRF check // but is never actually connected — the fail-closed gate refuses first. // The 30s budget is a generous belt against a slow CI runner; the refusal // returns in milliseconds. let (completed, stdout, _denials) = drive_connect_through_handler( " - { host: \"203.0.113.10\", port: 443 }\n", "203.0.113.10:443", std::time::Duration::from_secs(30), ) .await; assert!(completed, "the refusal must return promptly, not hang"); let resp = String::from_utf8_lossy(&stdout); assert!( resp.starts_with("HTTP/1.1 503 Service Unavailable"), "first bytes must be a 503, not a 200; got: {resp:?}" ); assert!( !resp.contains("200 Connection Established"), "no tunnel must be established before the refusal; got: {resp:?}" ); assert!( resp.contains("tls_termination_unavailable"), "must be the TLS-termination-unavailable refusal; got: {resp:?}" ); } /// Ordering regression (gator re-check item 1): during CA-init failure, an /// internal-address `CONNECT` must still receive the SSRF `403`, not the /// TLS-unavailable `503` — SSRF validation runs before the fail-closed gate. #[tokio::test] async fn connect_handler_returns_ssrf_403_for_internal_address_not_503() { if !cfg!(target_os = "linux") { eprintln!("skipping: handler identity binding requires /proc (Linux)"); return; } let (completed, stdout, _denials) = drive_connect_through_handler( " - { host: \"127.0.0.1\", port: 443 }\n", "127.0.0.1:443", std::time::Duration::from_secs(30), ) .await; assert!(completed, "the SSRF denial must return promptly, not hang"); let resp = String::from_utf8_lossy(&stdout); assert!( resp.starts_with("HTTP/1.1 403 Forbidden"), "internal address must get the SSRF 403; got: {resp:?}" ); assert!( resp.contains("ssrf_denied"), "must be an SSRF denial; got: {resp:?}" ); assert!( !resp.contains("503"), "SSRF runs before the fail-closed gate, so the 503 must not appear; got: {resp:?}" ); } /// A real `tls: skip` policy path through the handler is exempt from the /// fail-closed gate even with no TLS termination state: the handler proceeds /// past the refusal to the raw-tunnel upstream connect (which stalls on the /// unroutable target), emitting no 503 and no client response before it. #[tokio::test] async fn connect_handler_tls_skip_route_is_not_refused() { if !cfg!(target_os = "linux") { eprintln!("skipping: handler identity binding requires /proc (Linux)"); return; } // `_completed` is intentionally ignored: whether the unroutable connect // stalls (times out) or is rejected fast by a CI egress firewall, the // invariant is the same — a `tls: skip` route is exempt from the // fail-closed gate, so it proceeds to the tunnel connect and writes no // 503/denial to the client beforehand. let (_completed, stdout, denial_stages) = drive_connect_through_handler( " - { host: \"203.0.113.10\", port: 443, tls: skip }\n", "203.0.113.10:443", std::time::Duration::from_millis(800), ) .await; let resp = String::from_utf8_lossy(&stdout); assert!( stdout.is_empty(), "tls: skip must emit no refusal/denial before the tunnel connect; got: {resp:?}" ); // Positive allow evidence, so this test cannot pass vacuously on a // policy/identity deny (which would emit a DenialEvent): the handler // must have emitted no denial at any stage. assert!( denial_stages.is_empty(), "tls: skip must be allowed end-to-end, not denied at any stage; got: {denial_stages:?}" ); } /// Verifies the policy half of the Linux handler tests on every platform /// (no `/proc` needed): the temp-dir binary glob yields an Allow for a /// literal-IP endpoint, `tls: skip` reads back as `TlsMode::Skip`, and a /// terminating endpoint reads back as `TlsMode::Auto`. Locks the OPA /// preconditions so a policy-format regression is caught even where the /// process-identity binding can't run. #[test] fn handler_test_policy_allows_glob_binary_and_reads_tls_mode() { const POLICY_REGO: &str = include_str!("../data/sandbox-policy.rego"); let eval = |tls_line: &str| { let data = format!( r#"network_policies: test_allow: name: test_allow endpoints: - {{ host: "203.0.113.10", port: 443{tls_line} }} binaries: - {{ path: "/tmp/openshell-connect-test/*" }} "# ); let engine = OpaEngine::from_strings(POLICY_REGO, &data).expect("load policy"); let input = crate::opa::NetworkInput { host: "203.0.113.10".to_string(), port: 443, binary_path: PathBuf::from("/tmp/openshell-connect-test/connect-bash"), binary_sha256: "unused".to_string(), ancestors: vec![], cmdline_paths: vec![], }; let (action, generation) = engine .evaluate_network_action_with_generation(&input) .expect("evaluate"); match &action { NetworkAction::Allow { matched_policy } => { assert!(matched_policy.is_some(), "allow must carry the policy name"); } NetworkAction::Deny { reason } => { panic!("glob binary must be allowed, got deny: {reason}") } } let decision = ConnectDecision { action, generation, binary: Some(input.binary_path), binary_pid: Some(1), ancestors: vec![], cmdline_paths: vec![], }; query_tls_mode(&engine, &decision, "203.0.113.10", 443) }; assert_eq!( eval(", tls: skip"), crate::l7::TlsMode::Skip, "tls: skip endpoint must resolve to TlsMode::Skip" ); assert_eq!( eval(""), crate::l7::TlsMode::Auto, "terminating endpoint must resolve to TlsMode::Auto" ); } #[test] fn test_json_error_response_content_length_matches() { let resp = build_json_error_response(403, "Forbidden", "test", "detail"); let resp_str = String::from_utf8(resp).unwrap(); // Extract Content-Length value let cl_line = resp_str .lines() .find(|l| l.starts_with("Content-Length:")) .unwrap(); let cl: usize = cl_line.split(": ").nth(1).unwrap().trim().parse().unwrap(); // Verify body length matches let body_start = resp_str.find("\r\n\r\n").unwrap() + 4; assert_eq!(resp_str[body_start..].len(), cl); } /// End-to-end regression for the `docker cp` hot-swap hazard around /// unlinked process executables. /// /// `binary_path()` strips the kernel's `" (deleted)"` suffix so policy /// identity and logs use a clean display path. Integrity verification must /// not hash that display path after a hot-swap, because it may now point to /// unrelated replacement bytes. It hashes `/proc//exe` instead, which /// resolves to the live executable inode even after the original path was /// unlinked. /// /// Test shape (from the review comment on the initial PR): /// 1. Start a `TcpListener` in the test process. /// 2. Copy `/bin/bash` to a temp path we control. /// 3. Prime `BinaryIdentityCache` with that temp binary's hash. /// 4. Spawn the temp bash as a child with a `/dev/tcp` one-liner that /// opens a real TCP connection to the listener and holds it open /// inside the bash process. /// 5. Accept the connection on the listener side and capture the peer's /// ephemeral port — that's what `resolve_process_identity` uses to /// walk `/proc/net/tcp` back to the child PID. /// 6. Overwrite the temp bash on disk with different bytes to simulate /// a `docker cp` hot-swap. The running child is unaffected (it still /// executes from its in-memory image), but `/proc//exe` will /// now readlink to `" (deleted)"` OR the overwritten file, depending /// on whether the filesystem reused the inode. /// 7. Call `resolve_process_identity` and assert: /// - identity resolution succeeds using the live executable hash, and /// - the returned display path does not contain the kernel-added /// `"(deleted)"` suffix. #[cfg(target_os = "linux")] #[test] fn resolve_process_identity_hashes_live_exe_after_hot_swap() { use crate::identity::BinaryIdentityCache; use std::io::Read; use std::net::TcpListener; use std::os::unix::fs::PermissionsExt; use std::process::{Command, Stdio}; use std::time::Duration; // Skip if /bin/bash is not present (e.g. minimal containers). if !std::path::Path::new("/bin/bash").exists() { eprintln!("skipping: /bin/bash not available"); return; } // 1. Start a listener on loopback. let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); let listener_port = listener.local_addr().unwrap().port(); // 2. Copy /bin/bash to a temp path. let tmp = tempfile::TempDir::new().unwrap(); let bash_v1 = tmp.path().join("hotswap-bash"); std::fs::copy("/bin/bash", &bash_v1).expect("copy bash"); std::fs::set_permissions(&bash_v1, std::fs::Permissions::from_mode(0o755)).unwrap(); // 3. Prime the cache with the v1 hash of the temp bash. let cache = BinaryIdentityCache::new(); let v1_hash = cache .verify_or_cache(&bash_v1) .expect("prime cache with v1 bash hash"); assert!(!v1_hash.is_empty()); // 4. Spawn the temp bash with a /dev/tcp one-liner that opens a real // connection to the listener and blocks in bash's `read` builtin // to keep it open. Do not use an external command like `sleep`: // it inherits the socket fd and intentionally trips the shared // socket ambiguity guard instead of exercising the hot-swap path. let script = format!("exec 3<>/dev/tcp/127.0.0.1/{listener_port}; read -r -t 30 _ <&3 || true"); let mut child = Command::new(&bash_v1) .arg("-c") .arg(&script) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .expect("spawn hotswap-bash child"); // 5. Accept on the listener side, capture the peer port. listener.set_nonblocking(false).expect("blocking listener"); let (mut stream, peer_addr) = match listener.accept() { Ok(pair) => pair, Err(e) => { let _ = child.kill(); let _ = child.wait(); panic!("failed to accept child connection: {e}"); } }; let peer_port = peer_addr.port(); // Drain any spurious data; we just need the socket open. stream .set_read_timeout(Some(Duration::from_millis(50))) .ok(); let mut buf = [0u8; 16]; let _ = stream.read(&mut buf); // Give the kernel a moment so /proc//net/tcp and // /proc//fd/ both reflect the ESTABLISHED socket. std::thread::sleep(Duration::from_millis(50)); // 6. Simulate `docker cp`: unlink the running binary and create a // fresh file with different bytes at the same path. Writing // in place via O_TRUNC is rejected by the kernel with ETXTBSY // because the inode is still being executed. Unlink is cheap: // the inode persists in memory via the child's exec mapping, // so the child keeps running, but a new inode now lives at // `bash_v1` with a different SHA-256. std::fs::remove_file(&bash_v1).expect("unlink running bash_v1"); let tampered_bytes = b"#!/bin/sh\n# tampered bash v2 from hotswap test\nexit 0\n"; std::fs::write(&bash_v1, tampered_bytes).expect("write replacement bytes"); // 7. Resolve identity through the real helper and assert the // contract: hash the live executable via /proc//exe while // returning a clean display path for policy/logging. let test_pid = std::process::id(); let result = resolve_process_identity(test_pid, peer_port, &cache); let child_pid = child.id(); // Always clean up the child before asserting so a failure doesn't // leak a sleeping process across test runs. let _ = child.kill(); let _ = child.wait(); match result { Ok(identity) => { assert_eq!( identity.binary_pid, child_pid, "expected the hot-swapped bash child to own the socket" ); assert_eq!( identity.bin_path, bash_v1, "expected stripped display path to remain the original binary path" ); assert!( !identity.bin_path.to_string_lossy().contains("(deleted)"), "resolved binary path still tainted: {}", identity.bin_path.display() ); assert_eq!( identity.bin_hash, v1_hash, "expected integrity hash from the live executable, not replacement bytes" ); } Err(err) => panic!( "resolve_process_identity failed after hot-swap; expected live-exe identity: {}", err.reason ), } } #[cfg(target_os = "linux")] #[test] // TODO: exec'ing /bin/sleep (SELinux label bin_t) from a user_home_t test // binary causes /proc//exe readlink to return ENOENT on // SELinux-enforcing hosts. Fix by building a test-sleep-helper binary in // the same crate so it inherits the user_home_t label. fn resolve_process_identity_denies_fork_exec_shared_socket_ambiguity() { use crate::identity::BinaryIdentityCache; use std::ffi::CString; use std::net::{TcpListener, TcpStream}; use std::os::fd::AsRawFd; use std::time::{Duration, Instant}; struct ChildGuard(libc::pid_t); impl Drop for ChildGuard { fn drop(&mut self) { #[allow(unsafe_code)] unsafe { libc::kill(self.0, libc::SIGKILL); libc::waitpid(self.0, std::ptr::null_mut(), 0); } } } if !std::path::Path::new("/bin/sleep").exists() { eprintln!("skipping: /bin/sleep not available"); return; } if std::process::Command::new("getenforce") .output() .is_ok_and(|o| String::from_utf8_lossy(&o.stdout).trim() == "Enforcing") { eprintln!( "skipping: SELinux is enforcing — cross-label /proc//exe readlink fails" ); return; } let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener"); let listener_port = listener.local_addr().unwrap().port(); let stream = TcpStream::connect(("127.0.0.1", listener_port)).expect("connect"); let peer_port = stream.local_addr().unwrap().port(); let (_accepted, _) = listener.accept().expect("accept"); let fd = stream.as_raw_fd(); // libc/syscall FFI requires unsafe #[allow(unsafe_code)] unsafe { let flags = libc::fcntl(fd, libc::F_GETFD); assert!(flags >= 0, "F_GETFD failed"); assert_eq!( libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC), 0, "F_SETFD failed" ); } let sleep_path = CString::new("/bin/sleep").unwrap(); let arg0 = CString::new("sleep").unwrap(); let arg1 = CString::new("30").unwrap(); // libc/syscall FFI requires unsafe #[allow(unsafe_code)] let child_pid = unsafe { libc::fork() }; assert!(child_pid >= 0, "fork failed"); if child_pid == 0 { // libc/syscall FFI requires unsafe #[allow(unsafe_code)] unsafe { libc::execl( sleep_path.as_ptr(), arg0.as_ptr(), arg1.as_ptr(), std::ptr::null::(), ); libc::_exit(127); } } let _guard = ChildGuard(child_pid); let entrypoint_pid = std::process::id(); let deadline = Instant::now() + Duration::from_secs(5); loop { if let Ok(link) = std::fs::read_link(format!("/proc/{child_pid}/exe")) && link.to_string_lossy().contains("sleep") { break; } assert!( Instant::now() < deadline, "child pid {child_pid} did not exec into sleep within 5s" ); std::thread::sleep(Duration::from_millis(20)); } let cache = BinaryIdentityCache::new(); let mut result = resolve_process_identity(entrypoint_pid, peer_port, &cache); for _ in 0..10 { match &result { Err(err) if err.reason.contains("No such file or directory") || err.reason.contains("os error 2") => { // /proc//fd scan transiently failed; give procfs time to settle. std::thread::sleep(Duration::from_millis(50)); result = resolve_process_identity(entrypoint_pid, peer_port, &cache); } Ok(_) => { // On arm64 under heavy CI load the /proc fd scan can transiently // miss the parent process's socket fd, making the scan return only // the child as owner and yielding a spurious Ok. Retry to give // both owners time to appear consistently in /proc//fd. std::thread::sleep(Duration::from_millis(50)); result = resolve_process_identity(entrypoint_pid, peer_port, &cache); } _ => break, } } match result { Ok(identity) => panic!( "resolve_process_identity unexpectedly succeeded for shared socket owned by PID {}", identity.binary_pid ), Err(err) => { assert!( err.reason.contains("ambiguous shared socket ownership"), "expected ambiguous socket ownership error, got: {}", err.reason ); assert!( err.reason.contains(&entrypoint_pid.to_string()), "error should include parent PID; got: {}", err.reason ); assert!( err.reason.contains(&child_pid.to_string()), "error should include child PID; got: {}", err.reason ); } } } }