diff --git a/CHANGELOG.md b/CHANGELOG.md index d4c78a51d..876a89239 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,17 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. ## [Unreleased] +### Added + +- **Outbound TCP for capsules — `net.connect-tcp` host fn + `net_connect` capability.** Capsules can now open persistent TCP connections via the new `astrid:capsule/net.net-connect-tcp(host, port) -> stream-handle` host fn, gated by a per-capsule `net_connect = ["host:port", "host:*"]` allowlist in `Capsule.toml`. The returned handle flows through the existing `net-read` / `net-write` / `net-close-stream` plumbing, and the kernel reuses the same `is_safe_ip` airlock that gates `http-request` to reject loopback / private / link-local / multicast IPs after DNS resolution. Connect timeout bounded to 10s; per-capsule active-stream cap (`MAX_ACTIVE_STREAMS = 8`) shared with inbound `net-accept`. Unblocks WebSocket clients, MQTT, Discord/Telegram gateways, postgres/redis, and the immediate motivator: a Unicity-network capsule wrapping Sphere SDK (Fulcrum + Nostr WebSocket transports). Tracking issue: #745. RFC: [rfcs#27](https://github.com/unicity-astrid/rfcs/pull/27). WIT contract: [wit#5](https://github.com/unicity-astrid/wit/pull/5). +- **`NetStream` enum (Unix + Tcp)** in `engine::wasm::host_state` — replaces the bare `Arc>` value type in `active_streams`. The `net_read` / `net_write` dispatchers match on the variant; the inner framing (`read_frame` / `write_frame` generic helpers) is shared via `tokio::io::AsyncRead + AsyncWrite` trait bounds. Single-variant capsules see no behavior change. +- **`CapsuleSecurityGate::check_net_connect(capsule_id, host, port)`** — new trait method, default-deny. `ManifestSecurityGate` implements it by matching the requested `host:port` against the manifest's `net_connect` allowlist (case-insensitive host, exact-or-`*` port). + +### Changed + +- **`crates/astrid-capsule/src/manifest.rs` split into a `manifest/` submodule.** The 1000-line single file became `manifest/mod.rs` + `manifest/capabilities.rs` + `manifest/topics.rs`. `CapabilitiesDef`, `PublishDef`, `SubscribeDef` (with their custom deserializers and TOML parsing tests) live in dedicated submodules; the top-level `manifest::*` public API is preserved via `pub use` re-exports — no consumer-side change required. +- **`wit/astrid-capsule.wit` resynced from canonical `unicity-astrid/wit`** to pick up the new `net-connect-tcp` fn and the `ipc-message.principal` field (canonical PR #4 from May; was missing from the in-tree copy). Internal `IpcMessage → WitIpcMessage` conversion now forwards `principal`. + ## [0.6.0] - 2026-05-19 ### Breaking diff --git a/crates/astrid-capsule/Cargo.toml b/crates/astrid-capsule/Cargo.toml index c516d332f..15c2f1921 100644 --- a/crates/astrid-capsule/Cargo.toml +++ b/crates/astrid-capsule/Cargo.toml @@ -25,6 +25,7 @@ reqwest = { workspace = true } semver = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +socket2 = "0.6" tempfile = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } diff --git a/crates/astrid-capsule/src/engine/mcp_tests.rs b/crates/astrid-capsule/src/engine/mcp_tests.rs index f3789758c..9647be1d3 100644 --- a/crates/astrid-capsule/src/engine/mcp_tests.rs +++ b/crates/astrid-capsule/src/engine/mcp_tests.rs @@ -37,6 +37,7 @@ mod tests { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + net_connect: vec![], kv: vec![], fs_read: vec![], fs_write: vec![], diff --git a/crates/astrid-capsule/src/engine/wasm/host/fs.rs b/crates/astrid-capsule/src/engine/wasm/host/fs.rs index a020389d1..f7d84531c 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/fs.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/fs.rs @@ -578,6 +578,7 @@ mod tests { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + net_connect: vec![], kv: vec![], fs_read: vec!["home://".into()], fs_write: vec!["home://".into()], diff --git a/crates/astrid-capsule/src/engine/wasm/host/http.rs b/crates/astrid-capsule/src/engine/wasm/host/http.rs index 038c6b281..19746efb7 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/http.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/http.rs @@ -65,7 +65,7 @@ static SSRF_BYPASS: std::sync::LazyLock = std::sync::LazyLock::new(|| { false }); -fn is_safe_ip(mut ip: std::net::IpAddr) -> bool { +pub(super) fn is_safe_ip(mut ip: std::net::IpAddr) -> bool { if *SSRF_BYPASS { return true; } diff --git a/crates/astrid-capsule/src/engine/wasm/host/ipc.rs b/crates/astrid-capsule/src/engine/wasm/host/ipc.rs index 8959b153a..fcbdd2936 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/ipc.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/ipc.rs @@ -101,6 +101,7 @@ fn to_wit_ipc_message(msg: &IpcMessage) -> WitIpcMessage { topic: msg.topic.clone(), payload, source_id: msg.source_id.to_string(), + principal: msg.principal.clone(), } } diff --git a/crates/astrid-capsule/src/engine/wasm/host/net.rs b/crates/astrid-capsule/src/engine/wasm/host/net.rs deleted file mode 100644 index 49d15ff8c..000000000 --- a/crates/astrid-capsule/src/engine/wasm/host/net.rs +++ /dev/null @@ -1,570 +0,0 @@ -use astrid_core::session_token::{ - HandshakeRequest, HandshakeResponse, PROTOCOL_VERSION, SessionToken, -}; - -use crate::engine::wasm::bindings::astrid::capsule::net; -use crate::engine::wasm::bindings::astrid::capsule::types::NetReadStatus; -use crate::engine::wasm::host::util; -use crate::engine::wasm::host_state::HostState; - -/// Maximum concurrent socket connections per capsule. -/// Prevents resource exhaustion from malicious or runaway clients. -const MAX_ACTIVE_STREAMS: usize = 8; - -/// Returns true for IO errors that represent a normal peer disconnect. -/// These should NOT trap the WASM guest — the run loop handles dead streams. -fn is_peer_disconnect(e: &std::io::Error) -> bool { - matches!( - e.kind(), - std::io::ErrorKind::BrokenPipe - | std::io::ErrorKind::ConnectionReset - | std::io::ErrorKind::ConnectionAborted - | std::io::ErrorKind::UnexpectedEof - ) -} - -// --------------------------------------------------------------------------- -// Handshake helpers -// --------------------------------------------------------------------------- - -/// Timeout for individual handshake read/write operations (server-side). -const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); - -/// Maximum allowed size of a handshake request payload (bytes). -const MAX_HANDSHAKE_SIZE: usize = 4096; - -/// Validate the client handshake: read the `HandshakeRequest`, verify the token -/// and protocol version, then send back a `HandshakeResponse`. -/// -/// Returns `Ok(())` on success or `Err(reason)` with a human-readable rejection -/// reason. -async fn validate_handshake( - stream: &mut tokio::net::UnixStream, - expected_token: &SessionToken, -) -> Result<(), String> { - use tokio::io::AsyncReadExt; - - // 1. Read the handshake request (length-prefixed JSON, same wire format). - let mut len_buf = [0u8; 4]; - tokio::time::timeout(HANDSHAKE_TIMEOUT, stream.read_exact(&mut len_buf)) - .await - .map_err(|_| "handshake timed out (5s)".to_string())? - .map_err(|e| format!("handshake read error: {e}"))?; - - let len = u32::from_be_bytes(len_buf) as usize; - if len > MAX_HANDSHAKE_SIZE { - return Err(format!("handshake too large: {len} bytes")); - } - - let mut payload = vec![0u8; len]; - tokio::time::timeout(HANDSHAKE_TIMEOUT, stream.read_exact(&mut payload)) - .await - .map_err(|_| "handshake payload timed out".to_string())? - .map_err(|e| format!("handshake payload read error: {e}"))?; - - let request: HandshakeRequest = - serde_json::from_slice(&payload).map_err(|e| format!("invalid handshake JSON: {e}"))?; - - // 2. Validate protocol version FIRST - this check reveals no information - // about token validity. Checking version before token prevents an oracle - // where a "protocol mismatch" response confirms the token was correct. - if request.protocol_version != PROTOCOL_VERSION { - let reason = format!( - "Protocol version mismatch (client={}, server={}). \ - Restart the daemon with `astrid daemon restart`.", - request.protocol_version, PROTOCOL_VERSION, - ); - if let Err(e) = - send_handshake_response_timed(stream, &HandshakeResponse::error(&reason)).await - { - tracing::warn!(error = %e, "Failed to send handshake error response for protocol mismatch"); - } - return Err(reason); - } - - // 3. Validate token (constant-time comparison). - // Send a uniform error response on both malformed-hex and wrong-token - // paths to prevent an oracle that distinguishes the two failure modes. - let client_token = match SessionToken::from_hex(&request.token) { - Ok(t) => t, - Err(_) => { - if let Err(e) = send_handshake_response_timed( - stream, - &HandshakeResponse::error("authentication failed"), - ) - .await - { - tracing::warn!(error = %e, "Failed to send handshake error response"); - } - return Err("invalid session token".to_string()); - }, - }; - - if !expected_token.ct_eq(&client_token) { - if let Err(e) = send_handshake_response_timed( - stream, - &HandshakeResponse::error("authentication failed"), - ) - .await - { - tracing::warn!(error = %e, "Failed to send handshake error response"); - } - return Err("invalid session token".to_string()); - } - - // 4. All checks passed - send success response. - send_handshake_response_timed(stream, &HandshakeResponse::ok()) - .await - .map_err(|e| format!("failed to send handshake response: {e}"))?; - - // Truncate client_version to prevent log injection from oversized values. - // Use chars().take() to avoid panicking on multi-byte UTF-8 boundaries. - let safe_version: String = request.client_version.chars().take(64).collect(); - tracing::info!( - client_version = %safe_version, - "Socket handshake succeeded" - ); - Ok(()) -} - -/// Send a length-prefixed JSON handshake response with a 5s write timeout. -/// -/// Wraps [`send_handshake_response`] with a timeout to prevent a stalled -/// client from holding the accept loop hostage during the response write. -async fn send_handshake_response_timed( - stream: &mut tokio::net::UnixStream, - response: &HandshakeResponse, -) -> Result<(), std::io::Error> { - tokio::time::timeout(HANDSHAKE_TIMEOUT, send_handshake_response(stream, response)) - .await - .map_err(|_| std::io::Error::other("handshake response write timed out (5s)"))? -} - -/// Send a length-prefixed JSON handshake response. -async fn send_handshake_response( - stream: &mut tokio::net::UnixStream, - response: &HandshakeResponse, -) -> Result<(), std::io::Error> { - use tokio::io::AsyncWriteExt; - - let bytes = serde_json::to_vec(response) - .map_err(|e| std::io::Error::other(format!("serialize handshake response: {e}")))?; - let len = u32::try_from(bytes.len()) - .map_err(|_| std::io::Error::other("handshake response too large"))?; - - stream.write_all(&len.to_be_bytes()).await?; - stream.write_all(&bytes).await?; - stream.flush().await?; - Ok(()) -} - -/// Verify that the connecting process runs as the same UID as the daemon. -/// Returns `Err(reason)` if the UID does not match or credentials cannot -/// be retrieved. -#[cfg(unix)] -fn verify_peer_credentials(stream: &tokio::net::UnixStream) -> Result<(), String> { - match stream.peer_cred() { - Ok(cred) => { - let peer_uid = cred.uid(); - let my_uid = nix::unistd::geteuid().as_raw(); - if peer_uid != my_uid { - Err(format!( - "peer UID {peer_uid} does not match daemon UID {my_uid}" - )) - } else { - Ok(()) - } - }, - Err(e) => Err(format!("failed to check peer credentials: {e}")), - } -} - -impl net::Host for HostState { - /// Gate `net_bind` capability once at bind time (session-scoped). - /// - /// The kernel pre-binds the socket and provides it via `HostState`. This - /// function enforces the security gate before the capsule can use the - /// listener - subsequent `accept()` calls do not re-check. - fn net_bind_unix(&mut self, _listener_handle: u64) -> Result { - // Security gate: only capsules with net_bind capability may bind sockets. - if let Some(ref gate) = self.security { - let capsule_id = self.capsule_id.as_str().to_owned(); - let gate = gate.clone(); - let handle = self.runtime_handle.clone(); - let semaphore = self.host_semaphore.clone(); - util::bounded_block_on(&handle, &semaphore, async move { - gate.check_net_bind(&capsule_id).await - }) - .map_err(|e| format!("security denied net_bind: {e}"))?; - } - - // Return a dummy handle, since the socket is pre-bound. - Ok(1) - } - - fn net_accept(&mut self, _listener_handle: u64) -> Result { - // Pre-accept cap check: fast reject without blocking on accept(). - let stream_count = self.active_streams.len(); - if stream_count >= MAX_ACTIVE_STREAMS { - tracing::warn!( - max = MAX_ACTIVE_STREAMS, - current = stream_count, - "accept: connection cap reached, rejecting" - ); - return Err(format!( - "connection cap reached ({stream_count}/{MAX_ACTIVE_STREAMS})" - )); - } - - let listener_arc = self - .cli_socket_listener - .clone() - .ok_or_else(|| "No CLI Socket Listener available in HostState".to_string())?; - let rt_handle = self.runtime_handle.clone(); - let cancel_token = self.cancel_token.clone(); - let session_token = self.session_token.clone(); - let host_semaphore = self.host_semaphore.clone(); - - // Accept + authenticate loop. Authentication failures (wrong UID, bad - // token) retry accept immediately so a malicious client cannot gate - // legitimate connections behind the WASM-side 100ms backoff. - let stream = loop { - let accept_result = util::bounded_block_on_cancellable( - &rt_handle, - &host_semaphore, - &cancel_token, - async { - let l = listener_arc.lock().await; - l.accept().await - }, - ); - let (stream, _addr) = match accept_result { - Some(result) => result.map_err(|e| format!("accept error: {e}"))?, - None => return Err("capsule unloading".to_string()), - }; - - // Peer credential verification - reject connections from different UIDs. - #[cfg(unix)] - if let Err(reason) = verify_peer_credentials(&stream) { - tracing::warn!( - security_event = true, - reason = %reason, - "Rejected socket connection: peer credential check failed" - ); - drop(stream); - continue; - } - - // Authenticate the connection via session token handshake. - let mut stream = stream; - if let Some(ref token) = session_token { - let handshake_result = util::bounded_block_on_cancellable( - &rt_handle, - &host_semaphore, - &cancel_token, - validate_handshake(&mut stream, token), - ); - match handshake_result { - None => return Err("capsule unloading".to_string()), - Some(Ok(())) => break stream, - Some(Err(reason)) => { - tracing::warn!( - security_event = true, - reason = %reason, - "Rejected socket connection: handshake failed" - ); - drop(stream); - continue; - }, - } - } else { - // No session token configured (test/legacy mode) - accept without auth. - break stream; - } - }; - - // Defense-in-depth: re-check cap before insertion. - let stream_count = self.active_streams.len(); - if stream_count >= MAX_ACTIVE_STREAMS { - tracing::warn!( - max = MAX_ACTIVE_STREAMS, - current = stream_count, - "accept: connection cap reached post-handshake, dropping authenticated stream" - ); - drop(stream); - return Err(format!( - "connection cap reached ({stream_count}/{MAX_ACTIVE_STREAMS})" - )); - } - - // Use a monotonic counter to avoid handle ID reuse after stream removal. - let handle_id = self.next_stream_id; - self.next_stream_id = self - .next_stream_id - .checked_add(1) - .ok_or_else(|| "stream handle ID space exhausted".to_string())?; - debug_assert!( - !self.active_streams.contains_key(&handle_id), - "stream handle ID collision" - ); - self.active_streams.insert( - handle_id, - std::sync::Arc::new(tokio::sync::Mutex::new(stream)), - ); - - // The `client.v1.connected` event is intentionally NOT published - // here. At accept time the host has no idea which principal the - // socket will eventually claim, so an unstamped publish lands - // under `default` and breaks `astrid who` attribution (#22). The - // cli (uplink) capsule republishes the event with the correct - // claimed principal once the first principal-stamped ingress - // message arrives on the stream — at that point the principal - // is known and the kernel's `active_connections` counter is - // attributed correctly. A stream that connects but never sends - // a principal-claim message therefore does not appear in the - // per-principal roster, which matches the operator-facing - // semantics of `who` (idle/zombie sockets shouldn't count). - - Ok(handle_id) - } - - fn net_poll_accept(&mut self, _listener_handle: u64) -> Result, String> { - let listener_arc = self - .cli_socket_listener - .clone() - .ok_or_else(|| "No CLI Socket Listener available in HostState".to_string())?; - let rt_handle = self.runtime_handle.clone(); - let cancel_token = self.cancel_token.clone(); - let session_token = self.session_token.clone(); - let host_semaphore = self.host_semaphore.clone(); - let stream_count = self.active_streams.len(); - - // Enforce connection cap at the host level. - if stream_count >= MAX_ACTIVE_STREAMS { - tracing::warn!( - max = MAX_ACTIVE_STREAMS, - current = stream_count, - "poll_accept: connection cap reached, rejecting" - ); - return Ok(None); - } - - // Non-blocking accept with a short timeout. The 10ms window is long - // enough to catch a pending connection without meaningfully stalling - // the WASM loop. - let accept_result = - util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { - let l = listener_arc.lock().await; - tokio::time::timeout(std::time::Duration::from_millis(10), l.accept()).await - }); - - let (stream, _addr) = match accept_result { - // Cancellation: return None (capsule unloading). - None => return Ok(None), - // Timeout: no pending connection. - Some(Err(_)) => return Ok(None), - // Accept error: propagate. - Some(Ok(Err(e))) => return Err(format!("accept error: {e}")), - // Success: connection pending. - Some(Ok(Ok(pair))) => pair, - }; - - // Peer credential verification (same as accept_impl). - #[cfg(unix)] - if let Err(reason) = verify_peer_credentials(&stream) { - tracing::warn!( - security_event = true, - reason = %reason, - "poll_accept: rejected connection (peer credential check failed)" - ); - drop(stream); - return Ok(None); - } - - // Session token handshake. - let mut stream = stream; - if let Some(ref token) = session_token { - let handshake_result = util::bounded_block_on_cancellable( - &rt_handle, - &host_semaphore, - &cancel_token, - validate_handshake(&mut stream, token), - ); - match handshake_result { - None => return Ok(None), - Some(Err(reason)) => { - tracing::warn!( - security_event = true, - reason = %reason, - "poll_accept: rejected connection (handshake failed)" - ); - drop(stream); - return Ok(None); - }, - Some(Ok(())) => {}, - } - } - - // Store the authenticated stream. Re-check cap under lock for defense - // in depth. - if self.active_streams.len() >= MAX_ACTIVE_STREAMS { - drop(stream); - return Ok(None); - } - - let handle_id = self.next_stream_id; - self.next_stream_id = self - .next_stream_id - .checked_add(1) - .ok_or_else(|| "stream handle ID space exhausted".to_string())?; - debug_assert!( - !self.active_streams.contains_key(&handle_id), - "stream handle ID collision" - ); - self.active_streams.insert( - handle_id, - std::sync::Arc::new(tokio::sync::Mutex::new(stream)), - ); - - // See the matching comment in `net_accept` — `client.v1.connected` - // is published by the cli capsule on the first principal-stamped - // ingress message, not here, so the per-principal counter is - // attributed correctly (#22). - - Ok(Some(handle_id)) - } - - fn net_read(&mut self, stream_handle: u64) -> Result { - let stream_arc = self - .active_streams - .get(&stream_handle) - .ok_or_else(|| "Stream handle not found".to_string())? - .clone(); - - let rt_handle = self.runtime_handle.clone(); - let cancel_token = self.cancel_token.clone(); - let host_semaphore = self.host_semaphore.clone(); - - // Cancel safety: read_exact is not cancel-safe, so cancellation mid-read - // may leave a partial frame on the socket. This is acceptable because the - // capsule is unloading - the socket will be closed by Drop on - // active_streams and the client will see a hard EOF / connection reset. - use tokio::io::AsyncReadExt; - - let status = - util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { - let mut stream = stream_arc.lock().await; - let mut len_buf = [0u8; 4]; - - match tokio::time::timeout( - std::time::Duration::from_millis(50), - stream.read_exact(&mut len_buf), - ) - .await - { - Err(_) => return Ok(NetReadStatus::Pending), - Ok(Err(e)) if is_peer_disconnect(&e) => { - return Ok(NetReadStatus::Closed); - }, - Ok(Err(e)) => return Err(format!("socket read error: {e}")), - Ok(Ok(_)) => {}, - } - - let len = u32::from_be_bytes(len_buf) as usize; - if len > 10 * 1024 * 1024 { - return Err("Payload too large (max 10MB)".to_string()); - } - - let mut payload = vec![0u8; len]; - let timeout_ms = 5000 + (len as u64 / 1024); - match tokio::time::timeout( - std::time::Duration::from_millis(timeout_ms), - stream.read_exact(&mut payload), - ) - .await - { - Err(_) => return Err("Payload read timed out".to_string()), - Ok(Err(e)) if is_peer_disconnect(&e) => { - return Ok(NetReadStatus::Closed); - }, - Ok(Err(e)) => return Err(format!("socket payload read error: {e}")), - Ok(Ok(_)) => {}, - } - - Ok(NetReadStatus::Data(payload)) - }); - - // Cancellation (capsule unloading) -> Pending so the guest loop exits cleanly. - match status { - Some(r) => r, - None => Ok(NetReadStatus::Pending), - } - } - - fn net_write(&mut self, stream_handle: u64, data: Vec) -> Result<(), String> { - let stream_arc = self - .active_streams - .get(&stream_handle) - .ok_or_else(|| "Stream handle not found".to_string())? - .clone(); - - let rt_handle = self.runtime_handle.clone(); - let host_semaphore = self.host_semaphore.clone(); - let cancel_token = self.cancel_token.clone(); - - use tokio::io::AsyncWriteExt; - // Cancel safety: write_all is not cancel-safe, so cancellation mid-write - // may leave a partial frame on the socket. This is acceptable because the - // capsule is unloading - the socket will be closed by Drop on - // active_streams and the client will see a hard EOF / connection reset. - let result = - util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { - let mut stream = stream_arc.lock().await; - // In the CLI architecture, we expect length-prefixed writes back to the client - let len = u32::try_from(data.len()).map_err(|_| { - std::io::Error::other("write payload too large for length prefix") - })?; - stream.write_all(&len.to_be_bytes()).await?; - stream.write_all(&data).await?; - stream.flush().await?; - Ok::<(), std::io::Error>(()) - }); - match result { - Some(Ok(())) => {}, - Some(Err(e)) => { - // Write failed — client likely disconnected. Log and continue; - // the dead stream will be cleaned up on the next read. - tracing::debug!(error = %e, "net write failed, client likely disconnected"); - }, - None => return Err("capsule unloading".to_string()), - } - - Ok(()) - } - - fn net_close_stream(&mut self, stream_handle: u64) -> Result<(), String> { - // Idempotent: silently ignore if the handle was already removed. - // - // The `client.v1.disconnect` event used to be published here, but - // an unstamped publish lands under `default` regardless of which - // principal owned the connection, which double-counts on the - // kernel side (`connection_closed(default)` decrements default's - // counter even when the stream was attributed to alice). The cli - // capsule now publishes the disconnect stamped with the stream's - // bound principal before calling `close()` — see issue #22. - let _ = self.active_streams.remove(&stream_handle); - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn max_active_streams_pinned() { - // Changing MAX_ACTIVE_STREAMS requires explicit security review - // (resource exhaustion surface). Update this test deliberately. - assert_eq!(MAX_ACTIVE_STREAMS, 8); - } -} diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/handshake.rs b/crates/astrid-capsule/src/engine/wasm/host/net/handshake.rs new file mode 100644 index 000000000..3610d6445 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/net/handshake.rs @@ -0,0 +1,159 @@ +//! Inbound socket handshake: protocol-version + session-token verification +//! and peer-UID credential check. Used by `net-accept` / `net-poll-accept` +//! before an authenticated stream is exposed to the WASM guest. + +use astrid_core::session_token::{ + HandshakeRequest, HandshakeResponse, PROTOCOL_VERSION, SessionToken, +}; + +/// Timeout for individual handshake read/write operations (server-side). +pub(super) const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +/// Maximum allowed size of a handshake request payload (bytes). +const MAX_HANDSHAKE_SIZE: usize = 4096; + +/// Validate the client handshake: read the `HandshakeRequest`, verify the token +/// and protocol version, then send back a `HandshakeResponse`. +/// +/// Returns `Ok(())` on success or `Err(reason)` with a human-readable rejection +/// reason. +pub(super) async fn validate_handshake( + stream: &mut tokio::net::UnixStream, + expected_token: &SessionToken, +) -> Result<(), String> { + use tokio::io::AsyncReadExt; + + // 1. Read the handshake request (length-prefixed JSON, same wire format). + let mut len_buf = [0u8; 4]; + tokio::time::timeout(HANDSHAKE_TIMEOUT, stream.read_exact(&mut len_buf)) + .await + .map_err(|_| "handshake timed out (5s)".to_string())? + .map_err(|e| format!("handshake read error: {e}"))?; + + let len = u32::from_be_bytes(len_buf) as usize; + if len > MAX_HANDSHAKE_SIZE { + return Err(format!("handshake too large: {len} bytes")); + } + + let mut payload = vec![0u8; len]; + tokio::time::timeout(HANDSHAKE_TIMEOUT, stream.read_exact(&mut payload)) + .await + .map_err(|_| "handshake payload timed out".to_string())? + .map_err(|e| format!("handshake payload read error: {e}"))?; + + let request: HandshakeRequest = + serde_json::from_slice(&payload).map_err(|e| format!("invalid handshake JSON: {e}"))?; + + // 2. Validate protocol version FIRST - this check reveals no information + // about token validity. Checking version before token prevents an oracle + // where a "protocol mismatch" response confirms the token was correct. + if request.protocol_version != PROTOCOL_VERSION { + let reason = format!( + "Protocol version mismatch (client={}, server={}). \ + Restart the daemon with `astrid daemon restart`.", + request.protocol_version, PROTOCOL_VERSION, + ); + if let Err(e) = + send_handshake_response_timed(stream, &HandshakeResponse::error(&reason)).await + { + tracing::warn!(error = %e, "Failed to send handshake error response for protocol mismatch"); + } + return Err(reason); + } + + // 3. Validate token (constant-time comparison). + // Send a uniform error response on both malformed-hex and wrong-token + // paths to prevent an oracle that distinguishes the two failure modes. + let client_token = match SessionToken::from_hex(&request.token) { + Ok(t) => t, + Err(_) => { + if let Err(e) = send_handshake_response_timed( + stream, + &HandshakeResponse::error("authentication failed"), + ) + .await + { + tracing::warn!(error = %e, "Failed to send handshake error response"); + } + return Err("invalid session token".to_string()); + }, + }; + + if !expected_token.ct_eq(&client_token) { + if let Err(e) = send_handshake_response_timed( + stream, + &HandshakeResponse::error("authentication failed"), + ) + .await + { + tracing::warn!(error = %e, "Failed to send handshake error response"); + } + return Err("invalid session token".to_string()); + } + + // 4. All checks passed - send success response. + send_handshake_response_timed(stream, &HandshakeResponse::ok()) + .await + .map_err(|e| format!("failed to send handshake response: {e}"))?; + + // Truncate client_version to prevent log injection from oversized values. + // Use chars().take() to avoid panicking on multi-byte UTF-8 boundaries. + let safe_version: String = request.client_version.chars().take(64).collect(); + tracing::info!( + client_version = %safe_version, + "Socket handshake succeeded" + ); + Ok(()) +} + +/// Send a length-prefixed JSON handshake response with a 5s write timeout. +/// +/// Wraps [`send_handshake_response`] with a timeout to prevent a stalled +/// client from holding the accept loop hostage during the response write. +async fn send_handshake_response_timed( + stream: &mut tokio::net::UnixStream, + response: &HandshakeResponse, +) -> Result<(), std::io::Error> { + tokio::time::timeout(HANDSHAKE_TIMEOUT, send_handshake_response(stream, response)) + .await + .map_err(|_| std::io::Error::other("handshake response write timed out (5s)"))? +} + +/// Send a length-prefixed JSON handshake response. +async fn send_handshake_response( + stream: &mut tokio::net::UnixStream, + response: &HandshakeResponse, +) -> Result<(), std::io::Error> { + use tokio::io::AsyncWriteExt; + + let bytes = serde_json::to_vec(response) + .map_err(|e| std::io::Error::other(format!("serialize handshake response: {e}")))?; + let len = u32::try_from(bytes.len()) + .map_err(|_| std::io::Error::other("handshake response too large"))?; + + stream.write_all(&len.to_be_bytes()).await?; + stream.write_all(&bytes).await?; + stream.flush().await?; + Ok(()) +} + +/// Verify that the connecting process runs as the same UID as the daemon. +/// Returns `Err(reason)` if the UID does not match or credentials cannot +/// be retrieved. +#[cfg(unix)] +pub(super) fn verify_peer_credentials(stream: &tokio::net::UnixStream) -> Result<(), String> { + match stream.peer_cred() { + Ok(cred) => { + let peer_uid = cred.uid(); + let my_uid = nix::unistd::geteuid().as_raw(); + if peer_uid != my_uid { + Err(format!( + "peer UID {peer_uid} does not match daemon UID {my_uid}" + )) + } else { + Ok(()) + } + }, + Err(e) => Err(format!("failed to check peer credentials: {e}")), + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs new file mode 100644 index 000000000..208eac48a --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -0,0 +1,788 @@ +use crate::engine::wasm::bindings::astrid::capsule::net; +use crate::engine::wasm::bindings::astrid::capsule::types::{NetReadStatus, ShutdownHow}; +use crate::engine::wasm::host::http::is_safe_ip; +use crate::engine::wasm::host::util; +use crate::engine::wasm::host_state::{HostState, NetStream, TcpStreamSlot}; + +mod handshake; +mod stream; + +use handshake::{validate_handshake, verify_peer_credentials}; +use stream::{ + CONNECT_TIMEOUT, MAX_BYTES_PER_CALL, read_bytes_inner, read_frame, with_tcp_slot, + write_bytes_inner, write_frame, +}; + +/// DNS hostname guards before reaching the resolver. +/// +/// `lookup_host` will reject empty or null-byte input, but failing +/// early at the host-fn boundary keeps malformed guest input out of +/// the resolver / audit log / tracing spans entirely. 255 chars is +/// the RFC 1035 max-name-length. +fn validate_host(host: &str) -> Result<(), String> { + if host.is_empty() { + return Err("host must be non-empty".to_string()); + } + if host.len() > 255 { + return Err(format!( + "host too long ({} bytes, max 255 per RFC 1035)", + host.len() + )); + } + if host.bytes().any(|b| b == 0) { + return Err("host contains null byte".to_string()); + } + Ok(()) +} + +/// Maximum concurrent socket connections per capsule. +/// Prevents resource exhaustion from malicious or runaway clients. +const MAX_ACTIVE_STREAMS: usize = 8; + +impl net::Host for HostState { + /// Gate `net_bind` capability once at bind time (session-scoped). + /// + /// The kernel pre-binds the socket and provides it via `HostState`. This + /// function enforces the security gate before the capsule can use the + /// listener - subsequent `accept()` calls do not re-check. + fn net_bind_unix(&mut self, _listener_handle: u64) -> Result { + // Security gate: only capsules with net_bind capability may bind sockets. + if let Some(ref gate) = self.security { + let capsule_id = self.capsule_id.as_str().to_owned(); + let gate = gate.clone(); + let handle = self.runtime_handle.clone(); + let semaphore = self.host_semaphore.clone(); + util::bounded_block_on(&handle, &semaphore, async move { + gate.check_net_bind(&capsule_id).await + }) + .map_err(|e| format!("security denied net_bind: {e}"))?; + } + + // Return a dummy handle, since the socket is pre-bound. + Ok(1) + } + + fn net_accept(&mut self, _listener_handle: u64) -> Result { + // Pre-accept cap check: fast reject without blocking on accept(). + let stream_count = self.active_streams.len(); + if stream_count >= MAX_ACTIVE_STREAMS { + tracing::warn!( + max = MAX_ACTIVE_STREAMS, + current = stream_count, + "accept: connection cap reached, rejecting" + ); + return Err(format!( + "connection cap reached ({stream_count}/{MAX_ACTIVE_STREAMS})" + )); + } + + let listener_arc = self + .cli_socket_listener + .clone() + .ok_or_else(|| "No CLI Socket Listener available in HostState".to_string())?; + let rt_handle = self.runtime_handle.clone(); + let cancel_token = self.cancel_token.clone(); + let session_token = self.session_token.clone(); + let host_semaphore = self.host_semaphore.clone(); + + // Accept + authenticate loop. Authentication failures (wrong UID, bad + // token) retry accept immediately so a malicious client cannot gate + // legitimate connections behind the WASM-side 100ms backoff. + let stream = loop { + let accept_result = util::bounded_block_on_cancellable( + &rt_handle, + &host_semaphore, + &cancel_token, + async { + let l = listener_arc.lock().await; + l.accept().await + }, + ); + let (stream, _addr) = match accept_result { + Some(result) => result.map_err(|e| format!("accept error: {e}"))?, + None => return Err("capsule unloading".to_string()), + }; + + // Peer credential verification - reject connections from different UIDs. + #[cfg(unix)] + if let Err(reason) = verify_peer_credentials(&stream) { + tracing::warn!( + security_event = true, + reason = %reason, + "Rejected socket connection: peer credential check failed" + ); + drop(stream); + continue; + } + + // Authenticate the connection via session token handshake. + let mut stream = stream; + if let Some(ref token) = session_token { + let handshake_result = util::bounded_block_on_cancellable( + &rt_handle, + &host_semaphore, + &cancel_token, + validate_handshake(&mut stream, token), + ); + match handshake_result { + None => return Err("capsule unloading".to_string()), + Some(Ok(())) => break stream, + Some(Err(reason)) => { + tracing::warn!( + security_event = true, + reason = %reason, + "Rejected socket connection: handshake failed" + ); + drop(stream); + continue; + }, + } + } else { + // No session token configured (test/legacy mode) - accept without auth. + break stream; + } + }; + + // Defense-in-depth: re-check cap before insertion. + let stream_count = self.active_streams.len(); + if stream_count >= MAX_ACTIVE_STREAMS { + tracing::warn!( + max = MAX_ACTIVE_STREAMS, + current = stream_count, + "accept: connection cap reached post-handshake, dropping authenticated stream" + ); + drop(stream); + return Err(format!( + "connection cap reached ({stream_count}/{MAX_ACTIVE_STREAMS})" + )); + } + + // Use a monotonic counter to avoid handle ID reuse after stream removal. + let handle_id = self.next_stream_id; + self.next_stream_id = self + .next_stream_id + .checked_add(1) + .ok_or_else(|| "stream handle ID space exhausted".to_string())?; + debug_assert!( + !self.active_streams.contains_key(&handle_id), + "stream handle ID collision" + ); + self.active_streams.insert( + handle_id, + NetStream::Unix(std::sync::Arc::new(tokio::sync::Mutex::new(stream))), + ); + + // The `client.v1.connected` event is intentionally NOT published + // here. At accept time the host has no idea which principal the + // socket will eventually claim, so an unstamped publish lands + // under `default` and breaks `astrid who` attribution (#22). The + // cli (uplink) capsule republishes the event with the correct + // claimed principal once the first principal-stamped ingress + // message arrives on the stream — at that point the principal + // is known and the kernel's `active_connections` counter is + // attributed correctly. A stream that connects but never sends + // a principal-claim message therefore does not appear in the + // per-principal roster, which matches the operator-facing + // semantics of `who` (idle/zombie sockets shouldn't count). + + Ok(handle_id) + } + + fn net_poll_accept(&mut self, _listener_handle: u64) -> Result, String> { + let listener_arc = self + .cli_socket_listener + .clone() + .ok_or_else(|| "No CLI Socket Listener available in HostState".to_string())?; + let rt_handle = self.runtime_handle.clone(); + let cancel_token = self.cancel_token.clone(); + let session_token = self.session_token.clone(); + let host_semaphore = self.host_semaphore.clone(); + let stream_count = self.active_streams.len(); + + // Enforce connection cap at the host level. + if stream_count >= MAX_ACTIVE_STREAMS { + tracing::warn!( + max = MAX_ACTIVE_STREAMS, + current = stream_count, + "poll_accept: connection cap reached, rejecting" + ); + return Ok(None); + } + + // Non-blocking accept with a short timeout. The 10ms window is long + // enough to catch a pending connection without meaningfully stalling + // the WASM loop. + let accept_result = + util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { + let l = listener_arc.lock().await; + tokio::time::timeout(std::time::Duration::from_millis(10), l.accept()).await + }); + + let (stream, _addr) = match accept_result { + // Cancellation: return None (capsule unloading). + None => return Ok(None), + // Timeout: no pending connection. + Some(Err(_)) => return Ok(None), + // Accept error: propagate. + Some(Ok(Err(e))) => return Err(format!("accept error: {e}")), + // Success: connection pending. + Some(Ok(Ok(pair))) => pair, + }; + + // Peer credential verification (same as accept_impl). + #[cfg(unix)] + if let Err(reason) = verify_peer_credentials(&stream) { + tracing::warn!( + security_event = true, + reason = %reason, + "poll_accept: rejected connection (peer credential check failed)" + ); + drop(stream); + return Ok(None); + } + + // Session token handshake. + let mut stream = stream; + if let Some(ref token) = session_token { + let handshake_result = util::bounded_block_on_cancellable( + &rt_handle, + &host_semaphore, + &cancel_token, + validate_handshake(&mut stream, token), + ); + match handshake_result { + None => return Ok(None), + Some(Err(reason)) => { + tracing::warn!( + security_event = true, + reason = %reason, + "poll_accept: rejected connection (handshake failed)" + ); + drop(stream); + return Ok(None); + }, + Some(Ok(())) => {}, + } + } + + // Store the authenticated stream. Re-check cap under lock for defense + // in depth. + if self.active_streams.len() >= MAX_ACTIVE_STREAMS { + drop(stream); + return Ok(None); + } + + let handle_id = self.next_stream_id; + self.next_stream_id = self + .next_stream_id + .checked_add(1) + .ok_or_else(|| "stream handle ID space exhausted".to_string())?; + debug_assert!( + !self.active_streams.contains_key(&handle_id), + "stream handle ID collision" + ); + self.active_streams.insert( + handle_id, + NetStream::Unix(std::sync::Arc::new(tokio::sync::Mutex::new(stream))), + ); + + // See the matching comment in `net_accept` — `client.v1.connected` + // is published by the cli capsule on the first principal-stamped + // ingress message, not here, so the per-principal counter is + // attributed correctly (#22). + + Ok(Some(handle_id)) + } + + fn net_read(&mut self, stream_handle: u64) -> Result { + let stream = self + .active_streams + .get(&stream_handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + + let rt_handle = self.runtime_handle.clone(); + let cancel_token = self.cancel_token.clone(); + let host_semaphore = self.host_semaphore.clone(); + + // Cancel safety: read_exact is not cancel-safe, so cancellation mid-read + // may leave a partial frame on the socket. This is acceptable because the + // capsule is unloading - the socket will be closed by Drop on + // active_streams and the client will see a hard EOF / connection reset. + let status = + util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { + match stream { + NetStream::Unix(arc) => { + let mut s = arc.lock().await; + read_frame(&mut *s).await + }, + NetStream::Tcp(slot) => { + let mut s = slot.stream.lock().await; + read_frame(&mut *s).await + }, + } + }); + + // Cancellation (capsule unloading) -> Pending so the guest loop exits cleanly. + match status { + Some(r) => r, + None => Ok(NetReadStatus::Pending), + } + } + + fn net_write(&mut self, stream_handle: u64, data: Vec) -> Result<(), String> { + let stream = self + .active_streams + .get(&stream_handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + + let rt_handle = self.runtime_handle.clone(); + let host_semaphore = self.host_semaphore.clone(); + let cancel_token = self.cancel_token.clone(); + + // Cancel safety: write_all is not cancel-safe, so cancellation mid-write + // may leave a partial frame on the socket. This is acceptable because the + // capsule is unloading - the socket will be closed by Drop on + // active_streams and the client will see a hard EOF / connection reset. + let result = + util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { + match stream { + NetStream::Unix(arc) => { + let mut s = arc.lock().await; + write_frame(&mut *s, &data).await + }, + NetStream::Tcp(slot) => { + let mut s = slot.stream.lock().await; + write_frame(&mut *s, &data).await + }, + } + }); + match result { + Some(Ok(())) => {}, + Some(Err(e)) => { + // Write failed — client likely disconnected. Log and continue; + // the dead stream will be cleaned up on the next read. + tracing::debug!(error = %e, "net write failed, client likely disconnected"); + }, + None => return Err("capsule unloading".to_string()), + } + + Ok(()) + } + + fn net_connect_tcp(&mut self, host: String, port: u16) -> Result { + // Reject malformed host strings at the boundary — keeps unsafe + // input out of the resolver, the audit log, and tracing spans. + validate_host(&host)?; + // 1. Capability check — literal host:port against the manifest's + // net_connect allowlist. DNS / SSRF run after this gate. + 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.host_semaphore.clone(); + util::bounded_block_on(&rt, &semaphore, async move { + gate.check_net_connect(&capsule_id, &host_for_check, port) + .await + }) + .map_err(|e| format!("security denied net_connect: {e}"))?; + } + + // 2. Pre-insert active-stream cap check. + let stream_count = self.active_streams.len(); + if stream_count >= MAX_ACTIVE_STREAMS { + tracing::warn!( + max = MAX_ACTIVE_STREAMS, + current = stream_count, + "net_connect_tcp: connection cap reached, rejecting" + ); + return Err(format!( + "connection cap reached ({stream_count}/{MAX_ACTIVE_STREAMS})" + )); + } + + let rt_handle = self.runtime_handle.clone(); + let host_semaphore = self.host_semaphore.clone(); + let cancel_token = self.cancel_token.clone(); + + // 3. DNS resolve + SSRF airlock + TCP connect, all under a bounded + // timeout so a stalled handshake doesn't pin the WASM guest. + let connect_result = + util::bounded_block_on_cancellable(&rt_handle, &host_semaphore, &cancel_token, async { + tokio::time::timeout(CONNECT_TIMEOUT, async { + let addrs: Vec = + tokio::net::lookup_host((host.as_str(), port)) + .await + .map_err(|e| format!("dns: {e}"))? + .collect(); + if addrs.is_empty() { + return Err("dns: no addresses returned".to_string()); + } + for addr in &addrs { + if !is_safe_ip(addr.ip()) { + return Err(format!( + "net.connect-tcp denied: resolved IP {} is in a \ + private/loopback/link-local range (SSRF protection)", + addr.ip() + )); + } + } + tokio::net::TcpStream::connect(&addrs[..]) + .await + .map_err(|e| format!("connect: {e}")) + }) + .await + .map_err(|_| format!("connect timeout after {}s", CONNECT_TIMEOUT.as_secs())) + .and_then(|inner| inner) + }); + + let stream = match connect_result { + Some(Ok(s)) => s, + Some(Err(e)) => return Err(e), + None => return Err("capsule unloading".to_string()), + }; + + // 4. Defense in depth: re-check the cap before insertion. + if self.active_streams.len() >= MAX_ACTIVE_STREAMS { + drop(stream); + return Err(format!( + "connection cap reached ({}/{MAX_ACTIVE_STREAMS})", + self.active_streams.len() + )); + } + + let handle_id = self.next_stream_id; + self.next_stream_id = self + .next_stream_id + .checked_add(1) + .ok_or_else(|| "stream handle ID space exhausted".to_string())?; + debug_assert!( + !self.active_streams.contains_key(&handle_id), + "stream handle ID collision" + ); + self.active_streams.insert( + handle_id, + NetStream::Tcp(TcpStreamSlot { + stream: std::sync::Arc::new(tokio::sync::Mutex::new(stream)), + read_timeout: None, + write_timeout: None, + }), + ); + Ok(handle_id) + } + + fn net_close_stream(&mut self, stream_handle: u64) -> Result<(), String> { + // Idempotent: silently ignore if the handle was already removed. + // + // The `client.v1.disconnect` event used to be published here, but + // an unstamped publish lands under `default` regardless of which + // principal owned the connection, which double-counts on the + // kernel side (`connection_closed(default)` decrements default's + // counter even when the stream was attributed to alice). The cli + // capsule now publishes the disconnect stamped with the stream's + // bound principal before calling `close()` — see issue #22. + let _ = self.active_streams.remove(&stream_handle); + Ok(()) + } + + fn net_read_bytes(&mut self, stream_handle: u64, max_bytes: u32) -> Result, String> { + let stream = self + .active_streams + .get(&stream_handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + let rt = self.runtime_handle.clone(); + let sem = self.host_semaphore.clone(); + let tok = self.cancel_token.clone(); + let max = max_bytes as usize; + let result = util::bounded_block_on_cancellable(&rt, &sem, &tok, async { + match stream { + NetStream::Unix(arc) => { + let mut s = arc.lock().await; + read_bytes_inner(&mut *s, max, None).await + }, + NetStream::Tcp(slot) => { + let timeout = slot.read_timeout; + let mut s = slot.stream.lock().await; + read_bytes_inner(&mut *s, max, timeout).await + }, + } + }); + result.unwrap_or_else(|| Ok(Vec::new())) + } + + fn net_write_bytes(&mut self, stream_handle: u64, data: Vec) -> Result { + let stream = self + .active_streams + .get(&stream_handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + let rt = self.runtime_handle.clone(); + let sem = self.host_semaphore.clone(); + let tok = self.cancel_token.clone(); + let result = util::bounded_block_on_cancellable(&rt, &sem, &tok, async { + match stream { + NetStream::Unix(arc) => { + let mut s = arc.lock().await; + write_bytes_inner(&mut *s, &data, None).await + }, + NetStream::Tcp(slot) => { + let timeout = slot.write_timeout; + let mut s = slot.stream.lock().await; + write_bytes_inner(&mut *s, &data, timeout).await + }, + } + }); + match result { + Some(r) => r, + None => Err("capsule unloading".to_string()), + } + } + + fn net_peek(&mut self, stream_handle: u64, max_bytes: u32) -> Result, String> { + // tokio's TcpStream has `peek` natively; UnixStream does not. The + // Unix case is rare for outbound work — return an error rather + // than emulate via a buffered wrapper that other host fns would + // also have to learn about. + let stream = self + .active_streams + .get(&stream_handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + let rt = self.runtime_handle.clone(); + let sem = self.host_semaphore.clone(); + let tok = self.cancel_token.clone(); + // Same OOM cap as read_bytes — guest can pass u32::MAX which + // would be a 4 GB host-side allocation. + let max = (max_bytes as usize).min(MAX_BYTES_PER_CALL); + match stream { + NetStream::Tcp(slot) => { + let timeout = slot.read_timeout; + let result = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move { + let s = slot.stream.lock().await; + let mut buf = vec![0u8; max]; + let fut = s.peek(&mut buf); + let n = match timeout { + Some(d) => match tokio::time::timeout(d, fut).await { + Ok(Ok(n)) => n, + Ok(Err(e)) => return Err(format!("peek error: {e}")), + Err(_) => return Err("peek would block".to_string()), + }, + None => fut.await.map_err(|e| format!("peek error: {e}"))?, + }; + buf.truncate(n); + Ok(buf) + }); + result.unwrap_or_else(|| Ok(Vec::new())) + }, + NetStream::Unix(_) => Err("peek not supported on Unix streams".to_string()), + } + } + + fn net_shutdown(&mut self, stream_handle: u64, how: ShutdownHow) -> Result<(), String> { + let stream = self + .active_streams + .get(&stream_handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + let rt = self.runtime_handle.clone(); + let sem = self.host_semaphore.clone(); + let tok = self.cancel_token.clone(); + let std_how = match how { + ShutdownHow::Read => std::net::Shutdown::Read, + ShutdownHow::Write => std::net::Shutdown::Write, + ShutdownHow::Both => std::net::Shutdown::Both, + }; + let result = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move { + match stream { + NetStream::Tcp(slot) => { + // Use socket2's borrowed `SockRef` for full + // `Shutdown::{Read,Write,Both}` support — tokio's + // `TcpStream` only exposes the write half via the + // `AsyncWrite` trait, but the OS `shutdown(2)` + // syscall (reached through socket2) handles every + // direction cleanly. Borrowed; no FD ownership + // transfer; safe to call repeatedly. + let s = slot.stream.lock().await; + let sock_ref = socket2::SockRef::from(&*s); + sock_ref + .shutdown(std_how) + .map_err(|e| format!("shutdown({how:?}): {e}")) + }, + NetStream::Unix(_) => Err("shutdown not supported on Unix streams".to_string()), + } + }); + match result { + Some(r) => r, + None => Err("capsule unloading".to_string()), + } + } + + fn net_peer_addr(&mut self, stream_handle: u64) -> Result { + with_tcp_slot(self, stream_handle, |slot| { + slot.peer_addr() + .map(|a| a.to_string()) + .map_err(|e| format!("peer_addr: {e}")) + }) + } + + fn net_local_addr(&mut self, stream_handle: u64) -> Result { + with_tcp_slot(self, stream_handle, |slot| { + slot.local_addr() + .map(|a| a.to_string()) + .map_err(|e| format!("local_addr: {e}")) + }) + } + + fn net_set_nodelay(&mut self, stream_handle: u64, nodelay: bool) -> Result<(), String> { + with_tcp_slot(self, stream_handle, |slot| { + slot.set_nodelay(nodelay) + .map_err(|e| format!("set_nodelay: {e}")) + }) + } + + fn net_nodelay(&mut self, stream_handle: u64) -> Result { + with_tcp_slot(self, stream_handle, |slot| { + slot.nodelay().map_err(|e| format!("nodelay: {e}")) + }) + } + + fn net_set_read_timeout( + &mut self, + stream_handle: u64, + timeout_ms: Option, + ) -> Result<(), String> { + match self.active_streams.get_mut(&stream_handle) { + Some(NetStream::Tcp(slot)) => { + slot.read_timeout = timeout_ms.map(std::time::Duration::from_millis); + Ok(()) + }, + Some(NetStream::Unix(_)) => { + Err("set_read_timeout not supported on Unix streams".to_string()) + }, + None => Err("Stream handle not found".to_string()), + } + } + + fn net_read_timeout(&mut self, stream_handle: u64) -> Result, String> { + match self.active_streams.get(&stream_handle) { + Some(NetStream::Tcp(slot)) => Ok(slot + .read_timeout + .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))), + Some(NetStream::Unix(_)) => { + Err("read_timeout not supported on Unix streams".to_string()) + }, + None => Err("Stream handle not found".to_string()), + } + } + + fn net_set_write_timeout( + &mut self, + stream_handle: u64, + timeout_ms: Option, + ) -> Result<(), String> { + match self.active_streams.get_mut(&stream_handle) { + Some(NetStream::Tcp(slot)) => { + slot.write_timeout = timeout_ms.map(std::time::Duration::from_millis); + Ok(()) + }, + Some(NetStream::Unix(_)) => { + Err("set_write_timeout not supported on Unix streams".to_string()) + }, + None => Err("Stream handle not found".to_string()), + } + } + + fn net_write_timeout(&mut self, stream_handle: u64) -> Result, String> { + match self.active_streams.get(&stream_handle) { + Some(NetStream::Tcp(slot)) => Ok(slot + .write_timeout + .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))), + Some(NetStream::Unix(_)) => { + Err("write_timeout not supported on Unix streams".to_string()) + }, + None => Err("Stream handle not found".to_string()), + } + } + + fn net_set_ttl(&mut self, stream_handle: u64, ttl: u32) -> Result<(), String> { + with_tcp_slot(self, stream_handle, |slot| { + slot.set_ttl(ttl).map_err(|e| format!("set_ttl: {e}")) + }) + } + + fn net_ttl(&mut self, stream_handle: u64) -> Result { + with_tcp_slot(self, stream_handle, |slot| { + slot.ttl().map_err(|e| format!("ttl: {e}")) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn max_active_streams_pinned() { + // Changing MAX_ACTIVE_STREAMS requires explicit security review + // (resource exhaustion surface). Update this test deliberately. + assert_eq!(MAX_ACTIVE_STREAMS, 8); + } + + #[tokio::test] + async fn socket2_shutdown_supports_all_three_directions() { + // Smoke test that socket2::SockRef::from(&tokio::net::TcpStream) + // accepts every Shutdown direction. If a future tokio/socket2 bump + // breaks this glue, the test fails loudly here instead of + // surfacing as a runtime error on the first capsule that calls + // shutdown(Read|Both). + use std::net::Shutdown; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let (client, _server) = tokio::join!( + async { tokio::net::TcpStream::connect(addr).await.unwrap() }, + async { listener.accept().await.unwrap() }, + ); + let sock = socket2::SockRef::from(&client); + sock.shutdown(Shutdown::Write).unwrap(); + // Read shutdown after write may already be effectively done; calling + // it again must not panic. Some platforms return ENOTCONN here, + // which is fine — the point is the FFI call resolves without + // dropping into an "only write supported" branch. + let _ = sock.shutdown(Shutdown::Read); + let _ = sock.shutdown(Shutdown::Both); + } + + #[test] + fn validate_host_accepts_normal_names() { + assert!(validate_host("example.com").is_ok()); + assert!(validate_host("fulcrum.unicity.network").is_ok()); + assert!(validate_host("127.0.0.1").is_ok()); + assert!(validate_host("::1").is_ok()); + } + + #[test] + fn validate_host_rejects_empty() { + assert!(validate_host("").is_err()); + } + + #[test] + fn validate_host_rejects_null_bytes() { + assert!(validate_host("evil\0.com").is_err()); + } + + #[test] + fn validate_host_rejects_overlength() { + let long = "a".repeat(256); + let err = validate_host(&long).unwrap_err(); + assert!(err.contains("too long")); + } + + #[test] + fn validate_host_accepts_max_length() { + let max = "a".repeat(255); + assert!(validate_host(&max).is_ok()); + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs new file mode 100644 index 000000000..abc4f1991 --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs @@ -0,0 +1,192 @@ +//! Shared stream helpers: byte-stream framing for the framed (legacy) +//! `net-read` / `net-write` path, raw byte-stream helpers for the std-style +//! `net-read-bytes` / `net-write-bytes` path, and the `with_tcp_slot` wrapper +//! used by every TCP-only getter/setter (peer-addr, nodelay, ttl, …). + +use crate::engine::wasm::bindings::astrid::capsule::types::NetReadStatus; +use crate::engine::wasm::host::util; +use crate::engine::wasm::host_state::{HostState, NetStream}; + +/// Bounded timeout on a `net-connect-tcp` outbound handshake. +/// +/// DNS resolve + TCP `connect` together must complete within this window, +/// otherwise the host fn returns an error rather than holding the WASM +/// guest in a host call indefinitely. +pub(super) const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +/// Host-side cap on a single byte-stream read/peek buffer. +/// +/// Matches the framed `net-read` payload cap so a malicious or buggy +/// guest passing `u32::MAX` can't trigger a multi-GB allocation on the +/// host. 10 MB is well above any realistic single-call read for the +/// protocols this surface targets (TLS records ≤ 16 KB, WebSocket +/// frames typically ≤ 64 KB, MQTT control packets ≤ 256 MB but +/// streamed in chunks). +pub(super) const MAX_BYTES_PER_CALL: usize = 10 * 1024 * 1024; + +/// Returns true for IO errors that represent a normal peer disconnect. +/// These should NOT trap the WASM guest — the run loop handles dead streams. +pub(super) fn is_peer_disconnect(e: &std::io::Error) -> bool { + matches!( + e.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::UnexpectedEof + ) +} + +/// Read one length-prefixed frame from `stream`. Shared by UnixStream and +/// TcpStream paths — both implement [`tokio::io::AsyncRead`]. +pub(super) async fn read_frame(stream: &mut S) -> Result +where + S: tokio::io::AsyncRead + Unpin, +{ + use tokio::io::AsyncReadExt; + + let mut len_buf = [0u8; 4]; + match tokio::time::timeout( + std::time::Duration::from_millis(50), + stream.read_exact(&mut len_buf), + ) + .await + { + Err(_) => return Ok(NetReadStatus::Pending), + Ok(Err(e)) if is_peer_disconnect(&e) => return Ok(NetReadStatus::Closed), + Ok(Err(e)) => return Err(format!("socket read error: {e}")), + Ok(Ok(_)) => {}, + } + + let len = u32::from_be_bytes(len_buf) as usize; + if len > 10 * 1024 * 1024 { + return Err("Payload too large (max 10MB)".to_string()); + } + + let mut payload = vec![0u8; len]; + let timeout_ms = 5000 + (len as u64 / 1024); + match tokio::time::timeout( + std::time::Duration::from_millis(timeout_ms), + stream.read_exact(&mut payload), + ) + .await + { + Err(_) => return Err("Payload read timed out".to_string()), + Ok(Err(e)) if is_peer_disconnect(&e) => return Ok(NetReadStatus::Closed), + Ok(Err(e)) => return Err(format!("socket payload read error: {e}")), + Ok(Ok(_)) => {}, + } + + Ok(NetReadStatus::Data(payload)) +} + +/// Write one length-prefixed frame to `stream`. Shared across UnixStream +/// and TcpStream — both implement [`tokio::io::AsyncWrite`]. +pub(super) async fn write_frame(stream: &mut S, data: &[u8]) -> std::io::Result<()> +where + S: tokio::io::AsyncWrite + Unpin, +{ + use tokio::io::AsyncWriteExt; + let len = u32::try_from(data.len()) + .map_err(|_| std::io::Error::other("write payload too large for length prefix"))?; + stream.write_all(&len.to_be_bytes()).await?; + stream.write_all(data).await?; + stream.flush().await?; + Ok(()) +} + +/// Read up to `max_bytes` from `stream` without length-prefix framing. +/// +/// Contract: +/// - `Ok(empty Vec)` = **EOF** (peer disconnected). Unambiguous — matches +/// `std::io::Read::read` returning `Ok(0)`. +/// - `Ok(non-empty Vec)` = data read. +/// - `Err("read would block")` = the caller-supplied `timeout` expired with +/// no data. Maps to `std::io::ErrorKind::WouldBlock` SDK-side. Only +/// returned when the caller has set a read timeout — `timeout = None` +/// blocks indefinitely (until data, EOF, or capsule unload via the +/// outer cancellation token). +/// - `Err(other)` = transient IO error. +pub(super) async fn read_bytes_inner( + stream: &mut S, + max_bytes: usize, + timeout: Option, +) -> Result, String> +where + S: tokio::io::AsyncRead + Unpin, +{ + use tokio::io::AsyncReadExt; + let max_bytes = max_bytes.min(MAX_BYTES_PER_CALL); + let mut buf = vec![0u8; max_bytes]; + // Default (no timeout) blocks indefinitely — matches std::io::Read + // semantics. Caller cancellation comes from the outer + // bounded_block_on_cancellable, not from a tight internal poll. + let result = match timeout { + Some(d) => tokio::time::timeout(d, stream.read(&mut buf)).await, + None => Ok(stream.read(&mut buf).await), + }; + match result { + Ok(Ok(0)) => Ok(Vec::new()), + Ok(Ok(n)) => { + buf.truncate(n); + Ok(buf) + }, + Ok(Err(e)) if is_peer_disconnect(&e) => Ok(Vec::new()), + Ok(Err(e)) => Err(format!("read error: {e}")), + Err(_) => Err("read would block".to_string()), + } +} + +/// Write `data` to `stream` without framing. Returns bytes-written. +/// Honours an optional `timeout`; the host-default behaviour (no +/// timeout) blocks until the write completes or the peer disconnects. +pub(super) async fn write_bytes_inner( + stream: &mut S, + data: &[u8], + timeout: Option, +) -> Result +where + S: tokio::io::AsyncWrite + Unpin, +{ + use tokio::io::AsyncWriteExt; + let fut = stream.write(data); + let n = match timeout { + Some(d) => match tokio::time::timeout(d, fut).await { + Ok(Ok(n)) => n, + Ok(Err(e)) if is_peer_disconnect(&e) => return Err(format!("write disconnect: {e}")), + Ok(Err(e)) => return Err(format!("write error: {e}")), + Err(_) => return Err("write would block".to_string()), + }, + None => fut.await.map_err(|e| format!("write error: {e}"))?, + }; + Ok(u32::try_from(n).unwrap_or(u32::MAX)) +} + +/// Run `op` against the inner [`tokio::net::TcpStream`] of an outbound +/// TCP stream handle. Returns an error if the handle is missing or holds +/// a Unix-domain stream (most std-style getters/setters are TCP-only). +pub(super) fn with_tcp_slot(state: &mut HostState, handle: u64, op: F) -> Result +where + F: FnOnce(&tokio::net::TcpStream) -> Result, +{ + let stream = state + .active_streams + .get(&handle) + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; + match stream { + NetStream::Tcp(slot) => { + let rt = state.runtime_handle.clone(); + let sem = state.host_semaphore.clone(); + let tok = state.cancel_token.clone(); + let result = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move { + let s = slot.stream.lock().await; + op(&s) + }); + match result { + Some(r) => r, + None => Err("capsule unloading".to_string()), + } + }, + NetStream::Unix(_) => Err("operation not supported on Unix streams".to_string()), + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host_state.rs b/crates/astrid-capsule/src/engine/wasm/host_state.rs index 12dc9c5ae..20838a91f 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state.rs @@ -15,6 +15,45 @@ use astrid_core::uplink::{InboundMessage, MAX_UPLINKS_PER_CAPSULE, UplinkDescrip use astrid_storage::ScopedKvStore; use astrid_storage::secret::SecretStore; +/// An active network stream owned by a capsule. +/// +/// Holds either an inbound Unix-socket connection (accepted from the +/// capsule's pre-provisioned listener) or an outbound TCP connection +/// (opened via `net.connect-tcp`). The read/write/close host fns +/// dispatch on the variant; both variants implement +/// [`tokio::io::AsyncRead`] / [`tokio::io::AsyncWrite`], so the inner +/// framing logic is shared. +#[derive(Debug, Clone)] +pub enum NetStream { + /// Inbound Unix-domain socket accepted from the kernel's listener. + Unix(Arc>), + /// Outbound TCP connection opened via `net.connect-tcp`. + Tcp(TcpStreamSlot), +} + +/// Per-stream state for an outbound TCP connection. +/// +/// Holds the raw [`tokio::net::TcpStream`] (behind an `Arc>` so +/// host fns that all take `&mut HostState` can each lock it cooperatively) +/// plus the std-style configurable socket options that the WIT surface +/// exposes — read/write timeouts. `TCP_NODELAY` and `TTL` are not cached +/// here; they live on the underlying socket and are read back via +/// `TcpStream::nodelay()` / `ttl()` on demand. +#[derive(Debug, Clone)] +pub struct TcpStreamSlot { + /// The connected TCP socket. Shared via `Arc>` so the + /// existing `net-read` / `net-write` dispatchers can clone the + /// outer `NetStream` and then lock the inner stream. + pub stream: Arc>, + /// Read timeout applied to each `net-read-bytes` / `net-peek` call. + /// `None` → use the host default (~50 ms poll for `Pending` parity + /// with `net-read`). + pub read_timeout: Option, + /// Write timeout applied to each `net-write-bytes` call. + /// `None` → no extra timeout beyond the host cancellation token. + pub write_timeout: Option, +} + /// The lifecycle phase a capsule is currently executing in. /// /// Set on [`HostState`] during `#[install]` or `#[upgrade]` dispatch. @@ -214,9 +253,8 @@ pub struct HostState { pub registered_uplinks: Vec, /// Optional natively bound unix listener. pub cli_socket_listener: Option>>, - /// Active, mapped UnixStreams from the socket listener. - pub active_streams: - std::collections::HashMap>>, + /// Active network streams owned by this capsule (Unix accept + outbound TCP). + pub active_streams: std::collections::HashMap, /// Monotonic counter for stream handle IDs (avoids reuse after removal). /// Starts at 1 so that handle ID 0 is never issued — 0 is reserved as a /// sentinel / "no handle" value in the WASM ABI. diff --git a/crates/astrid-capsule/src/manifest/capabilities.rs b/crates/astrid-capsule/src/manifest/capabilities.rs new file mode 100644 index 000000000..70a7e7ab7 --- /dev/null +++ b/crates/astrid-capsule/src/manifest/capabilities.rs @@ -0,0 +1,97 @@ +//! The `[capabilities]` block — what a capsule asks for from the OS. +//! +//! Every field is fail-closed by default (empty `Vec` or `false`). The kernel +//! security gates consult these allowlists before granting access. + +use serde::{Deserialize, Serialize}; + +/// A collection of capabilities the capsule requests from the OS. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct CapabilitiesDef { + /// Whether the capsule acts as a long-lived uplink/daemon (e.g. the CLI proxy). + /// When true, the WASM execution timeout is disabled. + #[serde(default)] + pub uplink: bool, + /// Network domains the capsule wants to access. + #[serde(default)] + pub net: Vec, + /// Scoped KV store access requests. + /// Note: KV access is inherently scoped per-capsule at runtime, + /// so this field is currently not enforced via a security gate, but + /// is present for future cross-capsule KV request declarations. + #[serde(default)] + pub kv: Vec, + /// VFS read paths. + #[serde(default)] + pub fs_read: Vec, + /// VFS write paths. + #[serde(default)] + pub fs_write: Vec, + /// Legacy host process executions (the "Airlock Override"). + #[serde(default)] + pub host_process: Vec, + /// Unix/TCP socket bind addresses the capsule requires. + #[serde(default)] + pub net_bind: Vec, + /// Outbound TCP destinations the capsule is allowed to connect to. + /// + /// Each entry is a `"host:port"` pattern. The `host` portion is a + /// literal DNS name or `*` (universal — see security review note + /// before allowing). The `port` portion is a decimal `u16` or `*` + /// (any port for the named host). Empty list → no outbound TCP + /// (fail-closed). Gated by the `astrid:capsule/net.net-connect-tcp` + /// host fn; the same kernel-side SSRF airlock that gates + /// `http-request` runs on the resolved IP after the capability + /// check passes. + #[serde(default)] + pub net_connect: Vec, + /// IPC topic patterns this capsule is allowed to publish to. + /// + /// Supports exact matches and `*` wildcards per segment + /// (e.g. `registry.*`, `llm.stream.anthropic`). + /// An empty list means the capsule may NOT publish to any topic + /// (fail-closed). Capsules must explicitly declare at least one + /// pattern to be allowed to publish. + #[serde(default)] + pub ipc_publish: Vec, + /// IPC topic patterns this capsule is allowed to subscribe to. + /// + /// Uses the same matching semantics as `ipc_publish`: exact matches + /// and `*` wildcards per segment, with segment counts required to + /// match. An empty list means the capsule may NOT subscribe to any + /// topic (fail-closed). + /// + /// Note: the ACL gates the subscription *pattern string*, not + /// individual messages. The ACL uses `topic_matches` semantics + /// (single-segment `*`, equal segment count required), but the + /// `EventBus` delivers events using `EventReceiver::matches` where + /// a trailing `*` matches one or more segments. This means + /// `ipc_subscribe = ["foo.v1.*"]` authorizes subscribing to the + /// pattern `"foo.v1.*"`, which the EventBus will use to deliver + /// events at any depth under `foo.v1.` - not just single-segment. + /// Per-message ACL checking would be O(n) per delivery and is + /// architecturally wrong for a broadcast bus. + #[serde(default)] + pub ipc_subscribe: Vec, + /// Identity operations this capsule is allowed to perform. + /// + /// Valid values: `"resolve"` (read-only lookups), `"link"` (create/delete + /// links, list links), `"admin"` (create users). The hierarchy is + /// `admin > link > resolve` - higher levels imply all lower levels. + /// + /// An empty list means NO identity access (fail-closed). + #[serde(default)] + pub identity: Vec, + /// Whether the capsule may override or modify the system prompt via the + /// prompt builder's hook pipeline. + /// + /// When `false` (default), hook responses from this capsule have their + /// `systemPrompt`, `prependSystemContext`, and `appendSystemContext` + /// fields stripped. Only `prependContext` (user-visible context) passes + /// through. + /// + /// This is a critical security boundary: unprivileged capsules cannot + /// inject arbitrary instructions into the LLM's system prompt. + #[serde(default)] + pub allow_prompt_injection: bool, +} diff --git a/crates/astrid-capsule/src/manifest.rs b/crates/astrid-capsule/src/manifest/mod.rs similarity index 65% rename from crates/astrid-capsule/src/manifest.rs rename to crates/astrid-capsule/src/manifest/mod.rs index 1e66505db..f35980b94 100644 --- a/crates/astrid-capsule/src/manifest.rs +++ b/crates/astrid-capsule/src/manifest/mod.rs @@ -11,6 +11,12 @@ use std::path::PathBuf; use serde::{Deserialize, Serialize}; use astrid_core::UplinkProfile; + +mod capabilities; +mod topics; + +pub use capabilities::CapabilitiesDef; +pub use topics::{PublishDef, SubscribeDef}; /// A capsule manifest loaded from `Capsule.toml`. /// /// Describes everything the runtime needs to know about a capsule before @@ -395,85 +401,6 @@ pub struct ComponentDef { pub capabilities: Option, } -/// A collection of capabilities the capsule requests from the OS. -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct CapabilitiesDef { - /// Whether the capsule acts as a long-lived uplink/daemon (e.g. the CLI proxy). - /// When true, the WASM execution timeout is disabled. - #[serde(default)] - pub uplink: bool, - /// Network domains the capsule wants to access. - #[serde(default)] - pub net: Vec, - /// Scoped KV store access requests. - /// Note: KV access is inherently scoped per-capsule at runtime, - /// so this field is currently not enforced via a security gate, but - /// is present for future cross-capsule KV request declarations. - #[serde(default)] - pub kv: Vec, - /// VFS read paths. - #[serde(default)] - pub fs_read: Vec, - /// VFS write paths. - #[serde(default)] - pub fs_write: Vec, - /// Legacy host process executions (the "Airlock Override"). - #[serde(default)] - pub host_process: Vec, - /// Unix/TCP socket bind addresses the capsule requires. - #[serde(default)] - pub net_bind: Vec, - /// IPC topic patterns this capsule is allowed to publish to. - /// - /// Supports exact matches and `*` wildcards per segment - /// (e.g. `registry.*`, `llm.stream.anthropic`). - /// An empty list means the capsule may NOT publish to any topic - /// (fail-closed). Capsules must explicitly declare at least one - /// pattern to be allowed to publish. - #[serde(default)] - pub ipc_publish: Vec, - /// IPC topic patterns this capsule is allowed to subscribe to. - /// - /// Uses the same matching semantics as `ipc_publish`: exact matches - /// and `*` wildcards per segment, with segment counts required to - /// match. An empty list means the capsule may NOT subscribe to any - /// topic (fail-closed). - /// - /// Note: the ACL gates the subscription *pattern string*, not - /// individual messages. The ACL uses `topic_matches` semantics - /// (single-segment `*`, equal segment count required), but the - /// `EventBus` delivers events using `EventReceiver::matches` where - /// a trailing `*` matches one or more segments. This means - /// `ipc_subscribe = ["foo.v1.*"]` authorizes subscribing to the - /// pattern `"foo.v1.*"`, which the EventBus will use to deliver - /// events at any depth under `foo.v1.` - not just single-segment. - /// Per-message ACL checking would be O(n) per delivery and is - /// architecturally wrong for a broadcast bus. - #[serde(default)] - pub ipc_subscribe: Vec, - /// Identity operations this capsule is allowed to perform. - /// - /// Valid values: `"resolve"` (read-only lookups), `"link"` (create/delete - /// links, list links), `"admin"` (create users). The hierarchy is - /// `admin > link > resolve` - higher levels imply all lower levels. - /// - /// An empty list means NO identity access (fail-closed). - #[serde(default)] - pub identity: Vec, - /// Whether the capsule may override or modify the system prompt via the - /// prompt builder's hook pipeline. - /// - /// When `false` (default), hook responses from this capsule have their - /// `systemPrompt`, `prependSystemContext`, and `appendSystemContext` - /// fields stripped. Only `prependContext` (user-visible context) passes - /// through. - /// - /// This is a critical security boundary: unprivileged capsules cannot - /// inject arbitrary instructions into the LLM's system prompt. - #[serde(default)] - pub allow_prompt_injection: bool, -} - /// An environment variable required by the capsule. /// /// These are securely elicited from the user during `capsule install` (docking). @@ -699,210 +626,6 @@ pub struct TopicDef { pub wit_type: Option, } -/// A topic this capsule publishes (RFC: cargo-like-manifest). -/// -/// Carries a typed WIT payload reference plus optional source pinning. The -/// containing key in `[publish]` is the topic name (or wildcard pattern). -/// -/// Two TOML surfaces accepted: -/// - Short: `"topic" = "@scope/repo/iface/record"` — bare WIT ref string -/// - Long: `"topic" = { wit = "...", version = "^1.0", fanout = true, ... }` -/// -/// Exactly one of `version` / `tag` / `rev` / `branch` / `path` may be set -/// for any external (`@scope/...`) reference. Bare-name local refs (no `@`) -/// need no source pin. The kernel does not yet enforce these constraints — -/// future resolver work (registry + lockfile + BLAKE3 verification) lives -/// behind the same RFC. -#[derive(Debug, Clone, Serialize)] -pub struct PublishDef { - /// Required typed payload reference. Either a bare local record name - /// (looks in this capsule's `wit/`) or `@scope/repo//` - /// (resolves through the registry / git source). The literal string - /// `"opaque"` marks an entry whose payload is not type-checked — used - /// by uplink/proxy capsules that route opaque bytes. - pub wit: String, - /// Registry-resolved version requirement (semver). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub version: Option, - /// Git tag pin (registry bypass). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub tag: Option, - /// Git SHA pin (registry bypass). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub rev: Option, - /// Git branch pin (floating; lockfile pins SHA at lock-time). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub branch: Option, - /// Local filesystem path (development; no checksum verification). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub path: Option, - /// Marks a wildcard publish where the suffix segment names a recipient - /// (e.g. `llm.v1.request.generate.*` per provider). Documentation hint - /// for tooling — kernel routes wildcards either way. - #[serde(default, skip_serializing_if = "std::ops::Not::not")] - pub fanout: bool, -} - -impl<'de> Deserialize<'de> for PublishDef { - fn deserialize(deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - // Accept either a bare WIT ref string (short form) or a full table. - // Defining the long form via #[derive] would recursively call this - // impl — use a private mirror struct with the derived impl instead. - #[derive(Deserialize)] - struct LongForm { - wit: String, - #[serde(default)] - version: Option, - #[serde(default)] - tag: Option, - #[serde(default)] - rev: Option, - #[serde(default)] - branch: Option, - #[serde(default)] - path: Option, - #[serde(default)] - fanout: bool, - } - #[derive(Deserialize)] - #[serde(untagged)] - enum Raw { - Short(String), - Long(LongForm), - } - let raw = Raw::deserialize(deserializer)?; - Ok(match raw { - Raw::Short(wit) => PublishDef { - wit, - version: None, - tag: None, - rev: None, - branch: None, - path: None, - fanout: false, - }, - Raw::Long(l) => { - let pins = [&l.version, &l.tag, &l.rev, &l.branch, &l.path] - .iter() - .filter(|o| o.is_some()) - .count(); - if pins > 1 { - return Err(serde::de::Error::custom( - "[publish] entry: at most one of version / tag / rev / branch / path may be set", - )); - } - PublishDef { - wit: l.wit, - version: l.version, - tag: l.tag, - rev: l.rev, - branch: l.branch, - path: l.path, - fanout: l.fanout, - } - }, - }) - } -} - -/// A topic this capsule subscribes to (RFC: cargo-like-manifest). -/// -/// Mirrors [`PublishDef`] plus an optional `handler` field that binds the -/// topic to a `#[astrid::interceptor("...")]` export in the WASM guest. -/// Entries without `handler` grant ACL only — the guest must still call -/// `ipc::subscribe()` to actually receive events. -/// -/// Same dual TOML surface as [`PublishDef`] (short string or table form). -#[derive(Debug, Clone, Serialize)] -pub struct SubscribeDef { - /// Required typed payload reference. See [`PublishDef::wit`]. - pub wit: String, - /// Registry-resolved version requirement. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub version: Option, - /// Git tag pin. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub tag: Option, - /// Git SHA pin. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub rev: Option, - /// Git branch pin (floating). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub branch: Option, - /// Local filesystem path. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub path: Option, - /// Name of the `#[astrid::interceptor("...")]` export to bind. When set, - /// supersedes any `[[interceptor]]` block targeting the same event. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub handler: Option, -} - -impl<'de> Deserialize<'de> for SubscribeDef { - fn deserialize(deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - #[derive(Deserialize)] - struct LongForm { - wit: String, - #[serde(default)] - version: Option, - #[serde(default)] - tag: Option, - #[serde(default)] - rev: Option, - #[serde(default)] - branch: Option, - #[serde(default)] - path: Option, - #[serde(default)] - handler: Option, - } - #[derive(Deserialize)] - #[serde(untagged)] - enum Raw { - Short(String), - Long(LongForm), - } - let raw = Raw::deserialize(deserializer)?; - Ok(match raw { - Raw::Short(wit) => SubscribeDef { - wit, - version: None, - tag: None, - rev: None, - branch: None, - path: None, - handler: None, - }, - Raw::Long(l) => { - let pins = [&l.version, &l.tag, &l.rev, &l.branch, &l.path] - .iter() - .filter(|o| o.is_some()) - .count(); - if pins > 1 { - return Err(serde::de::Error::custom( - "[subscribe] entry: at most one of version / tag / rev / branch / path may be set", - )); - } - SubscribeDef { - wit: l.wit, - version: l.version, - tag: l.tag, - rev: l.rev, - branch: l.branch, - path: l.path, - handler: l.handler, - } - }, - }) - } -} - /// A tool this capsule surfaces to the LLM (RFC: cargo-like-manifest). /// /// `description_for_llm` is the *only* capsule-author-controlled string @@ -922,78 +645,3 @@ pub struct ToolDef { #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub mutable: bool, } - -#[cfg(test)] -mod tests { - use super::*; - - fn parse_publish(entry: &str) -> Result { - let toml = format!("\"x.v1.y\" = {entry}\n"); - let map: std::collections::HashMap = toml::from_str(&toml)?; - Ok(map.into_iter().next().unwrap().1) - } - - fn parse_subscribe(entry: &str) -> Result { - let toml = format!("\"x.v1.y\" = {entry}\n"); - let map: std::collections::HashMap = toml::from_str(&toml)?; - Ok(map.into_iter().next().unwrap().1) - } - - #[test] - fn publish_short_form_parses() { - let p = parse_publish("\"@scope/wit/iface/rec\"").unwrap(); - assert_eq!(p.wit, "@scope/wit/iface/rec"); - assert!(p.version.is_none() && p.tag.is_none() && p.rev.is_none()); - } - - #[test] - fn publish_long_form_zero_pins_parses() { - let p = parse_publish("{ wit = \"r\" }").unwrap(); - assert_eq!(p.wit, "r"); - } - - #[test] - fn publish_long_form_one_pin_parses() { - let p = parse_publish("{ wit = \"r\", version = \"1.0\" }").unwrap(); - assert_eq!(p.version.as_deref(), Some("1.0")); - } - - #[test] - fn publish_long_form_two_pins_rejected() { - let err = parse_publish("{ wit = \"r\", version = \"1.0\", tag = \"v1\" }").unwrap_err(); - let msg = err.to_string(); - assert!( - msg.contains("at most one of version / tag / rev / branch / path"), - "missing invariant message: {msg}" - ); - assert!( - msg.contains("x.v1.y"), - "TOML deserializer should include the topic key in the error context: {msg}" - ); - } - - #[test] - fn subscribe_short_form_parses() { - let s = parse_subscribe("\"r\"").unwrap(); - assert_eq!(s.wit, "r"); - assert!(s.handler.is_none()); - } - - #[test] - fn subscribe_long_form_one_pin_with_handler_parses() { - let s = parse_subscribe("{ wit = \"r\", rev = \"abc123\", handler = \"on_x\" }").unwrap(); - assert_eq!(s.rev.as_deref(), Some("abc123")); - assert_eq!(s.handler.as_deref(), Some("on_x")); - } - - #[test] - fn subscribe_long_form_two_pins_rejected() { - let err = - parse_subscribe("{ wit = \"r\", branch = \"main\", path = \"./local\" }").unwrap_err(); - let msg = err.to_string(); - assert!( - msg.contains("at most one of version / tag / rev / branch / path"), - "missing invariant message: {msg}" - ); - } -} diff --git a/crates/astrid-capsule/src/manifest/topics.rs b/crates/astrid-capsule/src/manifest/topics.rs new file mode 100644 index 000000000..f413e1e9f --- /dev/null +++ b/crates/astrid-capsule/src/manifest/topics.rs @@ -0,0 +1,289 @@ +//! Cargo-shaped `[publish]` / `[subscribe]` tables. +//! +//! Each entry carries a typed WIT payload reference plus optional source +//! pinning (`version` / `tag` / `rev` / `branch` / `path`). Two TOML +//! surfaces accepted per entry: a bare WIT-ref string (short form) or a +//! full inline table (long form). Exactly one source pin may be set on +//! the long form; the deserializer rejects ambiguous manifests at parse +//! time. + +use serde::{Deserialize, Serialize}; + +/// A topic this capsule publishes (RFC: cargo-like-manifest). +/// +/// Carries a typed WIT payload reference plus optional source pinning. The +/// containing key in `[publish]` is the topic name (or wildcard pattern). +/// +/// Two TOML surfaces accepted: +/// - Short: `"topic" = "@scope/repo/iface/record"` — bare WIT ref string +/// - Long: `"topic" = { wit = "...", version = "^1.0", fanout = true, ... }` +/// +/// Exactly one of `version` / `tag` / `rev` / `branch` / `path` may be set +/// for any external (`@scope/...`) reference. Bare-name local refs (no `@`) +/// need no source pin. The kernel does not yet enforce these constraints — +/// future resolver work (registry + lockfile + BLAKE3 verification) lives +/// behind the same RFC. +#[derive(Debug, Clone, Serialize)] +pub struct PublishDef { + /// Required typed payload reference. Either a bare local record name + /// (looks in this capsule's `wit/`) or `@scope/repo//` + /// (resolves through the registry / git source). The literal string + /// `"opaque"` marks an entry whose payload is not type-checked — used + /// by uplink/proxy capsules that route opaque bytes. + pub wit: String, + /// Registry-resolved version requirement (semver). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub version: Option, + /// Git tag pin (registry bypass). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tag: Option, + /// Git SHA pin (registry bypass). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rev: Option, + /// Git branch pin (floating; lockfile pins SHA at lock-time). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch: Option, + /// Local filesystem path (development; no checksum verification). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub path: Option, + /// Marks a wildcard publish where the suffix segment names a recipient + /// (e.g. `llm.v1.request.generate.*` per provider). Documentation hint + /// for tooling — kernel routes wildcards either way. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub fanout: bool, +} + +impl<'de> Deserialize<'de> for PublishDef { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + // Accept either a bare WIT ref string (short form) or a full table. + // Defining the long form via #[derive] would recursively call this + // impl — use a private mirror struct with the derived impl instead. + #[derive(Deserialize)] + struct LongForm { + wit: String, + #[serde(default)] + version: Option, + #[serde(default)] + tag: Option, + #[serde(default)] + rev: Option, + #[serde(default)] + branch: Option, + #[serde(default)] + path: Option, + #[serde(default)] + fanout: bool, + } + #[derive(Deserialize)] + #[serde(untagged)] + enum Raw { + Short(String), + Long(LongForm), + } + let raw = Raw::deserialize(deserializer)?; + Ok(match raw { + Raw::Short(wit) => PublishDef { + wit, + version: None, + tag: None, + rev: None, + branch: None, + path: None, + fanout: false, + }, + Raw::Long(l) => { + let pins = [&l.version, &l.tag, &l.rev, &l.branch, &l.path] + .iter() + .filter(|o| o.is_some()) + .count(); + if pins > 1 { + return Err(serde::de::Error::custom( + "[publish] entry: at most one of version / tag / rev / branch / path may be set", + )); + } + PublishDef { + wit: l.wit, + version: l.version, + tag: l.tag, + rev: l.rev, + branch: l.branch, + path: l.path, + fanout: l.fanout, + } + }, + }) + } +} + +/// A topic this capsule subscribes to (RFC: cargo-like-manifest). +/// +/// Mirrors [`PublishDef`] plus an optional `handler` field that binds the +/// topic to a `#[astrid::interceptor("...")]` export in the WASM guest. +/// Entries without `handler` grant ACL only — the guest must still call +/// `ipc::subscribe()` to actually receive events. +/// +/// Same dual TOML surface as [`PublishDef`] (short string or table form). +#[derive(Debug, Clone, Serialize)] +pub struct SubscribeDef { + /// Required typed payload reference. See [`PublishDef::wit`]. + pub wit: String, + /// Registry-resolved version requirement. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub version: Option, + /// Git tag pin. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tag: Option, + /// Git SHA pin. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rev: Option, + /// Git branch pin (floating). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch: Option, + /// Local filesystem path. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub path: Option, + /// Name of the `#[astrid::interceptor("...")]` export to bind. When set, + /// supersedes any `[[interceptor]]` block targeting the same event. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub handler: Option, +} + +impl<'de> Deserialize<'de> for SubscribeDef { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + struct LongForm { + wit: String, + #[serde(default)] + version: Option, + #[serde(default)] + tag: Option, + #[serde(default)] + rev: Option, + #[serde(default)] + branch: Option, + #[serde(default)] + path: Option, + #[serde(default)] + handler: Option, + } + #[derive(Deserialize)] + #[serde(untagged)] + enum Raw { + Short(String), + Long(LongForm), + } + let raw = Raw::deserialize(deserializer)?; + Ok(match raw { + Raw::Short(wit) => SubscribeDef { + wit, + version: None, + tag: None, + rev: None, + branch: None, + path: None, + handler: None, + }, + Raw::Long(l) => { + let pins = [&l.version, &l.tag, &l.rev, &l.branch, &l.path] + .iter() + .filter(|o| o.is_some()) + .count(); + if pins > 1 { + return Err(serde::de::Error::custom( + "[subscribe] entry: at most one of version / tag / rev / branch / path may be set", + )); + } + SubscribeDef { + wit: l.wit, + version: l.version, + tag: l.tag, + rev: l.rev, + branch: l.branch, + path: l.path, + handler: l.handler, + } + }, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parse_publish(entry: &str) -> Result { + let toml = format!("\"x.v1.y\" = {entry}\n"); + let map: std::collections::HashMap = toml::from_str(&toml)?; + Ok(map.into_iter().next().unwrap().1) + } + + fn parse_subscribe(entry: &str) -> Result { + let toml = format!("\"x.v1.y\" = {entry}\n"); + let map: std::collections::HashMap = toml::from_str(&toml)?; + Ok(map.into_iter().next().unwrap().1) + } + + #[test] + fn publish_short_form_parses() { + let p = parse_publish("\"@scope/wit/iface/rec\"").unwrap(); + assert_eq!(p.wit, "@scope/wit/iface/rec"); + assert!(p.version.is_none() && p.tag.is_none() && p.rev.is_none()); + } + + #[test] + fn publish_long_form_zero_pins_parses() { + let p = parse_publish("{ wit = \"r\" }").unwrap(); + assert_eq!(p.wit, "r"); + } + + #[test] + fn publish_long_form_one_pin_parses() { + let p = parse_publish("{ wit = \"r\", version = \"1.0\" }").unwrap(); + assert_eq!(p.version.as_deref(), Some("1.0")); + } + + #[test] + fn publish_long_form_two_pins_rejected() { + let err = parse_publish("{ wit = \"r\", version = \"1.0\", tag = \"v1\" }").unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("at most one of version / tag / rev / branch / path"), + "missing invariant message: {msg}" + ); + assert!( + msg.contains("x.v1.y"), + "TOML deserializer should include the topic key in the error context: {msg}" + ); + } + + #[test] + fn subscribe_short_form_parses() { + let s = parse_subscribe("\"r\"").unwrap(); + assert_eq!(s.wit, "r"); + assert!(s.handler.is_none()); + } + + #[test] + fn subscribe_long_form_one_pin_with_handler_parses() { + let s = parse_subscribe("{ wit = \"r\", rev = \"abc123\", handler = \"on_x\" }").unwrap(); + assert_eq!(s.rev.as_deref(), Some("abc123")); + assert_eq!(s.handler.as_deref(), Some("on_x")); + } + + #[test] + fn subscribe_long_form_two_pins_rejected() { + let err = + parse_subscribe("{ wit = \"r\", branch = \"main\", path = \"./local\" }").unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("at most one of version / tag / rev / branch / path"), + "missing invariant message: {msg}" + ); + } +} diff --git a/crates/astrid-capsule/src/security/manifest_gate.rs b/crates/astrid-capsule/src/security/manifest_gate.rs index bde1d760c..37cd12d47 100644 --- a/crates/astrid-capsule/src/security/manifest_gate.rs +++ b/crates/astrid-capsule/src/security/manifest_gate.rs @@ -265,6 +265,30 @@ impl CapsuleSecurityGate for ManifestSecurityGate { } } + async fn check_net_connect( + &self, + capsule_id: &str, + host: &str, + port: u16, + ) -> Result<(), String> { + // Each allowlist entry is "host:port" or "host:*". Match against the + // literal host the capsule named; DNS resolution and SSRF check run + // after this gate. + let allowed = self + .manifest + .capabilities + .net_connect + .iter() + .any(|entry| net_connect_pattern_matches(entry, host, port)); + if allowed { + Ok(()) + } else { + Err(format!( + "capsule '{capsule_id}' denied: \"{host}:{port}\" not in net_connect allowlist" + )) + } + } + async fn check_identity( &self, capsule_id: &str, @@ -283,6 +307,29 @@ impl CapsuleSecurityGate for ManifestSecurityGate { } } +/// Match a `net_connect` allowlist entry against a literal `host:port`. +/// +/// Patterns: +/// - `"host:port"` — exact match. +/// - `"host:*"` — any port for the named host. +/// +/// Hostnames are compared case-insensitively (DNS names are case-insensitive +/// per RFC 1035). The pattern host segment is taken literally — DNS-style +/// wildcards (`*.example.com`) are intentionally NOT supported in this version +/// (see RFC: rfcs#27 Unresolved questions). +fn net_connect_pattern_matches(pattern: &str, host: &str, port: u16) -> bool { + let Some((pat_host, pat_port)) = pattern.rsplit_once(':') else { + return false; + }; + if !pat_host.eq_ignore_ascii_case(host) { + return false; + } + match pat_port { + "*" => true, + p => p.parse::().is_ok_and(|n| n == port), + } +} + #[cfg(test)] mod tests { use std::collections::HashMap; @@ -317,6 +364,7 @@ mod tests { capabilities: CapabilitiesDef { net: net.into_iter().map(String::from).collect(), net_bind: vec![], + net_connect: vec![], kv: vec![], fs_read: fs_read.into_iter().map(String::from).collect(), fs_write: fs_write.into_iter().map(String::from).collect(), @@ -787,4 +835,102 @@ mod tests { .is_ok() ); } + + #[test] + fn net_connect_exact_match() { + assert!(net_connect_pattern_matches( + "example.com:443", + "example.com", + 443 + )); + } + + #[test] + fn net_connect_port_mismatch_is_denied() { + assert!(!net_connect_pattern_matches( + "example.com:443", + "example.com", + 80 + )); + } + + #[test] + fn net_connect_host_mismatch_is_denied() { + assert!(!net_connect_pattern_matches( + "example.com:443", + "evil.com", + 443 + )); + } + + #[test] + fn net_connect_port_wildcard_matches_any_port() { + assert!(net_connect_pattern_matches( + "example.com:*", + "example.com", + 1 + )); + assert!(net_connect_pattern_matches( + "example.com:*", + "example.com", + 65535 + )); + } + + #[test] + fn net_connect_host_is_case_insensitive() { + assert!(net_connect_pattern_matches( + "Example.COM:443", + "example.com", + 443 + )); + } + + #[test] + fn net_connect_missing_colon_is_denied() { + assert!(!net_connect_pattern_matches( + "example.com", + "example.com", + 80 + )); + } + + #[test] + fn net_connect_invalid_port_is_denied() { + assert!(!net_connect_pattern_matches( + "example.com:abc", + "example.com", + 80 + )); + } + + #[tokio::test] + async fn check_net_connect_default_denies_with_empty_allowlist() { + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_connect = vec![]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + let err = gate + .check_net_connect("c", "example.com", 443) + .await + .unwrap_err(); + assert!(err.contains("not in net_connect allowlist"), "{err}"); + } + + #[tokio::test] + async fn check_net_connect_matches_allowlist_entry() { + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_connect = vec!["example.com:443".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!( + gate.check_net_connect("c", "example.com", 443) + .await + .is_ok() + ); + assert!( + gate.check_net_connect("c", "example.com", 80) + .await + .is_err() + ); + assert!(gate.check_net_connect("c", "evil.com", 443).await.is_err()); + } } diff --git a/crates/astrid-capsule/src/security/mod.rs b/crates/astrid-capsule/src/security/mod.rs index f3c1fd93e..78f1fbb5d 100644 --- a/crates/astrid-capsule/src/security/mod.rs +++ b/crates/astrid-capsule/src/security/mod.rs @@ -120,6 +120,25 @@ pub trait CapsuleSecurityGate: Send + Sync { )) } + /// Check whether the capsule is allowed to open an outbound TCP connection. + /// + /// Default implementation denies. Override to permit capsules that + /// declare `net_connect` capabilities matching `host:port`. + /// + /// The check is on the literal `host:port` the capsule passed to + /// `net.connect-tcp`; DNS resolution and the SSRF airlock run + /// kernel-side *after* this gate returns `Ok`. + async fn check_net_connect( + &self, + capsule_id: &str, + _host: &str, + _port: u16, + ) -> Result<(), String> { + Err(format!( + "capsule '{capsule_id}' denied: net_connect 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/src/security/test_gates.rs b/crates/astrid-capsule/src/security/test_gates.rs index 3c3eb95dc..d2c1fca99 100644 --- a/crates/astrid-capsule/src/security/test_gates.rs +++ b/crates/astrid-capsule/src/security/test_gates.rs @@ -49,6 +49,15 @@ impl CapsuleSecurityGate for AllowAllGate { Ok(()) } + async fn check_net_connect( + &self, + _capsule_id: &str, + _host: &str, + _port: u16, + ) -> Result<(), String> { + Ok(()) + } + async fn check_uplink_register( &self, _capsule_id: &str, @@ -118,6 +127,17 @@ impl CapsuleSecurityGate for DenyAllGate { )) } + async fn check_net_connect( + &self, + capsule_id: &str, + host: &str, + port: u16, + ) -> Result<(), String> { + Err(format!( + "capsule '{capsule_id}' denied: net_connect {host}:{port} (DenyAllGate)" + )) + } + async fn check_uplink_register( &self, capsule_id: &str, diff --git a/crates/astrid-integration-tests/tests/mcp_e2e.rs b/crates/astrid-integration-tests/tests/mcp_e2e.rs index 1791263ca..0e37bd350 100644 --- a/crates/astrid-integration-tests/tests/mcp_e2e.rs +++ b/crates/astrid-integration-tests/tests/mcp_e2e.rs @@ -36,6 +36,7 @@ async fn test_mcp_host_engine_capability_validation() { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + net_connect: vec![], kv: vec![], fs_read: vec![], fs_write: vec![], diff --git a/crates/astrid-integration-tests/tests/wasm_e2e.rs b/crates/astrid-integration-tests/tests/wasm_e2e.rs index 73db629b1..a01169c6f 100644 --- a/crates/astrid-integration-tests/tests/wasm_e2e.rs +++ b/crates/astrid-integration-tests/tests/wasm_e2e.rs @@ -61,6 +61,7 @@ async fn setup_test_capsule( capabilities: CapabilitiesDef { net: net_caps, net_bind: vec![], + net_connect: vec![], kv: vec!["*".into()], fs_read: fs_read_caps, fs_write: fs_write_caps, @@ -165,6 +166,7 @@ async fn setup_test_capsule_with_home( capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + net_connect: vec![], kv: vec!["*".into()], fs_read: fs_read_caps, fs_write: fs_write_caps, diff --git a/crates/astrid-integration-tests/tests/wasm_env_e2e.rs b/crates/astrid-integration-tests/tests/wasm_env_e2e.rs index 61aaecd70..329de532e 100644 --- a/crates/astrid-integration-tests/tests/wasm_env_e2e.rs +++ b/crates/astrid-integration-tests/tests/wasm_env_e2e.rs @@ -85,6 +85,7 @@ async fn test_wasm_capsule_e2e_env_config_injection() { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + net_connect: vec![], kv: vec!["*".into()], fs_read: vec![], fs_write: vec![], diff --git a/wit/astrid-capsule.wit b/wit/astrid-capsule.wit index 53c802f59..05db39502 100644 --- a/wit/astrid-capsule.wit +++ b/wit/astrid-capsule.wit @@ -206,6 +206,20 @@ interface types { payload: string, /// UUID of the capsule that sent this message. source-id: string, + /// Principal attributed to the publisher of this message. + /// + /// For messages published via `ipc-publish`, this is the publishing + /// capsule's invocation principal (whoever the host attributed the + /// call to). For messages published via `ipc-publish-as`, this is + /// the principal the uplink claimed on behalf of an external caller. + /// + /// `none` for system / kernel-originated events that have no + /// attributable principal, and for legacy messages that predate + /// this field. Subscribers processing multi-message recv batches + /// should read this per-message rather than relying on the + /// invocation context (which only reflects the first message's + /// publisher). + principal: option, } /// Pre-registered interceptor handle mapping. @@ -338,6 +352,20 @@ interface types { /// Whether the action was approved. approved: bool, } + + /// Direction argument for `net-shutdown` — mirrors `std::net::Shutdown`. + enum shutdown-how { + /// Half-close the read side. Subsequent reads return EOF; writes + /// continue. + read, + /// Half-close the write side. The peer sees EOF on its read side; + /// reads continue. + write, + /// Close both directions. Equivalent to `net-close-stream` except + /// the handle entry is retained so getters (`peer-addr`, etc.) + /// still work until the explicit close call. + both, + } } // --------------------------------------------------------------------------- @@ -517,7 +545,7 @@ interface kv { /// /// Max 8 concurrent streams per capsule. interface net { - use types.{net-read-status}; + use types.{net-read-status, shutdown-how}; /// Bind and activate the pre-provisioned Unix socket listener. /// @@ -555,6 +583,122 @@ interface net { /// Idempotent: closing an already-closed handle is a no-op. /// Publishes a `client.v1.disconnect` IPC event. net-close-stream: func(stream-handle: u64) -> result<_, string>; + + /// Open an outbound TCP connection to `host:port`. + /// + /// Returns a stream handle compatible with the existing + /// `net-read` / `net-write` / `net-close-stream` functions + /// — the same handle type returned by `net-accept`. + /// + /// Security gates (all checked before any TCP syscall): + /// - The capsule's `net_connect` capability allowlist must + /// contain a pattern matching `host:port`. Patterns are + /// `"host:port"` exact matches or `"host:*"` (any port for + /// the named host). Missing or empty allowlist denies all + /// outbound TCP (fail-closed). + /// - DNS resolution runs after the capability check. The + /// resolved IP is rejected if it falls into a private, + /// loopback, link-local, multicast, or unspecified range, + /// matching the airlock that gates `http-request` / + /// `http-stream-start`. + /// - The capsule's per-instance active-stream cap (default + /// 8, shared with `net-accept`) is enforced. + /// - Connect attempts time out (default 10s) rather than + /// holding the WASM guest in a host fn indefinitely. + net-connect-tcp: func(host: string, port: u16) -> result; + + /// Read up to `max-bytes` from `stream` without length-prefix framing. + /// + /// Mirrors `std::net::TcpStream::read` (and `::read`). + /// Returns the bytes that were available — may be shorter than + /// `max-bytes`. Empty list means no data is ready under the current + /// read timeout (default: non-blocking, ~50 ms internal poll). + /// + /// Use this for byte-stream protocols (WebSocket, MQTT, postgres, + /// HTTP/1.1, TLS, …). The existing `net-read` keeps the length-prefix + /// framing required by the uplink-proxy use case. + net-read-bytes: func(stream-handle: u64, max-bytes: u32) -> result, string>; + + /// Write `data` to `stream` without length-prefix framing. + /// + /// Mirrors `::write`. Returns the number of + /// bytes actually written — may be less than `data.len()` when the + /// kernel-side socket buffer is full. Callers loop until `data` is + /// drained (the SDK `write_all` wrapper handles this). + net-write-bytes: func(stream-handle: u64, data: list) -> result; + + /// Read up to `max-bytes` from `stream` without consuming them. + /// + /// Mirrors `std::net::TcpStream::peek`. Subsequent `net-read-bytes` + /// returns the same bytes again. Useful for protocol detection on + /// the first frame of a connection. + net-peek: func(stream-handle: u64, max-bytes: u32) -> result, string>; + + /// Shut down the read side, write side, or both halves of `stream`. + /// + /// Mirrors `std::net::TcpStream::shutdown`. After `shutdown-how::write` + /// the peer sees EOF on its read side; further sends on the local + /// side return an error. `shutdown-how::both` closes both directions + /// but leaves the handle entry intact so accessors (e.g. + /// `peer-addr`) still work until `net-close-stream` is called. + net-shutdown: func(stream-handle: u64, how: shutdown-how) -> result<_, string>; + + /// Return the remote peer address as `"ip:port"`. + /// + /// Mirrors `std::net::TcpStream::peer_addr`. Returns an error for + /// Unix-domain stream handles. + net-peer-addr: func(stream-handle: u64) -> result; + + /// Return the local socket address as `"ip:port"`. + /// + /// Mirrors `std::net::TcpStream::local_addr`. Returns an error for + /// Unix-domain stream handles. + net-local-addr: func(stream-handle: u64) -> result; + + /// Enable or disable the `TCP_NODELAY` option (Nagle's algorithm off + /// when `true`). + /// + /// Mirrors `std::net::TcpStream::set_nodelay`. Returns an error for + /// Unix-domain stream handles. + net-set-nodelay: func(stream-handle: u64, nodelay: bool) -> result<_, string>; + + /// Return the current `TCP_NODELAY` setting. + /// + /// Mirrors `std::net::TcpStream::nodelay`. + net-nodelay: func(stream-handle: u64) -> result; + + /// Set the read timeout. `none` clears the timeout (reads still + /// honour the host's internal cancellation token; this controls + /// only the user-visible blocking duration of each `net-read-bytes` + /// call). + /// + /// Mirrors `std::net::TcpStream::set_read_timeout`. Subsequent + /// `net-read-bytes` and `net-peek` calls honour the new value. + net-set-read-timeout: func(stream-handle: u64, timeout-ms: option) -> result<_, string>; + + /// Return the current read timeout in milliseconds, or `none` if + /// unset. + net-read-timeout: func(stream-handle: u64) -> result, string>; + + /// Set the write timeout. `none` clears it. + /// + /// Mirrors `std::net::TcpStream::set_write_timeout`. + net-set-write-timeout: func(stream-handle: u64, timeout-ms: option) -> result<_, string>; + + /// Return the current write timeout in milliseconds, or `none` if + /// unset. + net-write-timeout: func(stream-handle: u64) -> result, string>; + + /// Set the IP `TTL` field on outgoing packets. + /// + /// Mirrors `std::net::TcpStream::set_ttl`. Returns an error for + /// Unix-domain stream handles. + net-set-ttl: func(stream-handle: u64, ttl: u32) -> result<_, string>; + + /// Return the current IP TTL. + /// + /// Mirrors `std::net::TcpStream::ttl`. + net-ttl: func(stream-handle: u64) -> result; } /// HTTP client operations with SSRF protection.