From 45d470a8d89f964f72c6ec05309145a6f8d0280e Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Wed, 22 Jul 2026 15:02:36 -0700 Subject: [PATCH 1/3] fix: eliminate parallel Rust test flakes Signed-off-by: Piotr Mlocek --- architecture/sandbox.md | 5 +- crates/openshell-cli/src/ssh.rs | 34 +++- .../src/runtime.rs | 7 +- crates/openshell-server/src/multiplex.rs | 25 ++- .../src/procfs.rs | 181 +++++++++++++----- .../openshell-supervisor-network/src/proxy.rs | 68 ++++--- 6 files changed, 221 insertions(+), 99 deletions(-) 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..c2f57d6aeb 100644 --- a/crates/openshell-supervisor-network/src/procfs.rs +++ b/crates/openshell-supervisor-network/src/procfs.rs @@ -9,6 +9,8 @@ use miette::Result; #[cfg(target_os = "linux")] use std::collections::HashSet; +#[cfg(target_os = "linux")] +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::path::Path; #[cfg(target_os = "linux")] use std::path::PathBuf; @@ -116,11 +118,15 @@ 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, + peer_addr: SocketAddr, + proxy_addr: SocketAddr, +) -> Result { + let owner = resolve_single_tcp_peer_owner(entrypoint_pid, peer_addr, proxy_addr)?; binary_path(owner.pid.cast_signed()) } @@ -132,17 +138,22 @@ 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, peer_addr, proxy_addr)?; 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, + peer_addr: SocketAddr, + proxy_addr: SocketAddr, +) -> Result { + let socket_owners = resolve_tcp_peer_socket_owners(entrypoint_pid, peer_addr, proxy_addr)?; match socket_owners.owners.as_slice() { [owner] => Ok(owner.clone()), owners => { @@ -164,8 +175,12 @@ 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, + peer_addr: SocketAddr, + proxy_addr: SocketAddr, +) -> Result<(PathBuf, u32)> { + let owner = resolve_single_tcp_peer_owner(entrypoint_pid, peer_addr, proxy_addr)?; let path = binary_path(owner.pid.cast_signed())?; Ok((path, owner.pid)) } @@ -274,7 +289,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,54 +303,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, peer_addr: SocketAddr, proxy_addr: SocketAddr) -> Result { // Check IPv4 first (most common), then IPv6. - for suffix in &["tcp", "tcp6"] { + for (suffix, ipv6) in [("tcp", false), ("tcp6", true)] { let path = format!("/proc/{pid}/net/{suffix}"); 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; - } - - // 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; - } - - 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); - } + if let Some(inode) = find_proc_net_tcp_inode(&content, ipv6, peer_addr, proxy_addr)? { + return Ok(inode); } } Err(miette::miette!( - "No ESTABLISHED TCP connection found for port {} in /proc/{}/net/tcp{{,6}}", - peer_port, + "No ESTABLISHED TCP connection found for {peer_addr} -> {proxy_addr} in /proc/{}/net/tcp{{,6}}", pid )) } +#[cfg(target_os = "linux")] +fn find_proc_net_tcp_inode( + content: &str, + ipv6: bool, + peer_addr: SocketAddr, + proxy_addr: SocketAddr, +) -> 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], ipv6) else { + continue; + }; + let Some(remote_addr) = parse_proc_socket_addr(fields[2], ipv6) else { + continue; + }; + + if socket_addrs_match(local_addr, peer_addr) && socket_addrs_match(remote_addr, proxy_addr) + { + let inode: u64 = fields[9] + .parse() + .map_err(|_| miette::miette!("Failed to parse inode from {}", fields[9]))?; + if inode != 0 { + return Ok(Some(inode)); + } + } + } + + Ok(None) +} + +#[cfg(target_os = "linux")] +fn parse_proc_socket_addr(value: &str, ipv6: bool) -> Option { + let (address, port) = value.rsplit_once(':')?; + let port = u16::from_str_radix(port, 16).ok()?; + + let ip = if 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)) + } else { + let address = u32::from_str_radix(address, 16).ok()?; + IpAddr::V4(Ipv4Addr::from(address.to_le_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. /// /// First scans descendants of `entrypoint_pid` (most likely owners), then falls @@ -756,6 +813,32 @@ 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", false).unwrap(); + let mapped_ipv6 = + parse_proc_socket_addr("0000000000000000FFFF00000100007F:C350", true).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 peer_addr = "127.0.0.1:50000".parse().unwrap(); + let proxy_addr = "127.0.0.1:8080".parse().unwrap(); + + let inode = find_proc_net_tcp_inode(content, false, peer_addr, proxy_addr).unwrap(); + + assert_eq!(inode, Some(22222)); + } + #[cfg(target_os = "linux")] #[test] fn resolve_tcp_peer_socket_owners_returns_all_forked_socket_holders() { @@ -774,9 +857,9 @@ 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 peer_addr = stream.local_addr().unwrap(); let (_accepted, _) = listener.accept().expect("accept"); // libc/syscall FFI requires unsafe @@ -797,7 +880,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, peer_addr, proxy_addr) .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..5f5d11e817 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -734,7 +734,7 @@ async fn handle_tcp_connection( } let peer_addr = client.peer_addr().into_diagnostic()?; - let _local_addr = client.local_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: @@ -746,6 +746,7 @@ async fn handle_tcp_connection( let decision = tokio::task::spawn_blocking(move || { evaluate_opa_tcp( peer_addr, + local_addr, &opa_clone, &cache_clone, &pid_clone, @@ -1718,16 +1719,18 @@ 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, + peer_addr: SocketAddr, + proxy_addr: SocketAddr, 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 socket_owners = + crate::procfs::resolve_tcp_peer_socket_owners(entrypoint_pid, peer_addr, proxy_addr) + .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 { @@ -1787,6 +1790,7 @@ fn resolve_process_identity( #[cfg(target_os = "linux")] fn evaluate_opa_tcp( peer_addr: SocketAddr, + proxy_addr: SocketAddr, engine: &OpaEngine, identity_cache: &BinaryIdentityCache, entrypoint_pid: &AtomicU32, @@ -1833,9 +1837,12 @@ 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, + peer_addr, + proxy_addr, + identity_cache, + ) { Ok(id) => id, Err(err) => { return deny( @@ -1939,6 +1946,7 @@ fn evaluate_endpoint_only_opa(engine: &OpaEngine, host: &str, port: u16) -> Conn #[cfg(not(target_os = "linux"))] fn evaluate_opa_tcp( _peer_addr: SocketAddr, + _proxy_addr: SocketAddr, engine: &OpaEngine, _identity_cache: &BinaryIdentityCache, _entrypoint_pid: &AtomicU32, @@ -3646,7 +3654,7 @@ async fn handle_forward_proxy( // 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 local_addr = client.local_addr().into_diagnostic()?; let opa_clone = opa_engine.clone(); let cache_clone = identity_cache.clone(); @@ -3655,6 +3663,7 @@ async fn handle_forward_proxy( let decision = tokio::task::spawn_blocking(move || { evaluate_opa_tcp( peer_addr, + local_addr, &opa_clone, &cache_clone, &pid_clone, @@ -9626,9 +9635,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 +9665,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 +9685,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,7 +9698,7 @@ 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() { Ok(pair) => pair, @@ -9697,7 +9708,6 @@ network_policies: 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))) @@ -9724,7 +9734,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, peer_addr, proxy_addr, &cache); let child_pid = child.id(); // Always clean up the child before asserting so a failure doesn't @@ -9799,9 +9809,9 @@ 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 peer_addr = stream.local_addr().unwrap(); let (_accepted, _) = listener.accept().expect("accept"); let fd = stream.as_raw_fd(); @@ -9857,7 +9867,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, peer_addr, proxy_addr, &cache); for _ in 0..10 { match &result { Err(err) @@ -9866,7 +9876,8 @@ 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, peer_addr, proxy_addr, &cache); } Ok(_) => { // On arm64 under heavy CI load the /proc fd scan can transiently @@ -9874,7 +9885,8 @@ 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, peer_addr, proxy_addr, &cache); } _ => break, } From 24b87cab1fba27c3d7269e30e8eeda22380d5692 Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Thu, 23 Jul 2026 11:23:18 +0200 Subject: [PATCH 2/3] refactor(supervisor-network): name workload proxy TCP connection Signed-off-by: Evan Lezar --- .../src/procfs.rs | 72 +++++++----- .../openshell-supervisor-network/src/proxy.rs | 106 +++++++++--------- 2 files changed, 97 insertions(+), 81 deletions(-) diff --git a/crates/openshell-supervisor-network/src/procfs.rs b/crates/openshell-supervisor-network/src/procfs.rs index c2f57d6aeb..0e6111967c 100644 --- a/crates/openshell-supervisor-network/src/procfs.rs +++ b/crates/openshell-supervisor-network/src/procfs.rs @@ -9,8 +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, SocketAddr}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::path::Path; #[cfg(target_os = "linux")] use std::path::PathBuf; @@ -42,6 +43,29 @@ 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, Debug, Eq, PartialEq)] struct DescendantPid { @@ -123,10 +147,9 @@ pub fn binary_path(pid: i32) -> Result { #[cfg(target_os = "linux")] pub fn resolve_tcp_peer_binary( entrypoint_pid: u32, - peer_addr: SocketAddr, - proxy_addr: SocketAddr, + connection: WorkloadProxyTcpConnection, ) -> Result { - let owner = resolve_single_tcp_peer_owner(entrypoint_pid, peer_addr, proxy_addr)?; + let owner = resolve_single_tcp_peer_owner(entrypoint_pid, connection)?; binary_path(owner.pid.cast_signed()) } @@ -138,10 +161,9 @@ pub fn resolve_tcp_peer_binary( #[cfg(target_os = "linux")] pub fn resolve_tcp_peer_socket_owners( entrypoint_pid: u32, - peer_addr: SocketAddr, - proxy_addr: SocketAddr, + connection: WorkloadProxyTcpConnection, ) -> Result { - let inode = parse_proc_net_tcp(entrypoint_pid, peer_addr, proxy_addr)?; + let inode = parse_proc_net_tcp(entrypoint_pid, connection)?; let owners = find_socket_inode_owners(inode, entrypoint_pid)?; Ok(TcpPeerSocketOwners { inode, owners }) } @@ -150,10 +172,9 @@ pub fn resolve_tcp_peer_socket_owners( #[cfg(target_os = "linux")] fn resolve_single_tcp_peer_owner( entrypoint_pid: u32, - peer_addr: SocketAddr, - proxy_addr: SocketAddr, + connection: WorkloadProxyTcpConnection, ) -> Result { - let socket_owners = resolve_tcp_peer_socket_owners(entrypoint_pid, peer_addr, proxy_addr)?; + let socket_owners = resolve_tcp_peer_socket_owners(entrypoint_pid, connection)?; match socket_owners.owners.as_slice() { [owner] => Ok(owner.clone()), owners => { @@ -177,10 +198,9 @@ fn resolve_single_tcp_peer_owner( #[cfg(target_os = "linux")] pub fn resolve_tcp_peer_identity( entrypoint_pid: u32, - peer_addr: SocketAddr, - proxy_addr: SocketAddr, + connection: WorkloadProxyTcpConnection, ) -> Result<(PathBuf, u32)> { - let owner = resolve_single_tcp_peer_owner(entrypoint_pid, peer_addr, proxy_addr)?; + let owner = resolve_single_tcp_peer_owner(entrypoint_pid, connection)?; let path = binary_path(owner.pid.cast_signed())?; Ok((path, owner.pid)) } @@ -303,7 +323,7 @@ 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_addr: SocketAddr, proxy_addr: SocketAddr) -> Result { +fn parse_proc_net_tcp(pid: u32, connection: WorkloadProxyTcpConnection) -> Result { // Check IPv4 first (most common), then IPv6. for (suffix, ipv6) in [("tcp", false), ("tcp6", true)] { let path = format!("/proc/{pid}/net/{suffix}"); @@ -311,14 +331,13 @@ fn parse_proc_net_tcp(pid: u32, peer_addr: SocketAddr, proxy_addr: SocketAddr) - continue; }; - if let Some(inode) = find_proc_net_tcp_inode(&content, ipv6, peer_addr, proxy_addr)? { + if let Some(inode) = find_proc_net_tcp_inode(&content, ipv6, connection)? { return Ok(inode); } } Err(miette::miette!( - "No ESTABLISHED TCP connection found for {peer_addr} -> {proxy_addr} in /proc/{}/net/tcp{{,6}}", - pid + "No ESTABLISHED TCP connection found for {connection} in /proc/{pid}/net/tcp{{,6}}" )) } @@ -326,8 +345,7 @@ fn parse_proc_net_tcp(pid: u32, peer_addr: SocketAddr, proxy_addr: SocketAddr) - fn find_proc_net_tcp_inode( content: &str, ipv6: bool, - peer_addr: SocketAddr, - proxy_addr: SocketAddr, + connection: WorkloadProxyTcpConnection, ) -> Result> { for line in content.lines().skip(1) { let fields: Vec<&str> = line.split_whitespace().collect(); @@ -342,7 +360,8 @@ fn find_proc_net_tcp_inode( continue; }; - if socket_addrs_match(local_addr, peer_addr) && socket_addrs_match(remote_addr, proxy_addr) + if socket_addrs_match(local_addr, connection.workload) + && socket_addrs_match(remote_addr, connection.proxy) { let inode: u64 = fields[9] .parse() @@ -831,10 +850,12 @@ mod tests { 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 peer_addr = "127.0.0.1:50000".parse().unwrap(); - let proxy_addr = "127.0.0.1:8080".parse().unwrap(); + 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, false, peer_addr, proxy_addr).unwrap(); + let inode = find_proc_net_tcp_inode(content, false, connection).unwrap(); assert_eq!(inode, Some(22222)); } @@ -859,7 +880,8 @@ mod tests { let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener"); let proxy_addr = listener.local_addr().unwrap(); let stream = TcpStream::connect(proxy_addr).expect("connect"); - let peer_addr = stream.local_addr().unwrap(); + 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 @@ -880,7 +902,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_addr, proxy_addr) + 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 5f5d11e817..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,8 +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, - local_addr, + connection, &opa_clone, &cache_clone, &pid_clone, @@ -804,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), @@ -887,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), @@ -941,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), @@ -989,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), @@ -1041,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), @@ -1091,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), @@ -1146,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), @@ -1202,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), @@ -1395,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), @@ -1518,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), @@ -1719,18 +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_addr: SocketAddr, - proxy_addr: SocketAddr, + connection: crate::procfs::WorkloadProxyTcpConnection, identity_cache: &BinaryIdentityCache, ) -> std::result::Result { - let socket_owners = - crate::procfs::resolve_tcp_peer_socket_owners(entrypoint_pid, peer_addr, proxy_addr) - .map_err(|e| IdentityError { - reason: format!("failed to resolve peer binary: {e}"), - binary: None, - binary_pid: None, - ancestors: vec![], - })?; + 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![], + })?; let mut identities = Vec::with_capacity(socket_owners.owners.len()); for owner in &socket_owners.owners { @@ -1789,8 +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, - proxy_addr: SocketAddr, + connection: crate::procfs::WorkloadProxyTcpConnection, engine: &OpaEngine, identity_cache: &BinaryIdentityCache, entrypoint_pid: &AtomicU32, @@ -1837,12 +1834,7 @@ fn evaluate_opa_tcp( }; let total_start = std::time::Instant::now(); - let identity = match resolve_process_identity( - proc_net_anchor_pid, - peer_addr, - proxy_addr, - identity_cache, - ) { + let identity = match resolve_process_identity(proc_net_anchor_pid, connection, identity_cache) { Ok(id) => id, Err(err) => { return deny( @@ -1945,8 +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, - _proxy_addr: SocketAddr, + _connection: crate::procfs::WorkloadProxyTcpConnection, engine: &OpaEngine, _identity_cache: &BinaryIdentityCache, _entrypoint_pid: &AtomicU32, @@ -3653,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(); @@ -3662,8 +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, - local_addr, + connection, &opa_clone, &cache_clone, &pid_clone, @@ -3719,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), @@ -3980,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), @@ -4222,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), @@ -4303,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), @@ -4360,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), @@ -4412,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), @@ -4469,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), @@ -4523,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), @@ -4598,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), @@ -4805,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), @@ -9700,7 +9694,7 @@ network_policies: // 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(); @@ -9708,6 +9702,7 @@ network_policies: panic!("failed to accept child connection: {e}"); } }; + 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))) @@ -9734,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_addr, proxy_addr, &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 @@ -9811,7 +9806,8 @@ network_policies: let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener"); let proxy_addr = listener.local_addr().unwrap(); let stream = TcpStream::connect(proxy_addr).expect("connect"); - let peer_addr = stream.local_addr().unwrap(); + 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(); @@ -9867,7 +9863,7 @@ network_policies: let cache = BinaryIdentityCache::new(); - let mut result = resolve_process_identity(entrypoint_pid, peer_addr, proxy_addr, &cache); + let mut result = resolve_process_identity(entrypoint_pid, connection, &cache); for _ in 0..10 { match &result { Err(err) @@ -9876,8 +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_addr, proxy_addr, &cache); + result = resolve_process_identity(entrypoint_pid, connection, &cache); } Ok(_) => { // On arm64 under heavy CI load the /proc fd scan can transiently @@ -9885,8 +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_addr, proxy_addr, &cache); + result = resolve_process_identity(entrypoint_pid, connection, &cache); } _ => break, } From 8006cb1a102b8494f1e7418ebbefed487b44b97a Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Thu, 23 Jul 2026 10:27:43 -0700 Subject: [PATCH 3/3] refactor(supervisor-network): name procfs address family Signed-off-by: Piotr Mlocek --- .../src/procfs.rs | 70 +++++++++++++------ 1 file changed, 47 insertions(+), 23 deletions(-) diff --git a/crates/openshell-supervisor-network/src/procfs.rs b/crates/openshell-supervisor-network/src/procfs.rs index 0e6111967c..8bc2fbb110 100644 --- a/crates/openshell-supervisor-network/src/procfs.rs +++ b/crates/openshell-supervisor-network/src/procfs.rs @@ -66,6 +66,23 @@ impl std::fmt::Display for WorkloadProxyTcpConnection { } } +#[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 { @@ -325,13 +342,13 @@ pub fn collect_cmdline_paths(pid: u32, stop_pid: u32, exclude: &[PathBuf]) -> Ve #[cfg(target_os = "linux")] fn parse_proc_net_tcp(pid: u32, connection: WorkloadProxyTcpConnection) -> Result { // Check IPv4 first (most common), then IPv6. - for (suffix, ipv6) in [("tcp", false), ("tcp6", true)] { - 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; }; - if let Some(inode) = find_proc_net_tcp_inode(&content, ipv6, connection)? { + if let Some(inode) = find_proc_net_tcp_inode(&content, address_family, connection)? { return Ok(inode); } } @@ -344,7 +361,7 @@ fn parse_proc_net_tcp(pid: u32, connection: WorkloadProxyTcpConnection) -> Resul #[cfg(target_os = "linux")] fn find_proc_net_tcp_inode( content: &str, - ipv6: bool, + address_family: ProcTcpAddressFamily, connection: WorkloadProxyTcpConnection, ) -> Result> { for line in content.lines().skip(1) { @@ -353,10 +370,10 @@ fn find_proc_net_tcp_inode( continue; } - let Some(local_addr) = parse_proc_socket_addr(fields[1], ipv6) else { + let Some(local_addr) = parse_proc_socket_addr(fields[1], address_family) else { continue; }; - let Some(remote_addr) = parse_proc_socket_addr(fields[2], ipv6) else { + let Some(remote_addr) = parse_proc_socket_addr(fields[2], address_family) else { continue; }; @@ -376,24 +393,27 @@ fn find_proc_net_tcp_inode( } #[cfg(target_os = "linux")] -fn parse_proc_socket_addr(value: &str, ipv6: bool) -> Option { +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 = if ipv6 { - if address.len() != 32 { - return None; + let ip = match address_family { + ProcTcpAddressFamily::Ipv4 => { + let address = u32::from_str_radix(address, 16).ok()?; + IpAddr::V4(Ipv4Addr::from(address.to_le_bytes())) } - 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()); + 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)) } - IpAddr::V6(Ipv6Addr::from(bytes)) - } else { - let address = u32::from_str_radix(address, 16).ok()?; - IpAddr::V4(Ipv4Addr::from(address.to_le_bytes())) }; Some(SocketAddr::new(ip, port)) @@ -835,9 +855,12 @@ mod tests { #[cfg(target_os = "linux")] #[test] fn parse_proc_socket_addr_decodes_ipv4_and_mapped_ipv6() { - let ipv4 = parse_proc_socket_addr("0100007F:C350", false).unwrap(); - let mapped_ipv6 = - parse_proc_socket_addr("0000000000000000FFFF00000100007F:C350", true).unwrap(); + 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)); @@ -855,7 +878,8 @@ mod tests { "127.0.0.1:8080".parse().unwrap(), ); - let inode = find_proc_net_tcp_inode(content, false, connection).unwrap(); + let inode = + find_proc_net_tcp_inode(content, ProcTcpAddressFamily::Ipv4, connection).unwrap(); assert_eq!(inode, Some(22222)); }