diff --git a/architecture/sandbox.md b/architecture/sandbox.md index d2cb44d7b8..bfb1c14741 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -53,7 +53,10 @@ paths, such as proxy support files or GPU device paths when a GPU is present. All ordinary agent egress is routed through the sandbox proxy. The proxy identifies the calling binary, checks trust-on-first-use binary identity, rejects -unsafe internal destinations, and evaluates the active policy. +unsafe internal destinations, and evaluates the active policy. On Linux, it +maps an accepted proxy connection back to the workload socket by matching the +complete local-to-remote TCP tuple before resolving every process that owns the +socket inode. For inspected HTTP traffic, the proxy can enforce REST method/path rules, WebSocket upgrade and text-message rules, GraphQL operation rules, and MCP method, tool, and supported params rules or generic JSON-RPC method rules diff --git a/crates/openshell-cli/src/ssh.rs b/crates/openshell-cli/src/ssh.rs index 50bc2bbee9..d277804bf1 100644 --- a/crates/openshell-cli/src/ssh.rs +++ b/crates/openshell-cli/src/ssh.rs @@ -18,6 +18,7 @@ use openshell_core::proto::{ }; use owo_colors::OwoColorize; use std::fs; +use std::future::Future; use std::io::{IsTerminal, Write}; #[cfg(unix)] use std::os::unix::process::CommandExt; @@ -447,9 +448,24 @@ async fn wait_for_forward_start(child: &mut Child, spec: &ForwardSpec) -> Result /// last probe error is folded into the timeout diagnostic, so a failure reports /// why the listener never opened, not just that it timed out. async fn wait_for_forward_listener(spec: &ForwardSpec, wait_for: Duration) -> Result<()> { + wait_for_forward_listener_with_probe(spec, wait_for, |spec| async move { + probe_forward_listener(&spec).await + }) + .await +} + +async fn wait_for_forward_listener_with_probe( + spec: &ForwardSpec, + wait_for: Duration, + mut probe: F, +) -> Result<()> +where + F: FnMut(ForwardSpec) -> Fut, + Fut: Future>, +{ let deadline = tokio::time::Instant::now() + wait_for; loop { - let probe_error = match probe_forward_listener(spec).await { + let probe_error = match probe(spec.clone()).await { Ok(()) => return Ok(()), Err(err) => err, }; @@ -1816,17 +1832,17 @@ mod tests { } #[tokio::test] - async fn wait_for_forward_listener_rejects_missing_listener() { - let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap(); - let port = listener.local_addr().unwrap().port(); - drop(listener); - let spec = ForwardSpec::new(port); + async fn wait_for_forward_listener_reports_failed_probe() { + let spec = ForwardSpec::new(12345); - let err = wait_for_forward_listener(&spec, Duration::from_millis(20)) - .await - .unwrap_err(); + let err = wait_for_forward_listener_with_probe(&spec, Duration::ZERO, |_| { + std::future::ready(Err("connection refused".to_string())) + }) + .await + .unwrap_err(); let text = format!("{err:?}"); assert!(text.contains("local forward listener did not open")); + assert!(text.contains("connection refused")); } #[test] diff --git a/crates/openshell-gateway-interceptors/src/runtime.rs b/crates/openshell-gateway-interceptors/src/runtime.rs index 04c9aba603..d510956aed 100644 --- a/crates/openshell-gateway-interceptors/src/runtime.rs +++ b/crates/openshell-gateway-interceptors/src/runtime.rs @@ -911,7 +911,6 @@ mod tests { .with_ansi(false) .without_time(); let subscriber = tracing_subscriber::registry().with(fmt_layer); - let dispatch = tracing::Dispatch::new(subscriber); let plan = BindingPlan { interceptor_name: "test".to_string(), binding_id: "binding".to_string(), @@ -940,7 +939,11 @@ mod tests { ..InterceptorResult::default() }; - tracing::dispatcher::with_default(&dispatch, || { + tracing::subscriber::with_default(subscriber, || { + // Other parallel tests may register this callsite while no subscriber + // is active. Refresh the process-wide cache after installing this + // thread-local subscriber so the event cannot remain disabled. + tracing::callsite::rebuild_interest_cache(); emit_evaluation_log(&plan, &result, "allow", 2); }); diff --git a/crates/openshell-server/src/multiplex.rs b/crates/openshell-server/src/multiplex.rs index 83e27d27cb..a58f66c916 100644 --- a/crates/openshell-server/src/multiplex.rs +++ b/crates/openshell-server/src/multiplex.rs @@ -1724,16 +1724,21 @@ mod tests { .with_span_events(FmtSpan::CLOSE); let subscriber = tracing_subscriber::registry().with(fmt_layer); - let _guard = tracing::subscriber::set_default(subscriber); - - let req = Request::builder() - .uri("/test-path") - .header("x-request-id", "trace-test-id-12345") - .body(Empty::::new()) - .unwrap(); - let span = make_request_span(&req); - drop(span.enter()); - drop(span); + tracing::subscriber::with_default(subscriber, || { + // Other parallel tests may register this callsite while no subscriber + // is active. Refresh the process-wide cache after installing this + // thread-local subscriber so the span cannot remain disabled. + tracing::callsite::rebuild_interest_cache(); + + let req = Request::builder() + .uri("/test-path") + .header("x-request-id", "trace-test-id-12345") + .body(Empty::::new()) + .unwrap(); + let span = make_request_span(&req); + drop(span.enter()); + drop(span); + }); let output = String::from_utf8(log_buf.lock().unwrap().clone()).unwrap(); assert!( diff --git a/crates/openshell-supervisor-network/src/procfs.rs b/crates/openshell-supervisor-network/src/procfs.rs index 3ac8dbe141..8bc2fbb110 100644 --- a/crates/openshell-supervisor-network/src/procfs.rs +++ b/crates/openshell-supervisor-network/src/procfs.rs @@ -9,6 +9,9 @@ use miette::Result; #[cfg(target_os = "linux")] use std::collections::HashSet; +use std::net::SocketAddr; +#[cfg(target_os = "linux")] +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::path::Path; #[cfg(target_os = "linux")] use std::path::PathBuf; @@ -40,6 +43,46 @@ pub struct TcpPeerSocketOwners { pub owners: Vec, } +/// TCP endpoints for a workload connection accepted by the sandbox proxy. +/// +/// `/proc//net/tcp{,6}` reports the socket from the workload side, so its +/// local endpoint should match `workload` and its remote endpoint should match +/// `proxy`. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct WorkloadProxyTcpConnection { + pub workload: SocketAddr, + pub proxy: SocketAddr, +} + +impl WorkloadProxyTcpConnection { + pub fn new(workload: SocketAddr, proxy: SocketAddr) -> Self { + Self { workload, proxy } + } +} + +impl std::fmt::Display for WorkloadProxyTcpConnection { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} -> {}", self.workload, self.proxy) + } +} + +#[cfg(target_os = "linux")] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ProcTcpAddressFamily { + Ipv4, + Ipv6, +} + +#[cfg(target_os = "linux")] +impl ProcTcpAddressFamily { + fn table_name(self) -> &'static str { + match self { + Self::Ipv4 => "tcp", + Self::Ipv6 => "tcp6", + } + } +} + #[cfg(target_os = "linux")] #[derive(Clone, Debug, Eq, PartialEq)] struct DescendantPid { @@ -116,11 +159,14 @@ pub fn binary_path(pid: i32) -> Result { /// Resolve the binary path of the TCP peer inside a sandbox network namespace. /// /// Uses `/proc//net/tcp` to find the socket inode for the given -/// ephemeral port, then scans the entrypoint process tree to find which PID owns -/// that socket, and finally reads `/proc//exe` to get the binary path. +/// connection tuple, then scans the entrypoint process tree to find which PID +/// owns that socket, and finally reads `/proc//exe` to get the binary path. #[cfg(target_os = "linux")] -pub fn resolve_tcp_peer_binary(entrypoint_pid: u32, peer_port: u16) -> Result { - let owner = resolve_single_tcp_peer_owner(entrypoint_pid, peer_port)?; +pub fn resolve_tcp_peer_binary( + entrypoint_pid: u32, + connection: WorkloadProxyTcpConnection, +) -> Result { + let owner = resolve_single_tcp_peer_owner(entrypoint_pid, connection)?; binary_path(owner.pid.cast_signed()) } @@ -132,17 +178,20 @@ pub fn resolve_tcp_peer_binary(entrypoint_pid: u32, peer_port: u16) -> Result Result { - let inode = parse_proc_net_tcp(entrypoint_pid, peer_port)?; + let inode = parse_proc_net_tcp(entrypoint_pid, connection)?; let owners = find_socket_inode_owners(inode, entrypoint_pid)?; Ok(TcpPeerSocketOwners { inode, owners }) } /// Resolve exactly one owner for the TCP peer, failing closed on ambiguity. #[cfg(target_os = "linux")] -fn resolve_single_tcp_peer_owner(entrypoint_pid: u32, peer_port: u16) -> Result { - let socket_owners = resolve_tcp_peer_socket_owners(entrypoint_pid, peer_port)?; +fn resolve_single_tcp_peer_owner( + entrypoint_pid: u32, + connection: WorkloadProxyTcpConnection, +) -> Result { + let socket_owners = resolve_tcp_peer_socket_owners(entrypoint_pid, connection)?; match socket_owners.owners.as_slice() { [owner] => Ok(owner.clone()), owners => { @@ -164,8 +213,11 @@ fn resolve_single_tcp_peer_owner(entrypoint_pid: u32, peer_port: u16) -> Result< /// /// Needed for the ancestor walk: we must know the PID to walk `/proc//status` `PPid` chain. #[cfg(target_os = "linux")] -pub fn resolve_tcp_peer_identity(entrypoint_pid: u32, peer_port: u16) -> Result<(PathBuf, u32)> { - let owner = resolve_single_tcp_peer_owner(entrypoint_pid, peer_port)?; +pub fn resolve_tcp_peer_identity( + entrypoint_pid: u32, + connection: WorkloadProxyTcpConnection, +) -> Result<(PathBuf, u32)> { + let owner = resolve_single_tcp_peer_owner(entrypoint_pid, connection)?; let path = binary_path(owner.pid.cast_signed())?; Ok((path, owner.pid)) } @@ -274,7 +326,7 @@ pub fn collect_cmdline_paths(pid: u32, stop_pid: u32, exclude: &[PathBuf]) -> Ve } /// Parse `/proc//net/tcp` (and `/proc//net/tcp6`) to find the socket -/// inode for a given local port. +/// inode for a complete local-to-remote TCP tuple. /// /// Checks both IPv4 and IPv6 tables because some clients (notably gRPC C-core) /// use `AF_INET6` sockets with IPv4-mapped addresses even for IPv4 connections. @@ -288,52 +340,96 @@ pub fn collect_cmdline_paths(pid: u32, stop_pid: u32, exclude: &[PathBuf]) -> Ve /// - State `01` = ESTABLISHED /// - Inode is field index 9 (0-indexed) #[cfg(target_os = "linux")] -fn parse_proc_net_tcp(pid: u32, peer_port: u16) -> Result { +fn parse_proc_net_tcp(pid: u32, connection: WorkloadProxyTcpConnection) -> Result { // Check IPv4 first (most common), then IPv6. - for suffix in &["tcp", "tcp6"] { - let path = format!("/proc/{pid}/net/{suffix}"); + for address_family in [ProcTcpAddressFamily::Ipv4, ProcTcpAddressFamily::Ipv6] { + let path = format!("/proc/{pid}/net/{}", address_family.table_name()); let Ok(content) = std::fs::read_to_string(&path) else { continue; }; - for line in content.lines().skip(1) { - let fields: Vec<&str> = line.split_whitespace().collect(); - if fields.len() < 10 { - continue; - } + if let Some(inode) = find_proc_net_tcp_inode(&content, address_family, connection)? { + return Ok(inode); + } + } - // Parse local_address to extract port. - // IPv4 format: AABBCCDD:PORT - // IPv6 format: 00000000000000000000000000000000:PORT - let local_addr = fields[1]; - let local_port = match local_addr.rsplit_once(':') { - Some((_, port_hex)) => u16::from_str_radix(port_hex, 16).unwrap_or(0), - None => continue, - }; - - // Check state is ESTABLISHED (01) - let state = fields[3]; - if state != "01" { - continue; - } + Err(miette::miette!( + "No ESTABLISHED TCP connection found for {connection} in /proc/{pid}/net/tcp{{,6}}" + )) +} - if local_port == peer_port { - let inode: u64 = fields[9] - .parse() - .map_err(|_| miette::miette!("Failed to parse inode from {}", fields[9]))?; - if inode == 0 { - continue; - } - return Ok(inode); +#[cfg(target_os = "linux")] +fn find_proc_net_tcp_inode( + content: &str, + address_family: ProcTcpAddressFamily, + connection: WorkloadProxyTcpConnection, +) -> Result> { + for line in content.lines().skip(1) { + let fields: Vec<&str> = line.split_whitespace().collect(); + if fields.len() < 10 || fields[3] != "01" { + continue; + } + + let Some(local_addr) = parse_proc_socket_addr(fields[1], address_family) else { + continue; + }; + let Some(remote_addr) = parse_proc_socket_addr(fields[2], address_family) else { + continue; + }; + + if socket_addrs_match(local_addr, connection.workload) + && socket_addrs_match(remote_addr, connection.proxy) + { + let inode: u64 = fields[9] + .parse() + .map_err(|_| miette::miette!("Failed to parse inode from {}", fields[9]))?; + if inode != 0 { + return Ok(Some(inode)); } } } - Err(miette::miette!( - "No ESTABLISHED TCP connection found for port {} in /proc/{}/net/tcp{{,6}}", - peer_port, - pid - )) + Ok(None) +} + +#[cfg(target_os = "linux")] +fn parse_proc_socket_addr(value: &str, address_family: ProcTcpAddressFamily) -> Option { + let (address, port) = value.rsplit_once(':')?; + let port = u16::from_str_radix(port, 16).ok()?; + + let ip = match address_family { + ProcTcpAddressFamily::Ipv4 => { + let address = u32::from_str_radix(address, 16).ok()?; + IpAddr::V4(Ipv4Addr::from(address.to_le_bytes())) + } + ProcTcpAddressFamily::Ipv6 => { + if address.len() != 32 { + return None; + } + let mut bytes = [0u8; 16]; + for (index, chunk) in address.as_bytes().chunks_exact(8).enumerate() { + let chunk = std::str::from_utf8(chunk).ok()?; + let word = u32::from_str_radix(chunk, 16).ok()?; + bytes[index * 4..index * 4 + 4].copy_from_slice(&word.to_le_bytes()); + } + IpAddr::V6(Ipv6Addr::from(bytes)) + } + }; + + Some(SocketAddr::new(ip, port)) +} + +#[cfg(target_os = "linux")] +fn socket_addrs_match(left: SocketAddr, right: SocketAddr) -> bool { + left.port() == right.port() && normalize_ip(left.ip()) == normalize_ip(right.ip()) +} + +#[cfg(target_os = "linux")] +fn normalize_ip(ip: IpAddr) -> IpAddr { + match ip { + IpAddr::V6(ipv6) => ipv6.to_ipv4_mapped().map_or(IpAddr::V6(ipv6), IpAddr::V4), + ipv4 @ IpAddr::V4(_) => ipv4, + } } /// Scan `/proc` to find every PID that owns a given socket inode. @@ -756,6 +852,38 @@ mod tests { assert_eq!(pids.len(), unique.len()); } + #[cfg(target_os = "linux")] + #[test] + fn parse_proc_socket_addr_decodes_ipv4_and_mapped_ipv6() { + let ipv4 = parse_proc_socket_addr("0100007F:C350", ProcTcpAddressFamily::Ipv4).unwrap(); + let mapped_ipv6 = parse_proc_socket_addr( + "0000000000000000FFFF00000100007F:C350", + ProcTcpAddressFamily::Ipv6, + ) + .unwrap(); + + assert_eq!(ipv4, "127.0.0.1:50000".parse().unwrap()); + assert!(socket_addrs_match(ipv4, mapped_ipv6)); + } + + #[cfg(target_os = "linux")] + #[test] + fn find_proc_net_tcp_inode_matches_complete_tuple() { + let content = "\ + sl local_address rem_address st tx_queue:rx_queue tr:tm->when retrnsmt uid timeout inode\n\ + 0: 0100007F:C350 0100007F:1F91 01 00000000:00000000 00:00000000 00000000 1000 0 11111\n\ + 1: 0100007F:C350 0100007F:1F90 01 00000000:00000000 00:00000000 00000000 1000 0 22222\n"; + let connection = WorkloadProxyTcpConnection::new( + "127.0.0.1:50000".parse().unwrap(), + "127.0.0.1:8080".parse().unwrap(), + ); + + let inode = + find_proc_net_tcp_inode(content, ProcTcpAddressFamily::Ipv4, connection).unwrap(); + + assert_eq!(inode, Some(22222)); + } + #[cfg(target_os = "linux")] #[test] fn resolve_tcp_peer_socket_owners_returns_all_forked_socket_holders() { @@ -774,9 +902,10 @@ mod tests { } 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 proxy_addr = listener.local_addr().unwrap(); + let stream = TcpStream::connect(proxy_addr).expect("connect"); + let workload_addr = stream.local_addr().unwrap(); + let connection = WorkloadProxyTcpConnection::new(workload_addr, proxy_addr); let (_accepted, _) = listener.accept().expect("accept"); // libc/syscall FFI requires unsafe @@ -797,7 +926,7 @@ mod tests { let entrypoint_pid = std::process::id(); let deadline = Instant::now() + Duration::from_secs(5); let owners = loop { - let owners = resolve_tcp_peer_socket_owners(entrypoint_pid, peer_port) + let owners = resolve_tcp_peer_socket_owners(entrypoint_pid, connection) .expect("resolve socket owners"); let owner_pids = owners .owners diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index c0cbe3005d..efee101aa7 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -733,8 +733,9 @@ async fn handle_tcp_connection( return Ok(()); } - let peer_addr = client.peer_addr().into_diagnostic()?; - let _local_addr = client.local_addr().into_diagnostic()?; + let workload_addr = client.peer_addr().into_diagnostic()?; + let proxy_addr = client.local_addr().into_diagnostic()?; + let connection = crate::procfs::WorkloadProxyTcpConnection::new(workload_addr, proxy_addr); // Evaluate OPA policy with process-identity binding. // Wrapped in spawn_blocking because identity resolution does heavy sync I/O: @@ -745,7 +746,7 @@ async fn handle_tcp_connection( let host_clone = host_lc.clone(); let decision = tokio::task::spawn_blocking(move || { evaluate_opa_tcp( - peer_addr, + connection, &opa_clone, &cache_clone, &pid_clone, @@ -803,7 +804,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -886,7 +887,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -940,7 +941,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -988,7 +989,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1040,7 +1041,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1090,7 +1091,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1145,7 +1146,7 @@ async fn handle_tcp_connection( .severity(SeverityId::High) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1201,7 +1202,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Informational) .status(StatusId::Success) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1394,7 +1395,7 @@ async fn handle_tcp_connection( .severity(SeverityId::High) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1517,7 +1518,7 @@ async fn handle_tcp_connection( .severity(SeverityId::Medium) .status(StatusId::Failure) .dst_endpoint(Endpoint::from_domain(&host_lc, port)) - .src_endpoint_addr(peer_addr.ip(), peer_addr.port()) + .src_endpoint_addr(workload_addr.ip(), workload_addr.port()) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -1718,16 +1719,16 @@ fn collect_ancestor_identities(start_pid: u32, stop_pid: u32) -> Vec<(u32, PathB #[cfg(target_os = "linux")] fn resolve_process_identity( entrypoint_pid: u32, - peer_port: u16, + connection: crate::procfs::WorkloadProxyTcpConnection, identity_cache: &BinaryIdentityCache, ) -> std::result::Result { - let socket_owners = crate::procfs::resolve_tcp_peer_socket_owners(entrypoint_pid, peer_port) + let socket_owners = crate::procfs::resolve_tcp_peer_socket_owners(entrypoint_pid, connection) .map_err(|e| IdentityError { - reason: format!("failed to resolve peer binary: {e}"), - binary: None, - binary_pid: None, - ancestors: vec![], - })?; + 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 { @@ -1786,7 +1787,7 @@ fn resolve_process_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, + connection: crate::procfs::WorkloadProxyTcpConnection, engine: &OpaEngine, identity_cache: &BinaryIdentityCache, entrypoint_pid: &AtomicU32, @@ -1833,9 +1834,7 @@ fn evaluate_opa_tcp( }; 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) { + let identity = match resolve_process_identity(proc_net_anchor_pid, connection, identity_cache) { Ok(id) => id, Err(err) => { return deny( @@ -1938,7 +1937,7 @@ fn evaluate_endpoint_only_opa(engine: &OpaEngine, host: &str, port: u16) -> Conn /// Non-Linux stub: OPA identity binding requires /proc. #[cfg(not(target_os = "linux"))] fn evaluate_opa_tcp( - _peer_addr: SocketAddr, + _connection: crate::procfs::WorkloadProxyTcpConnection, engine: &OpaEngine, _identity_cache: &BinaryIdentityCache, _entrypoint_pid: &AtomicU32, @@ -3645,8 +3644,9 @@ async fn handle_forward_proxy( 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 workload_addr = client.peer_addr().into_diagnostic()?; + let proxy_addr = client.local_addr().into_diagnostic()?; + let connection = crate::procfs::WorkloadProxyTcpConnection::new(workload_addr, proxy_addr); let opa_clone = opa_engine.clone(); let cache_clone = identity_cache.clone(); @@ -3654,7 +3654,7 @@ async fn handle_forward_proxy( let host_clone = host_lc.clone(); let decision = tokio::task::spawn_blocking(move || { evaluate_opa_tcp( - peer_addr, + connection, &opa_clone, &cache_clone, &pid_clone, @@ -3710,7 +3710,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -3971,7 +3971,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4213,7 +4213,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4294,7 +4294,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4351,7 +4351,10 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip( + workload_addr.ip(), + workload_addr.port(), + )) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4403,7 +4406,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4460,7 +4463,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4514,7 +4517,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4589,7 +4592,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -4796,7 +4799,7 @@ async fn handle_forward_proxy( 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())) + .src_endpoint(Endpoint::from_ip(workload_addr.ip(), workload_addr.port())) .actor_process( Process::from_bypass(&binary_str, &pid_str, &ancestors_str) .with_cmd_line(&cmdline_str), @@ -9626,9 +9629,9 @@ network_policies: /// 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. + /// 5. Accept the connection on the listener side and capture both socket + /// endpoints — 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 @@ -9656,7 +9659,7 @@ network_policies: // 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(); + let proxy_addr = listener.local_addr().unwrap(); // 2. Copy /bin/bash to a temp path. let tmp = tempfile::TempDir::new().unwrap(); @@ -9676,8 +9679,10 @@ network_policies: // 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 script = format!( + "exec 3<>/dev/tcp/127.0.0.1/{}; read -r -t 30 _ <&3 || true", + proxy_addr.port() + ); let mut child = Command::new(&bash_v1) .arg("-c") .arg(&script) @@ -9687,9 +9692,9 @@ network_policies: .spawn() .expect("spawn hotswap-bash child"); - // 5. Accept on the listener side, capture the peer port. + // 5. Accept on the listener side and capture the peer endpoint. listener.set_nonblocking(false).expect("blocking listener"); - let (mut stream, peer_addr) = match listener.accept() { + let (mut stream, workload_addr) = match listener.accept() { Ok(pair) => pair, Err(e) => { let _ = child.kill(); @@ -9697,7 +9702,7 @@ network_policies: panic!("failed to accept child connection: {e}"); } }; - let peer_port = peer_addr.port(); + let connection = crate::procfs::WorkloadProxyTcpConnection::new(workload_addr, proxy_addr); // Drain any spurious data; we just need the socket open. stream .set_read_timeout(Some(Duration::from_millis(50))) @@ -9724,7 +9729,7 @@ network_policies: // 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 result = resolve_process_identity(test_pid, connection, &cache); let child_pid = child.id(); // Always clean up the child before asserting so a failure doesn't @@ -9799,9 +9804,10 @@ network_policies: } 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 proxy_addr = listener.local_addr().unwrap(); + let stream = TcpStream::connect(proxy_addr).expect("connect"); + let workload_addr = stream.local_addr().unwrap(); + let connection = crate::procfs::WorkloadProxyTcpConnection::new(workload_addr, proxy_addr); let (_accepted, _) = listener.accept().expect("accept"); let fd = stream.as_raw_fd(); @@ -9857,7 +9863,7 @@ network_policies: let cache = BinaryIdentityCache::new(); - let mut result = resolve_process_identity(entrypoint_pid, peer_port, &cache); + let mut result = resolve_process_identity(entrypoint_pid, connection, &cache); for _ in 0..10 { match &result { Err(err) @@ -9866,7 +9872,7 @@ network_policies: { // /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); + result = resolve_process_identity(entrypoint_pid, connection, &cache); } Ok(_) => { // On arm64 under heavy CI load the /proc fd scan can transiently @@ -9874,7 +9880,7 @@ network_policies: // 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); + result = resolve_process_identity(entrypoint_pid, connection, &cache); } _ => break, }