diff --git a/CHANGELOG.md b/CHANGELOG.md index cacae0d7e..473376921 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -459,6 +459,7 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. refusal or eviction always falls back to verified reads. Cold reopen tests corrupt a neighbour frame after recovery and prove that no process-local evidence can suppress its checksum failure. Closes #1399. +- **Capsule inbound TCP bind (`bind_tcp`).** A capsule can bind an inbound TCP port and accept connections from its run loop via the WIT `bind-tcp(host, port) -> tcp-listener` fn (`accept` / `poll-accept` / `local-addr`), gated by the per-capsule `net_bind = ["host:port", "host:*"]` manifest capability (new fail-closed `check_net_tcp_bind` gate reusing the `net_connect` host:port matcher). A post-gate loopback airlock refuses non-loopback binds; accepted connections reuse the existing `NetStream::Tcp` read/write/close plumbing. The runtime primitive the srouter capsule needs (loopback ingress on 127.0.0.1:8788). Closes #1230. - **Windows local transport uses authenticated per-user named pipes.** A pipe name derived only from the caller's operating-system SID replaces filesystem endpoint naming on Windows. Local-only byte-mode instances use a diff --git a/crates/astrid-audit/src/entry.rs b/crates/astrid-audit/src/entry.rs index 5ef27fcca..2c4e8fbd2 100644 --- a/crates/astrid-audit/src/entry.rs +++ b/crates/astrid-audit/src/entry.rs @@ -467,6 +467,17 @@ pub enum AuditAction { #[serde(default, skip_serializing_if = "Option::is_none")] device_key_id: Option, }, + + /// Inbound TCP connection accepted by a capsule listener. + /// + /// Kept at the end of this fieldless enum so adding the action does not + /// change the implicit discriminants of any existing public variant. + NetAccept { + /// Host-observed local listener endpoint. + local_addr: String, + /// Host-observed remote peer endpoint. + peer_addr: String, + }, } impl AuditAction { @@ -540,6 +551,12 @@ impl AuditAction { Self::NetBind { addr } => { format!("Bound socket {addr}") }, + Self::NetAccept { + local_addr, + peer_addr, + } => { + format!("Accepted connection from {peer_addr} on {local_addr}") + }, Self::ProcessSpawn { command } => { format!("Spawned process {command}") }, diff --git a/crates/astrid-capsule/src/audit_sink.rs b/crates/astrid-capsule/src/audit_sink.rs index 478763ffd..75d4f8f56 100644 --- a/crates/astrid-capsule/src/audit_sink.rs +++ b/crates/astrid-capsule/src/audit_sink.rs @@ -61,6 +61,13 @@ pub enum HostAuditEvent<'a> { /// The command being executed. command: &'a str, }, + /// An inbound TCP connection accepted by a capsule listener. + NetAccept { + /// Host-observed local listener endpoint. + local_addr: &'a str, + /// Host-observed remote peer endpoint. + peer_addr: &'a str, + }, } /// The outcome of a sensitive host call, as seen at the host-fn seam. diff --git a/crates/astrid-capsule/src/engine/wasm/host/audit_sink_tests.rs b/crates/astrid-capsule/src/engine/wasm/host/audit_sink_tests.rs index 2e9c942d9..0a081f36e 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/audit_sink_tests.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/audit_sink_tests.rs @@ -22,6 +22,7 @@ enum CapturedEvent { FileDelete(String), NetConnect(String, u16), NetBind(String), + NetAccept(String, String), ProcessSpawn(String), } @@ -34,6 +35,10 @@ impl CapturedEvent { HostAuditEvent::NetConnect { host, port } => Self::NetConnect(host.to_owned(), port), HostAuditEvent::NetBind { addr } => Self::NetBind(addr.to_owned()), HostAuditEvent::ProcessSpawn { command } => Self::ProcessSpawn(command.to_owned()), + HostAuditEvent::NetAccept { + local_addr, + peer_addr, + } => Self::NetAccept(local_addr.to_owned(), peer_addr.to_owned()), } } } @@ -167,6 +172,28 @@ async fn audit_net_reports_connect() { ); } +#[tokio::test] +async fn audit_net_accept_carries_host_observed_endpoints() { + let (state, sink) = state_with_sink(tokio::runtime::Handle::current()); + let alice = PrincipalId::new("alice").unwrap(); + + super::net::audit_net_accept( + &state, + "127.0.0.1:8788", + "127.0.0.1:49152", + &Ok::<(), ()>(()), + ); + + assert_eq!( + sink.snapshot(), + vec![( + alice, + CapturedEvent::NetAccept("127.0.0.1:8788".into(), "127.0.0.1:49152".into()), + CapturedOutcome::Allowed, + )] + ); +} + #[tokio::test] async fn audit_net_reports_bind_denied() { // A denied socket bind (capsule lacks `net_bind`) currently leaves no diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs index 9f80c3b0f..c9aaca433 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -29,6 +29,7 @@ //! hop-limit, linger, reuseaddr socket options. use std::sync::Arc; +use std::sync::atomic::Ordering; use wasmtime::component::Resource; @@ -52,21 +53,108 @@ use stream::CONNECT_TIMEOUT; /// Maximum concurrent socket connections per capsule. Defense-in-depth /// cap on top of the per-principal profile quota. Tracked via -/// [`HostState::net_stream_count`], bumped on every successful +/// the capsule-wide stream counter, bumped on every successful /// `accept` / `connect-tcp` push and decremented in the resource /// drop path. pub(super) const MAX_ACTIVE_STREAMS: usize = 8; +/// Maximum simultaneously bound inbound TCP listeners per capsule instance, +/// matching the published `astrid:net` WIT contract. +pub(super) const MAX_ACTIVE_TCP_LISTENERS: usize = 4; + +impl HostState { + pub(in crate::engine::wasm) fn reserve_net_stream(&mut self) -> bool { + let reserved = self + .capsule_net_stream_count + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + (count < MAX_ACTIVE_STREAMS).then_some(count + 1) + }) + .is_ok(); + if reserved { + self.local_net_stream_count.fetch_add(1, Ordering::AcqRel); + self.net_stream_count += 1; + } + reserved + } + + pub(in crate::engine::wasm) fn release_net_stream(&mut self) { + let decremented = self.capsule_net_stream_count.fetch_update( + Ordering::AcqRel, + Ordering::Acquire, + |count| count.checked_sub(1), + ); + debug_assert!(decremented.is_ok(), "network stream quota underflow"); + let local = self.local_net_stream_count.fetch_update( + Ordering::AcqRel, + Ordering::Acquire, + |count| count.checked_sub(1), + ); + debug_assert!(local.is_ok(), "local network stream quota underflow"); + self.net_stream_count = self.net_stream_count.saturating_sub(1); + } + + pub(in crate::engine::wasm) fn claim_reserved_net_stream(&mut self) { + self.net_stream_count += 1; + } +} + /// Stamp marking a resource slot in the table as a `UnixListener` handle. /// The kernel pre-binds the listener; the resource handle is just a /// capability token that the capsule must hold to call `accept`. pub(super) struct UnixListenerSlot; -/// Stamp marking a resource slot as a `TcpListener` for future inbound -/// TCP server support. Pre-allocated so the type is in scope even though -/// `bind-tcp` is still a stub. -#[allow(dead_code)] -pub(super) struct TcpListenerSlot; +/// Resource slot holding a bound inbound TCP listener. The +/// `Resource` handed to the guest is a token over this slot; +/// `accept` / `poll-accept` / `local-addr` reach the `tokio` listener +/// through it, and `Drop` closes the socket. +pub(super) struct TcpListenerSlot { + pub(super) listener: Arc, + pub(super) pending: Arc, + pub(super) cancel_token: tokio_util::sync::CancellationToken, + pub(super) listener_count: Arc, +} + +pub(super) struct PendingTcpConnection { + pub(super) connection: tokio::sync::Mutex>, + pub(super) stream_count: Arc, + pub(super) local_stream_count: Arc, +} + +pub(super) struct PendingTcpAccepted { + pub(super) stream: tokio::net::TcpStream, + pub(super) local_addr: String, + pub(super) peer_addr: String, +} + +impl Drop for PendingTcpConnection { + fn drop(&mut self) { + if self.connection.get_mut().is_some() { + let decremented = + self.stream_count + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_sub(1) + }); + debug_assert!(decremented.is_ok(), "pending TCP quota underflow"); + let local = self.local_stream_count.fetch_update( + Ordering::AcqRel, + Ordering::Acquire, + |count| count.checked_sub(1), + ); + debug_assert!(local.is_ok(), "pending local TCP quota underflow"); + } + } +} + +impl Drop for TcpListenerSlot { + fn drop(&mut self) { + let decremented = + self.listener_count + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + count.checked_sub(1) + }); + debug_assert!(decremented.is_ok(), "TCP listener quota underflow"); + } +} /// Stamp marking a resource slot as a `UdpSocket`. Same reason as above. #[allow(dead_code)] @@ -86,6 +174,20 @@ pub(super) fn validate_host(host: &str) -> Result<(), ErrorCode> { Ok(()) } +/// Whether a TCP-bind host names a loopback interface. Capsule-hosted +/// servers are confined to loopback (see `bind_tcp`): `127.0.0.0/8`, `::1`, +/// or the literal `localhost`. A hostname other than `localhost` is refused +/// rather than resolved — binding must name a concrete local interface, and +/// resolving arbitrary names for a bind target is an SSRF-shaped footgun. +pub(super) fn is_loopback_bind_host(host: &str) -> bool { + if host.eq_ignore_ascii_case("localhost") { + return true; + } + host.parse::() + .map(|ip| ip.is_loopback()) + .unwrap_or(false) +} + /// Classify a tokio io::Error into the typed `net::ErrorCode`. pub(super) fn map_io_err(err: std::io::Error) -> ErrorCode { use std::io::ErrorKind; @@ -163,6 +265,36 @@ pub(crate) fn audit_net_connect( ); } +/// Record an inbound TCP accept with the host-observed local and peer +/// endpoints, so traffic entering a capsule retains durable provenance. +pub(crate) fn audit_net_accept( + state: &HostState, + local_addr: &str, + peer_addr: &str, + result: &Result, +) { + audit_net(state, "astrid:net/host.tcp-listener.accept", 0, result); + let Some(sink) = state.audit_sink.as_ref() else { + return; + }; + let error; + let outcome = match result { + Ok(_) => HostAuditOutcome::Allowed, + Err(err) => { + error = format!("{err:?}"); + HostAuditOutcome::Failed(&error) + }, + }; + sink.record( + &state.effective_principal(), + HostAuditEvent::NetAccept { + local_addr, + peer_addr, + }, + outcome, + ); +} + /// Report a denied net operation to the per-action audit sink. The connect /// gate rejects before any socket effect and early-returns, so this is the /// only audit report a denied connect makes (exactly-once recording). @@ -291,12 +423,115 @@ impl net::Host for HostState { Ok(Resource::new_own(res.rep())) } - fn bind_tcp(&mut self, _host: String, _port: u16) -> Result, ErrorCode> { - // Inbound TCP server hosting — needs a fresh tokio listener + - // capability gate (net_tcp_bind allowlist) + per-capsule accept - // loop. Lands in a follow-up commit; capsules importing - // `bind-tcp` today see CapabilityDenied so they fail closed. - Err(ErrorCode::CapabilityDenied) + fn bind_tcp(&mut self, host: String, port: u16) -> Result, ErrorCode> { + validate_host(&host)?; + let bind_addr = format!("tcp:{host}:{port}"); + + // Capability gate: host:port must match the capsule's `net_bind` + // allowlist (TCP entries share that field with unix binds). + if let Some(ref gate) = self.security { + let capsule_id = self.capsule_id.as_str().to_owned(); + let host_for_check = host.clone(); + let gate = gate.clone(); + let rt = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let check = util::bounded_block_on(&rt, &semaphore, async move { + gate.check_net_tcp_bind(&capsule_id, &host_for_check, port) + .await + }); + if let Err(reason) = check { + // Deny path records before the early return (exactly-once). + record_net_denied(self, HostAuditEvent::NetBind { addr: &bind_addr }, &reason); + return Err(ErrorCode::CapabilityDenied); + } + } + + // Security rail: capsule-hosted servers are loopback-only. Exposing a + // capsule listener beyond loopback is a deliberate future opt-in; this + // mirrors `connect-tcp`, which runs its `is_safe_ip` airlock AFTER the + // capability gate. A non-loopback bind is refused here, not silently + // downgraded. + if !is_loopback_bind_host(&host) { + let reason = "non-loopback TCP bind refused (capsule servers are loopback-only)"; + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(reason)); + return Err(ErrorCode::AirlockRejected); + } + + // Bind a fresh tokio listener on the daemon runtime. Quick op — the + // non-cancellable bounded_block_on is fine (accept, which blocks + // indefinitely, uses the cancellable variant instead). + let rt = self.runtime_handle.clone(); + let sem = self.blocking_semaphore.clone(); + // `localhost` is accepted for ergonomics, but never handed back to the + // resolver: local name service is mutable host configuration and may + // map it to a non-loopback address. Bind a concrete loopback literal. + let host_owned = if host.eq_ignore_ascii_case("localhost") { + "127.0.0.1".to_string() + } else { + host.clone() + }; + let bind_result: Result = + util::bounded_block_on(&rt, &sem, async move { + tokio::net::TcpListener::bind((host_owned.as_str(), port)).await + }); + let listener = match bind_result { + Ok(l) => l, + Err(e) => { + let mapped = map_io_err(e); + let reason = format!("{mapped:?}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(mapped); + }, + }; + let actual_addr = match listener.local_addr() { + Ok(addr) if addr.ip().is_loopback() => format!("tcp:{addr}"), + Ok(addr) => { + let reason = format!("resolved bind escaped loopback: {addr}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(ErrorCode::AirlockRejected); + }, + Err(e) => { + let mapped = map_io_err(e); + let reason = format!("{mapped:?}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(mapped); + }, + }; + + if self + .tcp_listener_count + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + (count < MAX_ACTIVE_TCP_LISTENERS).then_some(count + 1) + }) + .is_err() + { + let reason = "inbound TCP listener quota exceeded"; + audit_net_bind(self, &actual_addr, HostAuditOutcome::Failed(reason)); + return Err(ErrorCode::Quota); + } + + let slot = TcpListenerSlot { + listener: Arc::new(listener), + pending: Arc::new(PendingTcpConnection { + connection: tokio::sync::Mutex::new(None), + stream_count: Arc::clone(&self.capsule_net_stream_count), + local_stream_count: Arc::clone(&self.local_net_stream_count), + }), + cancel_token: self.effective_cancel_token(), + listener_count: Arc::clone(&self.tcp_listener_count), + }; + let res = match self.resource_table.push(slot) { + Ok(res) => res, + Err(e) => { + // The socket is already bound; the push consumes and drops the + // listener here, releasing it. Record the failure. + let reason = format!("resource table: {e}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(ErrorCode::Unknown(reason)); + }, + }; + audit_net_bind(self, &actual_addr, HostAuditOutcome::Allowed); + Ok(Resource::new_own(res.rep())) } fn connect_tcp(&mut self, host: String, port: u16) -> Result, ErrorCode> { @@ -324,7 +559,7 @@ impl net::Host for HostState { } } - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if self.capsule_net_stream_count.load(Ordering::Acquire) >= MAX_ACTIVE_STREAMS { let result: Result, ErrorCode> = Err(ErrorCode::Quota); audit_net_connect(self, &host, port, &result); return result; @@ -377,7 +612,7 @@ impl net::Host for HostState { }, }; - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if !self.reserve_net_stream() { drop(stream); let result: Result, ErrorCode> = Err(ErrorCode::Quota); audit_net_connect(self, &host, port, &result); @@ -392,6 +627,7 @@ impl net::Host for HostState { let res = match self.resource_table.push(net_stream) { Ok(res) => res, Err(e) => { + self.release_net_stream(); // The TCP connect ALREADY SUCCEEDED (the socket is open); the // push consumes and drops the stream here, aborting the // connection. Record the connect as having happened with a @@ -402,7 +638,6 @@ impl net::Host for HostState { return result; }, }; - self.net_stream_count += 1; let result: Result, ErrorCode> = Ok(Resource::new_own(res.rep())); audit_net_connect(self, &host, port, &result); result @@ -466,10 +701,86 @@ impl net::Host for HostState { #[cfg(test)] mod tests { use super::*; + use crate::engine::wasm::bindings::astrid::net::host::{Host as _, HostTcpListener}; + use crate::engine::wasm::test_fixtures::minimal_host_state; #[test] fn max_active_streams_pinned() { assert_eq!(MAX_ACTIVE_STREAMS, 8); + assert_eq!(MAX_ACTIVE_TCP_LISTENERS, 4); + } + + #[tokio::test(flavor = "multi_thread")] + async fn tcp_listener_quota_is_independent_and_released_on_drop() { + let mut state = minimal_host_state(tokio::runtime::Handle::current()); + let mut listeners = Vec::new(); + for _ in 0..MAX_ACTIVE_TCP_LISTENERS { + listeners.push(state.bind_tcp("127.0.0.1".into(), 0).unwrap()); + } + assert!(matches!( + state.bind_tcp("127.0.0.1".into(), 0), + Err(ErrorCode::Quota) + )); + + let released = listeners.pop().unwrap(); + HostTcpListener::drop(&mut state, released).unwrap(); + let replacement = state.bind_tcp("127.0.0.1".into(), 0).unwrap(); + HostTcpListener::drop(&mut state, replacement).unwrap(); + for listener in listeners { + HostTcpListener::drop(&mut state, listener).unwrap(); + } + assert_eq!(state.tcp_listener_count.load(Ordering::Acquire), 0); + } + + #[test] + fn public_store_count_excludes_other_pending_reservations() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let mut state = minimal_host_state(runtime.handle().clone()); + + assert!(state.reserve_net_stream()); + state + .capsule_net_stream_count + .fetch_add(1, Ordering::AcqRel); + state.local_net_stream_count.fetch_add(1, Ordering::AcqRel); + assert_eq!(state.net_stream_count, 1); + + state.release_net_stream(); + assert_eq!(state.net_stream_count, 0); + state.claim_reserved_net_stream(); + assert_eq!(state.net_stream_count, 1); + assert_eq!(state.capsule_net_stream_count.load(Ordering::Acquire), 1); + assert_eq!(state.local_net_stream_count.load(Ordering::Acquire), 1); + } + + #[test] + fn public_store_count_excludes_existing_pending_on_direct_reserve() { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let mut state = minimal_host_state(runtime.handle().clone()); + + state + .capsule_net_stream_count + .fetch_add(1, Ordering::AcqRel); + state.local_net_stream_count.fetch_add(1, Ordering::AcqRel); + assert_eq!(state.net_stream_count, 0); + + assert!(state.reserve_net_stream()); + assert_eq!(state.net_stream_count, 1); + state.claim_reserved_net_stream(); + assert_eq!(state.net_stream_count, 2); + assert_eq!(state.capsule_net_stream_count.load(Ordering::Acquire), 2); + assert_eq!(state.local_net_stream_count.load(Ordering::Acquire), 2); + } + + #[tokio::test(flavor = "multi_thread")] + async fn localhost_binds_a_concrete_loopback_address() { + let mut state = minimal_host_state(tokio::runtime::Handle::current()); + let listener = state.bind_tcp("localhost".into(), 0).unwrap(); + let local = state + .local_addr(Resource::new_borrow(listener.rep())) + .unwrap(); + let addr: std::net::SocketAddr = local.parse().unwrap(); + assert!(addr.ip().is_loopback()); + HostTcpListener::drop(&mut state, listener).unwrap(); } #[test] @@ -501,4 +812,24 @@ mod tests { let max = "a".repeat(255); assert!(validate_host(&max).is_ok()); } + + #[test] + fn loopback_bind_host_accepts_loopback() { + assert!(is_loopback_bind_host("127.0.0.1")); + assert!(is_loopback_bind_host("127.0.0.5")); + assert!(is_loopback_bind_host("::1")); + assert!(is_loopback_bind_host("localhost")); + assert!(is_loopback_bind_host("LOCALHOST")); + } + + #[test] + fn loopback_bind_host_rejects_non_loopback() { + assert!(!is_loopback_bind_host("0.0.0.0")); + assert!(!is_loopback_bind_host("192.168.1.10")); + assert!(!is_loopback_bind_host("8.8.8.8")); + assert!(!is_loopback_bind_host("::")); + // A hostname other than localhost is refused (not resolved). + assert!(!is_loopback_bind_host("example.com")); + assert!(!is_loopback_bind_host("")); + } } diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs index b8d8934ae..2c58008cc 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs @@ -1,43 +1,359 @@ //! `HostTcpListener` impl — inbound TCP server hosting. -//! -//! STUB SHELL — the bindings type exists in the WIT and the trait must -//! be implemented for the kernel to link. Every method returns -//! `CapabilityDenied` so capsules importing `bind-tcp` fail closed -//! rather than panic. Real impl lands alongside UDP in a follow-up. +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use async_trait::async_trait; use wasmtime::component::Resource; -use wasmtime_wasi::p2::DynPollable; +use wasmtime_wasi::p2::{DynPollable, Pollable, subscribe}; -use super::{HostState, TcpListenerSlot}; +use super::{ + HostState, MAX_ACTIVE_STREAMS, PendingTcpAccepted, PendingTcpConnection, TcpListenerSlot, + audit_net_accept, map_io_err, +}; +use crate::audit_sink::{HostAuditEvent, HostAuditOutcome, HostAuditSink}; use crate::engine::wasm::bindings::astrid::net::host::{ ErrorCode, HostTcpListener, TcpListener, TcpStream, }; +use crate::engine::wasm::host::util; +use crate::engine::wasm::host_state::{NetStream, TcpStreamSlot}; + +type TcpListenerParts = ( + Arc, + Arc, + tokio_util::sync::CancellationToken, +); + +/// Observe listener readability without consuming a connection. The actual +/// accept remains the single authority-bearing point for quota and audit. +struct TcpListenerReadiness { + listener: std::sync::Weak, + pending: std::sync::Weak, + cancel_token: tokio_util::sync::CancellationToken, + audit_sink: Option>, + principal: astrid_core::principal::PrincipalId, +} + +#[async_trait] +impl Pollable for TcpListenerReadiness { + async fn ready(&mut self) { + let (Some(listener), Some(pending)) = (self.listener.upgrade(), self.pending.upgrade()) + else { + return; + }; + // Holding this shared slot lock across accept serializes every watcher. + // Losing a WASI poll race simply drops the future and lock; no quota + // has been reserved and no connection has been consumed at that point. + let mut slot = pending.connection.lock().await; + if slot.is_some() { + return; + } + let accepted = tokio::select! { + result = listener.accept() => Some(result), + () = self.cancel_token.cancelled() => None, + }; + let Some(Ok((stream, peer_addr))) = accepted else { + return; + }; + let local_addr = stream.local_addr().map_or_else( + |error| format!("unknown ({error})"), + |addr| addr.to_string(), + ); + let peer_addr = peer_addr.to_string(); + if pending + .stream_count + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| { + (count < MAX_ACTIVE_STREAMS).then_some(count + 1) + }) + .is_err() + { + if let Some(sink) = &self.audit_sink { + sink.record( + &self.principal, + HostAuditEvent::NetAccept { + local_addr: &local_addr, + peer_addr: &peer_addr, + }, + HostAuditOutcome::Failed("network stream quota exceeded"), + ); + } + return; + } + pending.local_stream_count.fetch_add(1, Ordering::AcqRel); + if let Some(sink) = &self.audit_sink { + sink.record( + &self.principal, + HostAuditEvent::NetAccept { + local_addr: &local_addr, + peer_addr: &peer_addr, + }, + HostAuditOutcome::Allowed, + ); + } + *slot = Some(PendingTcpAccepted { + stream, + local_addr, + peer_addr, + }); + } +} + +impl HostState { + fn tcp_listener_slot(&self, rep: u32) -> Result { + let slot = self + .resource_table + .get::(&Resource::new_borrow(rep)) + .map_err(|_| ErrorCode::InvalidHandle)?; + Ok(( + Arc::clone(&slot.listener), + Arc::clone(&slot.pending), + slot.cancel_token.clone(), + )) + } + + fn register_accepted( + &mut self, + stream: tokio::net::TcpStream, + reserved: bool, + ) -> Result, ErrorCode> { + if !reserved && !self.reserve_net_stream() { + drop(stream); + return Err(ErrorCode::Quota); + } + if reserved { + self.claim_reserved_net_stream(); + } + let net_stream = NetStream::Tcp(TcpStreamSlot { + stream: Arc::new(tokio::sync::Mutex::new(stream)), + read_timeout: None, + write_timeout: None, + }); + let resource = match self.resource_table.push(net_stream) { + Ok(resource) => resource, + Err(error) => { + self.release_net_stream(); + return Err(ErrorCode::Unknown(format!("resource table: {error}"))); + }, + }; + Ok(Resource::new_own(resource.rep())) + } + + fn take_pending(&self, pending: Arc) -> Option { + let runtime = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let cancel = self.effective_cancel_token(); + util::bounded_block_on_cancellable(&runtime, &semaphore, &cancel, async move { + pending.connection.lock().await.take() + }) + .flatten() + } +} impl HostTcpListener for HostState { - fn accept(&mut self, _self_: Resource) -> Result, ErrorCode> { - Err(ErrorCode::CapabilityDenied) + fn accept(&mut self, self_: Resource) -> Result, ErrorCode> { + let (listener, pending, _) = self.tcp_listener_slot(self_.rep())?; + if self.capsule_net_stream_count.load(Ordering::Acquire) >= MAX_ACTIVE_STREAMS { + return Err(ErrorCode::Quota); + } + self.recv_yielded = true; + + let pending = self.take_pending(pending); + let (stream, local_addr, peer_addr, reserved) = if let Some(connection) = pending { + ( + connection.stream, + connection.local_addr, + connection.peer_addr, + true, + ) + } else { + let runtime = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let cancel = self.effective_cancel_token(); + let accepted = + util::bounded_block_on_cancellable(&runtime, &semaphore, &cancel, async move { + listener.accept().await + }); + let (stream, peer_addr) = match accepted { + Some(Ok(connection)) => connection, + Some(Err(error)) => return Err(map_io_err(error)), + None => return Err(ErrorCode::Closed), + }; + let local_addr = stream.local_addr().map_or_else( + |error| format!("unknown ({error})"), + |addr| addr.to_string(), + ); + (stream, local_addr, peer_addr.to_string(), false) + }; + let result = self.register_accepted(stream, reserved); + if !reserved { + audit_net_accept(self, &local_addr, &peer_addr, &result); + } + result } fn poll_accept( &mut self, - _self_: Resource, - _timeout_ms: u64, + self_: Resource, + timeout_ms: u64, ) -> Result>, ErrorCode> { - Err(ErrorCode::CapabilityDenied) + let (listener, pending, _) = self.tcp_listener_slot(self_.rep())?; + if self.capsule_net_stream_count.load(Ordering::Acquire) >= MAX_ACTIVE_STREAMS { + return Err(ErrorCode::Quota); + } + self.recv_yielded = true; + + if let Some(connection) = self.take_pending(pending) { + let result = self.register_accepted(connection.stream, true); + return result.map(Some); + } + let runtime = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let cancel = self.effective_cancel_token(); + let timeout = std::time::Duration::from_millis(timeout_ms); + let accepted = + util::bounded_block_on_cancellable(&runtime, &semaphore, &cancel, async move { + tokio::time::timeout(timeout, listener.accept()).await + }); + match accepted { + Some(Ok(Ok((stream, peer_addr)))) => { + let local_addr = stream.local_addr().map_or_else( + |error| format!("unknown ({error})"), + |addr| addr.to_string(), + ); + let peer_addr = peer_addr.to_string(); + let result = self.register_accepted(stream, false); + audit_net_accept(self, &local_addr, &peer_addr, &result); + result.map(Some) + }, + Some(Ok(Err(error))) => Err(map_io_err(error)), + Some(Err(_)) => Ok(None), + None => Err(ErrorCode::Closed), + } } - fn local_addr(&mut self, _self_: Resource) -> Result { - Err(ErrorCode::CapabilityDenied) + fn local_addr(&mut self, self_: Resource) -> Result { + let (listener, _, _) = self.tcp_listener_slot(self_.rep())?; + listener + .local_addr() + .map(|addr| addr.to_string()) + .map_err(map_io_err) } - fn subscribe_readiness(&mut self, _self_: Resource) -> Resource { - super::super::stubs::always_ready_pollable(&mut self.resource_table) + fn subscribe_readiness(&mut self, self_: Resource) -> Resource { + let (listener, pending, cancel_token) = self + .tcp_listener_slot(self_.rep()) + .expect("component model supplied a valid TCP listener resource"); + let watcher = self + .resource_table + .push(TcpListenerReadiness { + listener: Arc::downgrade(&listener), + pending: Arc::downgrade(&pending), + cancel_token, + audit_sink: self.audit_sink.clone(), + principal: self.effective_principal(), + }) + .expect("resource table accepted TCP readiness watcher"); + subscribe(&mut self.resource_table, watcher) + .expect("resource table accepted TCP readiness pollable") } fn drop(&mut self, rep: Resource) -> wasmtime::Result<()> { - let _ = self - .resource_table - .delete::(Resource::new_own(rep.rep())); + self.resource_table + .delete::(Resource::new_own(rep.rep()))?; Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn pending(count: &Arc) -> Arc { + Arc::new(PendingTcpConnection { + connection: tokio::sync::Mutex::new(None), + stream_count: Arc::clone(count), + local_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + }) + } + + #[tokio::test] + async fn readiness_accepts_once_and_reserves_exactly_once() { + let listener = Arc::new(tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap()); + let addr = listener.local_addr().unwrap(); + let count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let pending = pending(&count); + let mut readiness = TcpListenerReadiness { + listener: Arc::downgrade(&listener), + pending: Arc::downgrade(&pending), + cancel_token: tokio_util::sync::CancellationToken::new(), + audit_sink: None, + principal: astrid_core::principal::PrincipalId::default(), + }; + let client = tokio::spawn(tokio::net::TcpStream::connect(addr)); + + Pollable::ready(&mut readiness).await; + + assert_eq!(count.load(Ordering::Acquire), 1); + let accepted = pending.connection.lock().await.take().unwrap(); + assert_eq!(accepted.stream.local_addr().unwrap(), addr); + client.await.unwrap().unwrap(); + count.fetch_sub(1, Ordering::AcqRel); + pending.local_stream_count.fetch_sub(1, Ordering::AcqRel); + } + + #[tokio::test] + async fn readiness_wakes_when_cancelled() { + let listener = Arc::new(tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap()); + let count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let pending = pending(&count); + let cancel = tokio_util::sync::CancellationToken::new(); + let mut readiness = TcpListenerReadiness { + listener: Arc::downgrade(&listener), + pending: Arc::downgrade(&pending), + cancel_token: cancel.clone(), + audit_sink: None, + principal: astrid_core::principal::PrincipalId::default(), + }; + cancel.cancel(); + + tokio::time::timeout( + std::time::Duration::from_secs(1), + Pollable::ready(&mut readiness), + ) + .await + .expect("cancelled readiness must wake"); + assert_eq!(count.load(Ordering::Acquire), 0); + } + + #[tokio::test] + async fn readiness_pollable_does_not_keep_listener_or_quota_alive() { + let listener = Arc::new(tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap()); + let listener_count = Arc::new(std::sync::atomic::AtomicUsize::new(1)); + let stream_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let pending = pending(&stream_count); + let mut table = wasmtime::component::ResourceTable::new(); + let listener_resource = table + .push(TcpListenerSlot { + listener: Arc::clone(&listener), + pending: Arc::clone(&pending), + cancel_token: tokio_util::sync::CancellationToken::new(), + listener_count: Arc::clone(&listener_count), + }) + .unwrap(); + let watcher = table + .push(TcpListenerReadiness { + listener: Arc::downgrade(&listener), + pending: Arc::downgrade(&pending), + cancel_token: tokio_util::sync::CancellationToken::new(), + audit_sink: None, + principal: astrid_core::principal::PrincipalId::default(), + }) + .unwrap(); + let pollable = subscribe(&mut table, watcher).unwrap(); + drop(listener); + + table.delete(listener_resource).unwrap(); + assert_eq!(listener_count.load(Ordering::Acquire), 0); + table.delete(pollable).unwrap(); + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs index 0821789f1..5af7d92a2 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs @@ -518,7 +518,7 @@ impl HostTcpStream for HostState { .delete::(Resource::new_own(table_rep)) .is_ok() { - self.net_stream_count = self.net_stream_count.saturating_sub(1); + self.release_net_stream(); } // Drop any verified per-connection principal binding (issue #45/#852) // so the registry does not leak entries for closed connections. A diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs b/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs index ae9395cd1..5400b5365 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs @@ -5,6 +5,7 @@ //! `NetStream::Unix` resource handle. use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; use astrid_core::local_transport; @@ -21,7 +22,7 @@ use crate::engine::wasm::host::util; impl HostUnixListener for HostState { fn accept(&mut self, _self_: Resource) -> Result, ErrorCode> { - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if self.capsule_net_stream_count.load(Ordering::Acquire) >= MAX_ACTIVE_STREAMS { return Err(ErrorCode::Quota); } @@ -115,17 +116,19 @@ impl HostUnixListener for HostState { } }; - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if !self.reserve_net_stream() { drop(stream); return Err(ErrorCode::Quota); } let net_stream = NetStream::Unix(Arc::new(tokio::sync::Mutex::new(stream))); - let res = self - .resource_table - .push(net_stream) - .map_err(|e| ErrorCode::Unknown(format!("resource table: {e}")))?; - self.net_stream_count += 1; + let res = match self.resource_table.push(net_stream) { + Ok(resource) => resource, + Err(error) => { + self.release_net_stream(); + return Err(ErrorCode::Unknown(format!("resource table: {error}"))); + }, + }; let rep = res.rep(); // Record the verified principal AND its authenticating device key_id // (issue #45/#852) keyed by the stream resource rep, now that the rep @@ -164,7 +167,7 @@ impl HostUnixListener for HostState { let session_token = self.session_token.clone(); let blocking_semaphore = self.blocking_semaphore.clone(); - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if self.capsule_net_stream_count.load(Ordering::Acquire) >= MAX_ACTIVE_STREAMS { return Ok(None); } @@ -237,17 +240,19 @@ impl HostUnixListener for HostState { } } - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if !self.reserve_net_stream() { drop(stream); return Ok(None); } let net_stream = NetStream::Unix(Arc::new(tokio::sync::Mutex::new(stream))); - let res = self - .resource_table - .push(net_stream) - .map_err(|e| ErrorCode::Unknown(format!("resource table: {e}")))?; - self.net_stream_count += 1; + let res = match self.resource_table.push(net_stream) { + Ok(resource) => resource, + Err(error) => { + self.release_net_stream(); + return Err(ErrorCode::Unknown(format!("resource table: {error}"))); + }, + }; let rep = res.rep(); // Same per-connection principal + device-key binding as `accept` // (issue #45/#852). diff --git a/crates/astrid-capsule/src/engine/wasm/host_state.rs b/crates/astrid-capsule/src/engine/wasm/host_state.rs index b1330c204..b1cc92f7f 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state.rs @@ -615,13 +615,24 @@ pub struct HostState { /// here — off the wasmtime resource table — which is what lets them /// survive instance churn. pub persistent_processes: Arc, - /// Live count of `NetStream` entries currently in the resource table. - /// Maintained alongside `ResourceTable` insertions / drops so the - /// `MAX_ACTIVE_STREAMS` gate is O(1) instead of iterating every - /// resource (the table may hold hundreds of pollables / errors / - /// http handles unrelated to net). Single-threaded: wasmtime - /// stores are owned by exactly one OS thread. + /// Live count of accepted/connected streams owned by this Store. + /// + /// Retained as a public compatibility mirror for callers that inspect the + /// per-Store resource count. Capsule-wide quota enforcement uses the + /// internal atomic below because readiness pollables can reserve a stream + /// while the guest is outside a host call. pub net_stream_count: usize, + /// Live accepted/connected streams across every pooled Store for this + /// capsule, including readiness reservations not yet claimed by the guest. + pub(crate) capsule_net_stream_count: Arc, + /// Streams reserved by this pooled Store. Reset subtracts this exact + /// contribution from the shared capsule-wide counter. + pub(crate) local_net_stream_count: Arc, + /// Live count of bound inbound TCP listeners, shared by every pooled + /// store for this capsule. The WIT contract limits the capsule as a whole + /// to four listeners; a store-local counter would multiply that limit by + /// the dynamic pool size. + pub(crate) tcp_listener_count: Arc, /// Live count of `SubscriptionEntry` entries. Same rationale as /// `net_stream_count`. pub subscription_count: usize, diff --git a/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs b/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs index b5c591fb5..55f74e498 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs @@ -124,6 +124,9 @@ impl HostState { process_tracker, persistent_processes, net_stream_count: 0, + capsule_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + local_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + tcp_listener_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), subscription_count: 0, process_count_total: 0, process_count_by_principal: HashMap::new(), diff --git a/crates/astrid-capsule/src/engine/wasm/mod.rs b/crates/astrid-capsule/src/engine/wasm/mod.rs index eb5e01aa0..00fcfabb4 100644 --- a/crates/astrid-capsule/src/engine/wasm/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/mod.rs @@ -1645,14 +1645,20 @@ impl ExecutionEngine for WasmEngine { // the legacy [capabilities] arrays (helper falls back if empty). let ipc_publish_v = manifest.effective_ipc_publish_patterns(); let ipc_subscribe_v = manifest.effective_ipc_subscribe_patterns(); - // Only capsules declaring net_bind (the CLI proxy) get the socket - // listener / session token. - let cli_listener = if manifest.capabilities.net_bind.is_empty() { + // Only an explicit Unix bind declaration grants the pre-bound CLI + // listener and its session token. TCP host:port declarations use + // the same manifest field but must not cross-authorize Unix IPC. + let has_unix_bind = manifest + .capabilities + .net_bind + .iter() + .any(|entry| entry.starts_with("unix:")); + let cli_listener = if !has_unix_bind { None } else { ctx.cli_socket_listener.clone() }; - let session_tok = if manifest.capabilities.net_bind.is_empty() { + let session_tok = if !has_unix_bind { None } else { ctx.session_token.clone() @@ -1807,6 +1813,8 @@ impl ExecutionEngine for WasmEngine { let client_connections: Arc< dashmap::DashMap, > = Arc::new(dashmap::DashMap::new()); + let tcp_listener_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let capsule_net_stream_count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let make_state: Arc HostState + Send + Sync> = Arc::new(move || HostState { wasi_ctx: build_wasi_ctx(), resource_table: wasmtime::component::ResourceTable::new(), @@ -1898,6 +1906,9 @@ impl ExecutionEngine for WasmEngine { process_tracker: process_tracker.clone(), persistent_processes: persistent_registry.clone(), net_stream_count: 0, + capsule_net_stream_count: Arc::clone(&capsule_net_stream_count), + local_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + tcp_listener_count: Arc::clone(&tcp_listener_count), subscription_count: 0, process_count_total: 0, process_count_by_principal: std::collections::HashMap::new(), @@ -3039,6 +3050,9 @@ async fn build_lifecycle_host_state( tokio::runtime::Handle::current(), )), net_stream_count: 0, + capsule_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + local_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + tcp_listener_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), subscription_count: 0, process_count_total: 0, process_count_by_principal: std::collections::HashMap::new(), diff --git a/crates/astrid-capsule/src/engine/wasm/pool.rs b/crates/astrid-capsule/src/engine/wasm/pool.rs index 8c9f869c5..24a483f9f 100644 --- a/crates/astrid-capsule/src/engine/wasm/pool.rs +++ b/crates/astrid-capsule/src/engine/wasm/pool.rs @@ -31,6 +31,7 @@ //! via auto-subscribed IPC inside `run()`, not via `invoke_interceptor`. use std::collections::VecDeque; +use std::sync::atomic::Ordering; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -460,7 +461,16 @@ fn clear_on_return(state: &mut HostState, reset_resources: bool) { // them to the empty-table baseline so the per-(principal) gates start // from zero for the next lease. state.active_http_streams.clear(); + let local_streams = state.local_net_stream_count.swap(0, Ordering::AcqRel); state.net_stream_count = 0; + if local_streams != 0 { + let decremented = state.capsule_net_stream_count.fetch_update( + Ordering::AcqRel, + Ordering::Acquire, + |count| count.checked_sub(local_streams), + ); + debug_assert!(decremented.is_ok(), "shared network stream quota underflow"); + } state.subscription_count = 0; state.process_count_total = 0; state.process_count_by_principal.clear(); @@ -514,6 +524,8 @@ mod tests { .push(DropFlag(Arc::clone(&dropped))) .expect("push test resource"); state.net_stream_count = 1; + state.capsule_net_stream_count.store(1, Ordering::Release); + state.local_net_stream_count.store(1, Ordering::Release); state.subscription_count = 2; state.process_count_total = 1; state @@ -539,6 +551,7 @@ mod tests { ); // Mirror counters back to the empty-table baseline. assert_eq!(state.net_stream_count, 0); + assert_eq!(state.capsule_net_stream_count.load(Ordering::Acquire), 0); assert_eq!(state.subscription_count, 0); assert_eq!(state.process_count_total, 0); assert!(state.process_count_by_principal.is_empty()); @@ -550,6 +563,28 @@ mod tests { ); } + #[test] + fn returning_one_store_preserves_another_stores_stream_quota() { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("runtime"); + let shared = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut first = minimal_host_state(runtime.handle().clone()); + let mut second = minimal_host_state(runtime.handle().clone()); + first.capsule_net_stream_count = Arc::clone(&shared); + second.capsule_net_stream_count = Arc::clone(&shared); + + assert!(first.reserve_net_stream()); + assert!(second.reserve_net_stream()); + first.release_net_stream(); + clear_on_return(&mut first, true); + + assert_eq!(shared.load(Ordering::Acquire), 1); + assert_eq!(second.local_net_stream_count.load(Ordering::Acquire), 1); + clear_on_return(&mut second, true); + assert_eq!(shared.load(Ordering::Acquire), 0); + } + /// The `host_process` carve-out (`reset_resources = false`) deliberately /// keeps its `ManagedProcess` handles across invocations — the resource /// table and its counters must survive the return. diff --git a/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs b/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs index f5d44d9a3..d79626e83 100644 --- a/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs +++ b/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs @@ -146,6 +146,9 @@ pub(crate) fn minimal_host_state(rt: tokio::runtime::Handle) -> HostState { crate::engine::wasm::host::process::PersistentProcessRegistry::new(rt), ), net_stream_count: 0, + capsule_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + local_net_stream_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + tcp_listener_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)), subscription_count: 0, process_count_total: 0, process_count_by_principal: HashMap::new(), diff --git a/crates/astrid-capsule/src/security/manifest_gate.rs b/crates/astrid-capsule/src/security/manifest_gate.rs index 322b07f81..ab1767ad5 100644 --- a/crates/astrid-capsule/src/security/manifest_gate.rs +++ b/crates/astrid-capsule/src/security/manifest_gate.rs @@ -271,14 +271,14 @@ impl CapsuleSecurityGate for ManifestSecurityGate { } async fn check_net_bind(&self, capsule_id: &str) -> Result<(), String> { - // Require at least one non-empty net_bind entry. Empty strings in the - // manifest are treated as malformed and do not grant capability. + // The kernel's pre-bound CLI socket is Unix-only authority. A TCP + // host:port declaration must never make that listener available. let has_valid_entry = self .manifest .capabilities .net_bind .iter() - .any(|entry| !entry.is_empty()); + .any(|entry| entry.starts_with("unix:")); if has_valid_entry { Ok(()) } else { @@ -312,6 +312,33 @@ impl CapsuleSecurityGate for ManifestSecurityGate { } } + async fn check_net_tcp_bind( + &self, + capsule_id: &str, + host: &str, + port: u16, + ) -> Result<(), String> { + // Reuse the `net_bind` allowlist (its field documents "Unix/TCP socket + // bind addresses"). TCP entries are `host:port` / `host:*` patterns, + // matched with the SAME semantics as `net_connect`. Unix entries + // (`unix:*`) never match a TCP host:port, so the two socket families + // share the field without cross-authorizing. The host fn confines the + // bind to loopback after this gate returns Ok. + let allowed = self + .manifest + .capabilities + .net_bind + .iter() + .any(|entry| net_connect_pattern_matches(entry, host, port)); + if allowed { + Ok(()) + } else { + Err(format!( + "capsule '{capsule_id}' denied: TCP bind \"{host}:{port}\" not in net_bind allowlist" + )) + } + } + async fn check_identity( &self, capsule_id: &str, diff --git a/crates/astrid-capsule/src/security/manifest_gate_tests.rs b/crates/astrid-capsule/src/security/manifest_gate_tests.rs index 43193d79b..979348edf 100644 --- a/crates/astrid-capsule/src/security/manifest_gate_tests.rs +++ b/crates/astrid-capsule/src/security/manifest_gate_tests.rs @@ -394,11 +394,17 @@ async fn net_bind_gate_enforced() { let gate2 = ManifestSecurityGate::new(manifest2, workspace_root(), None); assert!(gate2.check_net_bind("test").await.is_ok()); - // Empty string in net_bind is treated as malformed -> denied + // Empty string in net_bind is treated as malformed -> denied. let mut manifest3 = make_manifest(vec![], vec![], vec![]); manifest3.capabilities.net_bind = vec!["".into()]; let gate3 = ManifestSecurityGate::new(manifest3, workspace_root(), None); assert!(gate3.check_net_bind("test").await.is_err()); + + // A TCP-only declaration must not authorize the shared Unix CLI listener. + let mut manifest4 = make_manifest(vec![], vec![], vec![]); + manifest4.capabilities.net_bind = vec!["127.0.0.1:8799".into()]; + let gate4 = ManifestSecurityGate::new(manifest4, workspace_root(), None); + assert!(gate4.check_net_bind("test").await.is_err()); } #[tokio::test] @@ -599,3 +605,67 @@ async fn check_net_connect_matches_allowlist_entry() { ); assert!(gate.check_net_connect("c", "evil.com", 443).await.is_err()); } + +#[tokio::test] +async fn check_net_tcp_bind_matches_net_bind_host_port() { + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_bind = vec!["127.0.0.1:8799".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + // Exact host:port allowed. + assert!( + gate.check_net_tcp_bind("c", "127.0.0.1", 8799) + .await + .is_ok() + ); + // Wrong port denied. + assert!( + gate.check_net_tcp_bind("c", "127.0.0.1", 9000) + .await + .is_err() + ); + // Wrong host denied. + assert!(gate.check_net_tcp_bind("c", "0.0.0.0", 8799).await.is_err()); +} + +#[tokio::test] +async fn check_net_tcp_bind_wildcard_port() { + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_bind = vec!["127.0.0.1:*".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!( + gate.check_net_tcp_bind("c", "127.0.0.1", 8799) + .await + .is_ok() + ); + assert!( + gate.check_net_tcp_bind("c", "127.0.0.1", 1234) + .await + .is_ok() + ); +} + +#[tokio::test] +async fn check_net_tcp_bind_unix_entry_does_not_authorize_tcp() { + // The CLI proxy declares `net_bind = ["unix:*"]`. That entry must NEVER + // authorize an inbound TCP bind — the two socket families share the field + // without cross-authorizing. + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_bind = vec!["unix:*".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!( + gate.check_net_tcp_bind("c", "127.0.0.1", 8799) + .await + .is_err() + ); +} + +#[tokio::test] +async fn check_net_tcp_bind_empty_net_bind_denies() { + let manifest = make_manifest(vec![], vec![], vec![]); + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!( + gate.check_net_tcp_bind("c", "127.0.0.1", 8799) + .await + .is_err() + ); +} diff --git a/crates/astrid-capsule/src/security/mod.rs b/crates/astrid-capsule/src/security/mod.rs index 55b19dd97..6705d8663 100644 --- a/crates/astrid-capsule/src/security/mod.rs +++ b/crates/astrid-capsule/src/security/mod.rs @@ -151,6 +151,32 @@ pub trait CapsuleSecurityGate: Send + Sync { )) } + /// Check whether the capsule is allowed to bind an INBOUND TCP listener + /// on `host:port` (capsule-hosted server). + /// + /// Default denies (fail-closed). The manifest gate overrides this to match + /// `host:port` against the capsule's `net_bind` allowlist (whose field + /// documents "Unix/TCP socket bind addresses"). A `unix:*` entry never + /// matches a TCP `host:port`, so the unix-listener path + /// ([`check_net_bind`](Self::check_net_bind)) and this TCP path share the + /// `net_bind` field without cross-authorizing. + /// + /// SECURITY: this gate only enforces the manifest allowlist. The host fn + /// (`bind-tcp`) additionally confines the bind to loopback — the same + /// gate-then-airlock split `connect-tcp` uses (`check_net_connect` then + /// `is_safe_ip`). Exposing a capsule-hosted server beyond loopback is a + /// deliberate future opt-in, not reachable through this method today. + async fn check_net_tcp_bind( + &self, + capsule_id: &str, + _host: &str, + _port: u16, + ) -> Result<(), String> { + Err(format!( + "capsule '{capsule_id}' denied: net_tcp_bind not permitted (default)" + )) + } + /// Check whether the capsule is allowed to register a uplink. /// /// Default implementation permits all registrations. Override to enforce diff --git a/crates/astrid-capsule/wit-staging/deps/astrid-net@1.0.0/net@1.0.0.wit b/crates/astrid-capsule/wit-staging/deps/astrid-net@1.0.0/net@1.0.0.wit index b4b3b4011..0811b59c4 100644 --- a/crates/astrid-capsule/wit-staging/deps/astrid-net@1.0.0/net@1.0.0.wit +++ b/crates/astrid-capsule/wit-staging/deps/astrid-net@1.0.0/net@1.0.0.wit @@ -337,12 +337,11 @@ interface host { /// Bind a TCP listener for inbound connections. /// - /// `"0.0.0.0"` / `"::"` exposes the listener to every network - /// interface (server posture) — restrict in `Capsule.toml - /// [capabilities] net_tcp_bind` to loopback-only patterns unless - /// the capsule genuinely needs to serve. Port `0` selects an - /// ephemeral port. Gated by a `net_tcp_bind` capability allowlist - /// distinct from `net_connect`. + /// Astrid confines TCP listeners to loopback hosts: `localhost`, + /// `127.0.0.0/8`, or `::1`. Wildcard and non-loopback interfaces are + /// rejected. Port `0` selects an ephemeral port. The requested + /// `host:port` must match the capsule's `[capabilities].net_bind` + /// allowlist; this authority is distinct from `net_connect`. bind-tcp: func(host: string, port: u16) -> result; /// Open an outbound TCP connection to `host:port`. Goes through diff --git a/crates/astrid-capsule/wit-staging/deps/astrid-process@1.0.0/process@1.0.0.wit b/crates/astrid-capsule/wit-staging/deps/astrid-process@1.0.0/process@1.0.0.wit index 4c93985c5..34aaf4de3 100644 --- a/crates/astrid-capsule/wit-staging/deps/astrid-process@1.0.0/process@1.0.0.wit +++ b/crates/astrid-capsule/wit-staging/deps/astrid-process@1.0.0/process@1.0.0.wit @@ -170,9 +170,9 @@ interface host { /// Environment variables. Replaces the host's default sandbox /// environment except for a small kernel-passthrough allowlist. env: list, - /// Working directory relative to the workspace, or a capability-gated - /// `home://` path. It must resolve inside an authorized sandbox root; - /// absolute paths and `..` escapes are rejected with `boundary-escape`. + /// Working directory relative to the workspace. Must resolve + /// inside the sandbox; absolute paths and `..` escapes are + /// rejected with `boundary-escape`. cwd: option, /// Per-child OS resource ceilings. Applies to EVERY tier. /// (NOT YET ENFORCED — see `resource-limits`.) diff --git a/crates/astrid-capsule/wit-staging/deps/astrid-process@1.1.0/process@1.1.0.wit b/crates/astrid-capsule/wit-staging/deps/astrid-process@1.1.0/process@1.1.0.wit index 8f2bed5c1..c192f956c 100644 --- a/crates/astrid-capsule/wit-staging/deps/astrid-process@1.1.0/process@1.1.0.wit +++ b/crates/astrid-capsule/wit-staging/deps/astrid-process@1.1.0/process@1.1.0.wit @@ -66,8 +66,10 @@ interface host { /// Per-principal CONCURRENT background-process cap exhausted (shared /// between ephemeral `spawn-background` and `spawn-persistent`). quota, - /// Stdin payload exceeded the per-call 1 MB cap or the - /// cumulative-per-process write quota. + /// Payload exceeded the applicable cap: the 4 MiB + /// `spawn-request.stdin` prelude (per spawn, cumulative), the + /// 1 MiB per-call `write-stdin` limit, or the cumulative + /// per-process write quota. too-large, /// Handle has been closed (process exited and reaped). closed, @@ -239,9 +241,9 @@ interface host { /// Environment variables. Replaces the host's default sandbox /// environment except for a small kernel-passthrough allowlist. env: list, - /// Working directory relative to the workspace, or a capability-gated - /// `home://` path. It must resolve inside an authorized sandbox root; - /// absolute paths and `..` escapes are rejected with `boundary-escape`. + /// Working directory relative to the workspace. Must resolve + /// inside the sandbox; absolute paths and `..` escapes are + /// rejected with `boundary-escape`. cwd: option, /// Per-child OS resource ceilings. Applies to EVERY tier. /// (NOT YET ENFORCED — see `resource-limits`.) diff --git a/crates/astrid-kernel/src/audit_sink.rs b/crates/astrid-kernel/src/audit_sink.rs index 2bb7fd950..446b847a4 100644 --- a/crates/astrid-kernel/src/audit_sink.rs +++ b/crates/astrid-kernel/src/audit_sink.rs @@ -118,6 +118,40 @@ impl KernelAuditSink { HostAuditEvent::ProcessSpawn { command } => AuditAction::ProcessSpawn { command: truncate_guest_str(command), }, + HostAuditEvent::NetAccept { + local_addr, + peer_addr, + } => AuditAction::NetAccept { + local_addr: truncate_guest_str(local_addr), + peer_addr: truncate_guest_str(peer_addr), + }, + } + } + + fn record_action( + &self, + principal: &PrincipalId, + action: AuditAction, + outcome: HostAuditOutcome<'_>, + ) { + let (proof, audit_outcome) = Self::to_proof_outcome(outcome); + // Native-only sync bridge onto the async audit log; see + // [`block_on_audit`]. Persistence failure degrades to "continue + + // alert", never a panic or a blocked host call. + let result = block_on_audit(self.audit_log.append_with_principal( + self.session_id.clone(), + principal.clone(), + action, + proof, + audit_outcome, + )); + if let Err(e) = result { + warn!( + security_event = true, + %principal, + error = %e, + "Failed to persist per-action audit entry — continuing" + ); } } @@ -200,25 +234,7 @@ impl HostAuditSink for KernelAuditSink { outcome: HostAuditOutcome<'_>, ) { let action = Self::to_action(event); - let (proof, audit_outcome) = Self::to_proof_outcome(outcome); - // Native-only sync bridge onto the async audit log; see - // [`block_on_audit`]. Persistence failure degrades to "continue + - // alert", never a panic or a blocked host call. - let result = block_on_audit(self.audit_log.append_with_principal( - self.session_id.clone(), - principal.clone(), - action, - proof, - audit_outcome, - )); - if let Err(e) = result { - warn!( - security_event = true, - %principal, - error = %e, - "Failed to persist per-action audit entry — continuing" - ); - } + self.record_action(principal, action, outcome); } } @@ -273,6 +289,14 @@ mod tests { }, HostAuditOutcome::Allowed, ); + sink.record( + &p, + HostAuditEvent::NetAccept { + local_addr: "127.0.0.1:8788", + peer_addr: "127.0.0.1:49152", + }, + HostAuditOutcome::Allowed, + ); sink.record( &p, HostAuditEvent::ProcessSpawn { command: "ls" }, @@ -283,7 +307,7 @@ mod tests { .get_principal_entries(&session, Some(&p)) .await .expect("read principal entries"); - assert_eq!(entries.len(), 6, "all six events must persist"); + assert_eq!(entries.len(), 7, "all seven events must persist"); // Every entry is stamped with the acting principal. for e in &entries { @@ -316,12 +340,20 @@ mod tests { &entries[4].action, AuditAction::NetBind { addr } if addr == "127.0.0.1:0" )); + // NetAccept → success with host-observed endpoints. + assert!(matches!( + &entries[5].action, + AuditAction::NetAccept { + local_addr, + peer_addr, + } if local_addr == "127.0.0.1:8788" && peer_addr == "127.0.0.1:49152" + )); // ProcessSpawn Denied → Failure + Denied proof. assert!(matches!( ( - &entries[5].action, - &entries[5].authorization, - &entries[5].outcome + &entries[6].action, + &entries[6].authorization, + &entries[6].outcome ), ( AuditAction::ProcessSpawn { command }, diff --git a/wit b/wit index 278dbca3e..148ecec5f 160000 --- a/wit +++ b/wit @@ -1 +1 @@ -Subproject commit 278dbca3e32f327d0f2358644fc86559779ba0fd +Subproject commit 148ecec5f2d149e0a0130b463900c938b7677b8d