From b1572d523d78fc72552b864c140d6ef1cc981a20 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 19 May 2026 14:21:32 +0400 Subject: [PATCH 1/5] feat(net): net-connect-tcp host fn + net_connect capability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds one outbound-TCP host fn — astrid:capsule/net.net-connect-tcp(host, port) -> stream-handle — gated by a per-capsule net_connect allowlist in Capsule.toml. Reuses the existing net-read / net-write / net-close-stream plumbing (single new variant on a NetStream enum; framing helpers shared across UnixStream and TcpStream via tokio::io::AsyncRead/AsyncWrite trait bounds). SSRF airlock from http-request runs on the resolved IP after the capability check. 10s connect timeout. Per-capsule active-stream cap (8) shared with inbound net-accept. Unblocks WebSocket clients, MQTT, Discord/Telegram, postgres/redis. The immediate motivator is a Unicity-network capsule wrapping Sphere SDK (Fulcrum + Nostr WebSocket transports). Also splits the now-1000-line manifest.rs into a manifest/ submodule (capabilities.rs, topics.rs) to land under the 1000-line CI cap. Public API preserved via pub use re-exports. And resyncs wit/astrid-capsule.wit from canonical (picks up the principal field on ipc-message that drifted in from canonical PR #4). RFC: unicity-astrid/rfcs#27 WIT: unicity-astrid/wit#5 SDK: unicity-astrid/sdk-rust feat/net-connect-tcp Tracking issue: closes #745 --- CHANGELOG.md | 11 + crates/astrid-capsule/src/engine/mcp_tests.rs | 1 + .../astrid-capsule/src/engine/wasm/host/fs.rs | 1 + .../src/engine/wasm/host/http.rs | 2 +- .../src/engine/wasm/host/ipc.rs | 1 + .../src/engine/wasm/host/net.rs | 244 +++++++++--- .../src/engine/wasm/host_state.rs | 21 +- .../src/manifest/capabilities.rs | 97 +++++ .../src/{manifest.rs => manifest/mod.rs} | 364 +----------------- crates/astrid-capsule/src/manifest/topics.rs | 289 ++++++++++++++ .../src/security/manifest_gate.rs | 146 +++++++ crates/astrid-capsule/src/security/mod.rs | 19 + .../astrid-capsule/src/security/test_gates.rs | 20 + .../astrid-integration-tests/tests/mcp_e2e.rs | 1 + .../tests/wasm_e2e.rs | 2 + .../tests/wasm_env_e2e.rs | 1 + wit/astrid-capsule.wit | 37 ++ 17 files changed, 838 insertions(+), 419 deletions(-) create mode 100644 crates/astrid-capsule/src/manifest/capabilities.rs rename crates/astrid-capsule/src/{manifest.rs => manifest/mod.rs} (65%) create mode 100644 crates/astrid-capsule/src/manifest/topics.rs 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/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 index 49d15ff8c..f06ae0eaf 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net.rs @@ -4,8 +4,16 @@ use astrid_core::session_token::{ use crate::engine::wasm::bindings::astrid::capsule::net; use crate::engine::wasm::bindings::astrid::capsule::types::NetReadStatus; +use crate::engine::wasm::host::http::is_safe_ip; use crate::engine::wasm::host::util; -use crate::engine::wasm::host_state::HostState; +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. +const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); /// Maximum concurrent socket connections per capsule. /// Prevents resource exhaustion from malicious or runaway clients. @@ -23,6 +31,64 @@ fn is_peer_disconnect(e: &std::io::Error) -> bool { ) } +/// Read one length-prefixed frame from `stream`. Shared by UnixStream and +/// TcpStream paths — both implement [`tokio::io::AsyncRead`]. +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`]. +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(()) +} + // --------------------------------------------------------------------------- // Handshake helpers // --------------------------------------------------------------------------- @@ -309,7 +375,7 @@ impl net::Host for HostState { ); self.active_streams.insert( handle_id, - std::sync::Arc::new(tokio::sync::Mutex::new(stream)), + NetStream::Unix(std::sync::Arc::new(tokio::sync::Mutex::new(stream))), ); // The `client.v1.connected` event is intentionally NOT published @@ -423,7 +489,7 @@ impl net::Host for HostState { ); self.active_streams.insert( handle_id, - std::sync::Arc::new(tokio::sync::Mutex::new(stream)), + NetStream::Unix(std::sync::Arc::new(tokio::sync::Mutex::new(stream))), ); // See the matching comment in `net_accept` — `client.v1.connected` @@ -435,11 +501,11 @@ impl net::Host for HostState { } fn net_read(&mut self, stream_handle: u64) -> Result { - let stream_arc = self + let stream = self .active_streams .get(&stream_handle) - .ok_or_else(|| "Stream handle not found".to_string())? - .clone(); + .cloned() + .ok_or_else(|| "Stream handle not found".to_string())?; let rt_handle = self.runtime_handle.clone(); let cancel_token = self.cancel_token.clone(); @@ -449,49 +515,18 @@ impl net::Host for HostState { // 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); + match stream { + NetStream::Unix(arc) => { + let mut s = arc.lock().await; + read_frame(&mut *s).await }, - 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); + NetStream::Tcp(arc) => { + let mut s = arc.lock().await; + read_frame(&mut *s).await }, - 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. @@ -502,32 +537,32 @@ impl net::Host for HostState { } fn net_write(&mut self, stream_handle: u64, data: Vec) -> Result<(), String> { - let stream_arc = self + let stream = self .active_streams .get(&stream_handle) - .ok_or_else(|| "Stream handle not found".to_string())? - .clone(); + .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(); - 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 stream { + NetStream::Unix(arc) => { + let mut s = arc.lock().await; + write_frame(&mut *s, &data).await + }, + NetStream::Tcp(arc) => { + let mut s = arc.lock().await; + write_frame(&mut *s, &data).await + }, + } }); match result { Some(Ok(())) => {}, @@ -542,6 +577,101 @@ impl net::Host for HostState { Ok(()) } + fn net_connect_tcp(&mut self, host: String, port: u16) -> Result { + // 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(std::sync::Arc::new(tokio::sync::Mutex::new(stream))), + ); + Ok(handle_id) + } + fn net_close_stream(&mut self, stream_handle: u64) -> Result<(), String> { // Idempotent: silently ignore if the handle was already removed. // diff --git a/crates/astrid-capsule/src/engine/wasm/host_state.rs b/crates/astrid-capsule/src/engine/wasm/host_state.rs index 12dc9c5ae..26fc062fb 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state.rs @@ -15,6 +15,22 @@ 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(Arc>), +} + /// The lifecycle phase a capsule is currently executing in. /// /// Set on [`HostState`] during `#[install]` or `#[upgrade]` dispatch. @@ -214,9 +230,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..aa07e3eb8 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. @@ -555,6 +569,29 @@ 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; } /// HTTP client operations with SSRF protection. From 61b138893c4e2985a508b51643bf14e3da962017 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 19 May 2026 15:07:38 +0400 Subject: [PATCH 2/5] feat(net): full std::net::TcpStream surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements 14 new host fns to give capsules complete std::net::TcpStream parity. All are kernel-side; the SDK wraps them in unicity-astrid/sdk-rust. New fns on net::Host: - net_read_bytes / net_write_bytes: byte-stream read/write (no length-prefix framing). Read returns empty Vec on EOF; timeout surfaces as 'read would block' / 'write would block' to match std::io::ErrorKind::WouldBlock. - net_peek: peek without consuming - net_shutdown(read|write|both): half-close (write supported; read/both return clear errors pending socket2 integration) - net_peer_addr / net_local_addr: address introspection - net_set_nodelay / net_nodelay: TCP_NODELAY toggle - net_set_read_timeout / net_read_timeout (stored per-stream) - net_set_write_timeout / net_write_timeout (stored per-stream) - net_set_ttl / net_ttl: IP TTL Refactor: - NetStream::Tcp now holds TcpStreamSlot { stream, read_timeout, write_timeout } instead of a bare Arc>. Per-stream timeout state lives next to the socket so the byte-stream read/write fns can apply tokio::time::timeout per call without global state. - net.rs split into net/ submodule (handshake.rs + stream.rs + mod.rs) to stay under the 1000-line CI cap. Unix variants reject TCP-only ops (nodelay, ttl, addresses, timeouts) with clear errors — those are TCP socket options that don't apply. RFC: unicity-astrid/rfcs#27 WIT: unicity-astrid/wit#5 Tracking issue: closes #745 --- .../src/engine/wasm/host/net/handshake.rs | 159 ++++++ .../engine/wasm/host/{net.rs => net/mod.rs} | 505 +++++++++--------- .../src/engine/wasm/host/net/stream.rs | 168 ++++++ .../src/engine/wasm/host_state.rs | 25 +- wit/astrid-capsule.wit | 109 +++- 5 files changed, 721 insertions(+), 245 deletions(-) create mode 100644 crates/astrid-capsule/src/engine/wasm/host/net/handshake.rs rename crates/astrid-capsule/src/engine/wasm/host/{net.rs => net/mod.rs} (64%) create mode 100644 crates/astrid-capsule/src/engine/wasm/host/net/stream.rs 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.rs b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs similarity index 64% rename from crates/astrid-capsule/src/engine/wasm/host/net.rs rename to crates/astrid-capsule/src/engine/wasm/host/net/mod.rs index f06ae0eaf..d2203a213 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -1,250 +1,21 @@ -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::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}; +use crate::engine::wasm::host_state::{HostState, NetStream, TcpStreamSlot}; -/// 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. -const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +mod handshake; +mod stream; + +use handshake::{validate_handshake, verify_peer_credentials}; +use stream::{ + CONNECT_TIMEOUT, read_bytes_inner, read_frame, with_tcp_slot, write_bytes_inner, write_frame, +}; /// 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 - ) -} - -/// Read one length-prefixed frame from `stream`. Shared by UnixStream and -/// TcpStream paths — both implement [`tokio::io::AsyncRead`]. -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`]. -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(()) -} - -// --------------------------------------------------------------------------- -// 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). /// @@ -522,8 +293,8 @@ impl net::Host for HostState { let mut s = arc.lock().await; read_frame(&mut *s).await }, - NetStream::Tcp(arc) => { - let mut s = arc.lock().await; + NetStream::Tcp(slot) => { + let mut s = slot.stream.lock().await; read_frame(&mut *s).await }, } @@ -558,8 +329,8 @@ impl net::Host for HostState { let mut s = arc.lock().await; write_frame(&mut *s, &data).await }, - NetStream::Tcp(arc) => { - let mut s = arc.lock().await; + NetStream::Tcp(slot) => { + let mut s = slot.stream.lock().await; write_frame(&mut *s, &data).await }, } @@ -667,7 +438,11 @@ impl net::Host for HostState { ); self.active_streams.insert( handle_id, - NetStream::Tcp(std::sync::Arc::new(tokio::sync::Mutex::new(stream))), + NetStream::Tcp(TcpStreamSlot { + stream: std::sync::Arc::new(tokio::sync::Mutex::new(stream)), + read_timeout: None, + write_timeout: None, + }), ); Ok(handle_id) } @@ -685,6 +460,250 @@ impl net::Host for HostState { 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(); + let max = max_bytes as usize; + 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 tokio::io::AsyncWriteExt; + let mut s = slot.stream.lock().await; + match how { + // Tokio TcpStream only exposes shutdown for the write + // side via the AsyncWrite trait. For Read / Both we + // need the inner std-level shutdown via std_into. + ShutdownHow::Write => s + .shutdown() + .await + .map_err(|e| format!("shutdown(write): {e}")), + _ => { + let std = s.as_ref(); + // No-op for unsupported direction on the read + // side of tokio's TcpStream — std::Shutdown::Read + // / Both delegate to the underlying socket via + // socket2 in a future iteration. For now we + // accept the documented limitation: only + // `shutdown-how::write` is fully supported. + let _ = std; + let _ = std_how; + Err(format!( + "shutdown({:?}): only `write` half-close is supported in this version", + how + )) + }, + } + }, + 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)] 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..93379ba5b --- /dev/null +++ b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs @@ -0,0 +1,168 @@ +//! 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); + +/// 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. +/// Empty result means EOF (peer disconnected); timeout returns the +/// `would block` error so callers can distinguish "no data yet" from +/// "stream closed" — matching `std::io::ErrorKind::WouldBlock`. +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 mut buf = vec![0u8; max_bytes]; + let effective = timeout.unwrap_or(std::time::Duration::from_millis(50)); + match tokio::time::timeout(effective, stream.read(&mut buf)).await { + 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(_) if timeout.is_none() => Ok(Vec::new()), + 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 26fc062fb..20838a91f 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state.rs @@ -28,7 +28,30 @@ pub enum NetStream { /// Inbound Unix-domain socket accepted from the kernel's listener. Unix(Arc>), /// Outbound TCP connection opened via `net.connect-tcp`. - Tcp(Arc>), + 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. diff --git a/wit/astrid-capsule.wit b/wit/astrid-capsule.wit index aa07e3eb8..05db39502 100644 --- a/wit/astrid-capsule.wit +++ b/wit/astrid-capsule.wit @@ -352,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, + } } // --------------------------------------------------------------------------- @@ -531,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. /// @@ -592,6 +606,99 @@ interface net { /// - 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. From ca5fb5e6f2d504add1e3c85b5503e2cb226eb5cb Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 19 May 2026 15:15:28 +0400 Subject: [PATCH 3/5] fix(net): full Shutdown::{Read,Write,Both} via socket2::SockRef MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the temporary 'only write half-close supported' error with the real impl. tokio::net::TcpStream only exposes the write half via the AsyncWrite trait, but socket2::SockRef::from(&stream) wraps the borrowed FD and exposes shutdown(std::net::Shutdown) for all three directions — the same OS shutdown(2) syscall std::net::TcpStream uses internally. No FD ownership transfer; the SockRef borrows the tokio stream so subsequent reads/writes on the slot still work (within whatever half-closed state the shutdown produced). Added a unit test that exercises Read/Write/Both against a real loopback TcpStream — guards against a future tokio or socket2 bump breaking the glue. socket2 was already a transitive dep via tokio; promoted to a direct dep on astrid-capsule. --- crates/astrid-capsule/Cargo.toml | 1 + .../src/engine/wasm/host/net/mod.rs | 62 +++++++++++-------- 2 files changed, 37 insertions(+), 26 deletions(-) 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/wasm/host/net/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs index d2203a213..850c074a4 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -570,32 +570,18 @@ impl net::Host for HostState { let result = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move { match stream { NetStream::Tcp(slot) => { - use tokio::io::AsyncWriteExt; - let mut s = slot.stream.lock().await; - match how { - // Tokio TcpStream only exposes shutdown for the write - // side via the AsyncWrite trait. For Read / Both we - // need the inner std-level shutdown via std_into. - ShutdownHow::Write => s - .shutdown() - .await - .map_err(|e| format!("shutdown(write): {e}")), - _ => { - let std = s.as_ref(); - // No-op for unsupported direction on the read - // side of tokio's TcpStream — std::Shutdown::Read - // / Both delegate to the underlying socket via - // socket2 in a future iteration. For now we - // accept the documented limitation: only - // `shutdown-how::write` is fully supported. - let _ = std; - let _ = std_how; - Err(format!( - "shutdown({:?}): only `write` half-close is supported in this version", - how - )) - }, - } + // 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()), } @@ -716,4 +702,28 @@ mod tests { // (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); + } } From e12e60b43c45f7f75722d4e86961bbbeeee34fd4 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 19 May 2026 16:03:38 +0400 Subject: [PATCH 4/5] fix(net): unambiguous EOF in net_read_bytes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Gemini code-review feedback on unicity-astrid/sdk-rust#42. Before: - timeout=None falls through to a 50ms internal timeout. - Timeout expiry with no data returned Ok(empty Vec). - Caller can't distinguish 'transient no data' from 'peer disconnected'. After: - timeout=None drops the tokio::time::timeout wrapper entirely. Reads block until data, EOF, or capsule unload (via the outer bounded_block_on_cancellable cancellation token). - Empty Vec is now ALWAYS EOF — matches std::io::Read's Ok(0) contract. - 'read would block' error only surfaces when the caller has explicitly set a read timeout that expired. SDK-side, that maps to std::io::ErrorKind::WouldBlock so std::Read callers can distinguish. The peek and write_bytes paths already had the right shape; only net_read_bytes carried the bug. --- .../src/engine/wasm/host/net/stream.rs | 25 ++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs index 93379ba5b..73493d9c2 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs @@ -85,9 +85,17 @@ where } /// Read up to `max_bytes` from `stream` without length-prefix framing. -/// Empty result means EOF (peer disconnected); timeout returns the -/// `would block` error so callers can distinguish "no data yet" from -/// "stream closed" — matching `std::io::ErrorKind::WouldBlock`. +/// +/// 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, @@ -98,8 +106,14 @@ where { use tokio::io::AsyncReadExt; let mut buf = vec![0u8; max_bytes]; - let effective = timeout.unwrap_or(std::time::Duration::from_millis(50)); - match tokio::time::timeout(effective, stream.read(&mut buf)).await { + // 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); @@ -107,7 +121,6 @@ where }, Ok(Err(e)) if is_peer_disconnect(&e) => Ok(Vec::new()), Ok(Err(e)) => Err(format!("read error: {e}")), - Err(_) if timeout.is_none() => Ok(Vec::new()), Err(_) => Err("read would block".to_string()), } } From 0ba2a4aa993b364641098d1a42d0b8a3afb9221b Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 19 May 2026 16:36:08 +0400 Subject: [PATCH 5/5] fix(net): security review on #746 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Gemini code review feedback on unicity-astrid/astrid#746. OOM cap on byte-stream reads (security-critical): - Guest-supplied max_bytes was used directly as the buffer size in read_bytes_inner and net_peek. u32::MAX would allocate 4 GB on the host. New shared constant MAX_BYTES_PER_CALL = 10 MB (matches the framed net-read payload cap) is applied as .min(max_bytes_usize) at both sites. - 10 MB is well above any realistic single-call read for the targeted protocols: TLS records cap at 16 KB, WebSocket frames typically ≤64 KB, MQTT control packets stream in chunks. Host string validation (security-medium): - net_connect_tcp now rejects empty hosts, hosts >255 bytes (RFC 1035 max-name-length), and hosts containing null bytes — at the host-fn boundary, before the resolver / audit log / tracing spans see the input. New validate_host helper + 5 unit tests covering each gate. Already addressed by the prior commit (e12e60b): - read_bytes_inner EOF-vs-timeout ambiguity. timeout=None now blocks indefinitely instead of falling through to a 50ms internal poll; empty Vec is unambiguously EOF. Not addressed in this PR: - Mutex-across-await architectural concern (full-duplex on one handle). Splitting the stream into read/write halves with independent locks would touch every method dispatcher and the HostState layout. Worth doing but distinct from the TCP surface this PR ships — filed for a follow-up issue. --- .../src/engine/wasm/host/net/mod.rs | 63 ++++++++++++++++++- .../src/engine/wasm/host/net/stream.rs | 11 ++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs index 850c074a4..208eac48a 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -9,9 +9,32 @@ mod stream; use handshake::{validate_handshake, verify_peer_credentials}; use stream::{ - CONNECT_TIMEOUT, read_bytes_inner, read_frame, with_tcp_slot, write_bytes_inner, write_frame, + 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; @@ -349,6 +372,9 @@ impl net::Host for HostState { } 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 { @@ -528,7 +554,9 @@ impl net::Host for HostState { let rt = self.runtime_handle.clone(); let sem = self.host_semaphore.clone(); let tok = self.cancel_token.clone(); - let max = max_bytes as usize; + // 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; @@ -726,4 +754,35 @@ mod tests { 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 index 73493d9c2..abc4f1991 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs @@ -14,6 +14,16 @@ use crate::engine::wasm::host_state::{HostState, NetStream}; /// 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 { @@ -105,6 +115,7 @@ 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