From fb83e0c067605aac66edbb2a691921b9b8f0dc0b Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 22:49:27 +0000 Subject: [PATCH] WIP T29: credential hook, pinned HTTP transport and guest networking --- Cargo.lock | 35 ++- host/Cargo.toml | 9 + host/src/exec.rs | 5 +- host/src/harness/environment.rs | 2 +- host/src/harness/mod.rs | 33 ++- host/src/process/http.rs | 385 ++++++++++++++++++++++++++++++++ host/src/process/mod.rs | 108 ++++++++- host/src/process/net.rs | 85 +++++++ host/src/process/services.rs | 2 +- host/tests/hook.rs | 332 +++++++++++++++++++++++++++ host/tests/hook_git.rs | 208 +++++++++++++++++ host/tests/hook_guest.rs | 64 ++++++ host/tests/net.rs | 125 +++++++++++ host/tests/process.rs | 2 +- 14 files changed, 1364 insertions(+), 31 deletions(-) create mode 100644 host/src/process/http.rs create mode 100644 host/src/process/net.rs create mode 100644 host/tests/hook.rs create mode 100644 host/tests/hook_git.rs create mode 100644 host/tests/hook_guest.rs create mode 100644 host/tests/net.rs diff --git a/Cargo.lock b/Cargo.lock index 7d50be6..81c69c0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3245,6 +3245,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "if-addrs" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "if_chain" version = "1.0.3" @@ -4995,6 +5005,7 @@ dependencies = [ "anyhow", "async-trait", "base64 0.22.1", + "bytes", "chacha20poly1305", "clap", "deno_ast", @@ -5012,6 +5023,11 @@ dependencies = [ "hkdf", "hmac", "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "if-addrs", "openssl", "p256", "rsa", @@ -5027,6 +5043,8 @@ dependencies = [ "tempfile", "thiserror 2.0.21", "tokio", + "tokio-rustls", + "toml", "url", "uuid", "wasmtime", @@ -5035,6 +5053,7 @@ dependencies = [ "wat", "webauthn-authenticator-rs", "webauthn-rs", + "webpki-roots", "zeroize", ] @@ -5213,9 +5232,9 @@ dependencies = [ [[package]] name = "serde_spanned" -version = "1.1.2" +version = "1.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c" +checksum = "9c187e0a369471c550af028cba56d4f4fa2a489ad28151ae95a4d340a2a8b782" dependencies = [ "serde_core", ] @@ -6391,9 +6410,9 @@ dependencies = [ [[package]] name = "toml" -version = "1.1.8+spec-1.1.0" +version = "1.2.1+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20489e00e4d8741d6be680764cc12e270655e375a20d1011e844a9c3379e678d" +checksum = "d5755c51863cdac711f1956baf8eb9588e060755f7d3c4c979e70ea1e1f32cd7" dependencies = [ "indexmap", "serde_core", @@ -6406,9 +6425,9 @@ dependencies = [ [[package]] name = "toml_datetime" -version = "1.1.2+spec-1.1.0" +version = "1.1.3+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b86d767906c6c42421dcba507eb9d203e779497710a47782a224bb871653053" +checksum = "173f25c883cd2176a2d8212155638c0a60d484a9d84546d270866f40f56b6e49" dependencies = [ "serde_core", ] @@ -6436,9 +6455,9 @@ dependencies = [ [[package]] name = "toml_writer" -version = "1.1.3+spec-1.1.0" +version = "1.2.0+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770" +checksum = "009dbb7700f11c55ef83aebbe4c0e9cf1f1ac0844bd8d6f00dc38f302026e13a" [[package]] name = "tower" diff --git a/host/Cargo.toml b/host/Cargo.toml index 88863a7..f495fc2 100644 --- a/host/Cargo.toml +++ b/host/Cargo.toml @@ -51,6 +51,15 @@ ssh-key = { version = "0.6.7", features = ["encryption", "ed25519", "ecdsa", "rs uuid = { version = "1.29.0", features = ["v4"] } webauthn-rs = "0.5" openssl = { version = "0.10.82", features = ["vendored"] } +hyper-util = { version = "0.1", features = ["tokio"] } +tokio-rustls = "0.26.6" +webpki-roots = "1.0.11" +if-addrs = "0.15.0" +toml = "1.2.1" +bytes = "1.12.1" +http-body = "1.1.0" +http-body-util = "0.1.5" +hyper = { version = "1.12.0", features = ["client", "http1", "server"] } [dev-dependencies] tempfile = "3" diff --git a/host/src/exec.rs b/host/src/exec.rs index cce79ec..21870ef 100644 --- a/host/src/exec.rs +++ b/host/src/exec.rs @@ -10,14 +10,13 @@ pub enum Command { } /// How to run it. `cwd` is a world path. `timeout` ends the command and all its processes. -/// `conversation` is the harness conversation that runs the command. Its processes and their -/// descendants carry it, so the host knows the caller (for example in the `wasi:http` hook). +/// `conversations` carries the nearest-first conversation chain. Child processes inherit it. #[derive(Clone, Debug, Default)] pub struct Options { pub cwd: String, pub env: Vec<(String, String)>, pub timeout: Option, - pub conversation: Option, + pub conversations: Vec, } /// The stream that one output chunk came from. diff --git a/host/src/harness/environment.rs b/host/src/harness/environment.rs index 8ed9ef0..d732cbc 100644 --- a/host/src/harness/environment.rs +++ b/host/src/harness/environment.rs @@ -88,7 +88,7 @@ pub fn op_exec( }; let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); let task = tokio::spawn(async move { - let result = exec.exec(command, Options { cwd, conversation, timeout: timeout.map(std::time::Duration::from_secs_f64), ..Default::default() }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await; + let result = exec.exec(command, Options { cwd, conversations: conversation.into_iter().collect(), timeout: timeout.map(std::time::Duration::from_secs_f64), ..Default::default() }, &mut |stream, bytes| { let _ = sender.send(json!({"stream": if stream == Stream::Stdout {"stdout"} else {"stderr"}, "bytes": bytes})); }).await; let _ = sender.send(match result { Ok(exit) => json!({"exitCode": exit}), Err(e) => json!({"error": e.to_string()}), diff --git a/host/src/harness/mod.rs b/host/src/harness/mod.rs index 246bb37..a70c774 100644 --- a/host/src/harness/mod.rs +++ b/host/src/harness/mod.rs @@ -92,12 +92,25 @@ deno_core::extension!( /// Start the console with the process implementation supplied by the host. pub async fn run_with_exec(world: World, exec: Arc) -> anyhow::Result<()> { + let cred = Arc::new(crate::cred::Cred::open(&world.host_dir)?); + run_with_shared_credentials(world, exec, cred).await +} +async fn run_with_shared_credentials( + world: World, + exec: Arc, + cred: Arc, +) -> anyhow::Result<()> { // Deno's unsynchronized operations require a current-thread Tokio runtime. tokio::task::spawn_blocking(move || { tokio::runtime::Builder::new_current_thread() .enable_all() .build()? - .block_on(run_module(world, exec, include_str!("../../js/bundle.js"))) + .block_on(run_module_with_credentials( + world, + exec, + cred, + include_str!("../../js/bundle.js"), + )) }) .await? .map_err(|error: anyhow::Error| { @@ -232,13 +245,19 @@ pub async fn run(path: &Path) -> anyhow::Result<()> { /// Start the console with the shared process table and an optional userland bundle. pub async fn run_with_userland(path: &Path, userland: Option<&Path>) -> anyhow::Result<()> { let world = World::open_with_userland(path, userland)?; - let runtime = crate::process::Runtime::new(World { - root: world.root.clone(), - host_dir: world.host_dir.clone(), - processes: world.processes.clone(), - })?; + let cred = Arc::new(crate::cred::Cred::open(&world.host_dir)?); + let runtime = crate::process::Runtime::new_with_hooks( + World { + root: world.root.clone(), + host_dir: world.host_dir.clone(), + processes: world.processes.clone(), + }, + cred.clone(), + crate::process::default_script(), + None, + )?; let services = crate::process::supervise(runtime.clone()); - let result = run_with_exec(world, Arc::new(runtime)).await; + let result = run_with_shared_credentials(world, Arc::new(runtime), cred).await; services.abort(); let _ = services.await; result diff --git a/host/src/process/http.rs b/host/src/process/http.rs new file mode 100644 index 0000000..2c3e9cd --- /dev/null +++ b/host/src/process/http.rs @@ -0,0 +1,385 @@ +//! Per-process credential hook and pinned-address HTTP transport. +//! Transport derives from Wasmtime 49's default sender, with a policy check before connect. +use crate::cred::{Caller, Cred, inject::Outcome, script::Script}; +use http_body_util::{BodyExt, Full}; +use std::sync::Arc; +use wasmtime_wasi_http::{Error, RequestOptions}; +use wasmtime_wasi_http::{WasiBody, WasiHttpHooks}; +type Io = Box> + Send>; +pub type Router = fn(&Caller, Request) -> Response; +pub(super) struct Hooks { + pub caller: Caller, + pub cred: Arc, + pub script: Arc, + pub policy: Arc, + pub router: Router, +} +pub fn whoami(caller: &Caller, request: Request) -> Response { + let (status, body) = if request.uri().path() == "/whoami" { + (200, serde_json::to_vec(caller).unwrap()) + } else { + ( + 404, + serde_json::to_vec(&serde_json::json!({"error":"unknown host API route"})).unwrap(), + ) + }; + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Bytes::from(body)) + .unwrap() +} +fn response(response: Response) -> (Response, Io) { + ( + response.map(|b| Full::new(b).map_err(|n| match n {}).boxed_unsync()), + Box::new(async { Ok(()) }), + ) +} +impl WasiHttpHooks for Hooks { + // Admit every scheme until the URI supplies the host name. + fn is_supported_scheme(&mut self, _: &http::uri::Scheme) -> bool { true } + fn send_request( + &mut self, + request: Request, + options: Option, + fut: Io, + ) -> Box, Io), Error>> + Send> { + drop(fut); + let caller = self.caller.clone(); + let cred = self.cred.clone(); + let script = self.script.clone(); + let policy = self.policy.clone(); + let router = self.router; + Box::new(async move { + let (parts, mut body) = request.into_parts(); + if parts + .uri + .host() + .is_some_and(|h| h.eq_ignore_ascii_case("host.sausage")) + { + let bytes = body.collect().await?.to_bytes(); + return Ok(response(router(&caller, Request::from_parts(parts, bytes)))); + } + if !matches!(parts.uri.scheme_str(), Some("http" | "https")) { + while let Some(frame) = body.frame().await { frame?; } + return Err(Error::HttpRequestUriInvalid); + } + let result = + tokio::task::spawn_blocking(move || cred.inject(parts, &caller, script.as_ref())) + .await; + let mut parts = match result { + Ok(Ok(Outcome::Forward(parts))) => parts, + outcome => { + // Drain without retaining uploads, so the guest can finish its output stream. + while let Some(frame) = body.frame().await { + frame?; + } + let (status, value) = match outcome { + Ok(Ok(Outcome::Denied { item, request })) => ( + 403, + serde_json::json!({"message": format!("Access to {item} needs approval. Request {request} awaits a decision."), "item": item, "request": request}), + ), + Ok(Err(e)) => (502, serde_json::json!({"error": e.to_string()})), + _ => ( + 502, + serde_json::json!({"error": "credential worker failed"}), + ), + }; + return Ok(response( + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Bytes::from(serde_json::to_vec(&value).unwrap())) + .unwrap(), + )); + } + }; + let authority = parts.uri.authority().ok_or(Error::HttpRequestUriInvalid)?; + if authority.as_str().contains('@') { + return Err(Error::HttpRequestUriInvalid); + } + parts.headers.insert( + http::header::HOST, + authority + .as_str() + .parse() + .map_err(|_| Error::HttpRequestUriInvalid)?, + ); + for value in parts.headers.values_mut() { + value.set_sensitive(true); + } + // Hyper otherwise discards unknown-length GET bodies before WASI finishes them. + if !parts.headers.contains_key(http::header::CONTENT_LENGTH) { + parts.headers.insert( + http::header::TRANSFER_ENCODING, + http::HeaderValue::from_static("chunked"), + ); + } + let (res, io) = + send_request(Request::from_parts(parts, body), options, &policy).await?; + Ok((res.map(BodyExt::boxed_unsync), Box::new(io) as Io)) + }) + } +} +use bytes::Bytes; +use core::future::poll_fn; +use core::pin::{Pin, pin}; +use core::task::{Poll, ready}; +use http::uri::Scheme; +use http::{Request, Response}; +use http_body::Body; +use std::time::Duration; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::net::TcpStream; + +trait TokioStream: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static { + fn boxed(self) -> Box + where + Self: Sized, + { + Box::new(self) + } +} +impl TokioStream for T where T: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static {} + +/// Preserve Wasmtime's HTTP/1 transport and request deadlines, but connect only to checked addresses. +/// The returned future drives the connection after the response headers arrive. +async fn send_request( + mut req: Request + Send + 'static>, + options: Option, + policy: &super::net::NetPolicy, +) -> Result< + ( + Response>, + impl Future> + Send, + ), + Error, +> { + let uri = req.uri().clone(); + let authority = uri.authority().ok_or(Error::HttpRequestUriInvalid)?; + let use_tls = uri.scheme() == Some(&Scheme::HTTPS); + let authority = if authority.port().is_some() { + authority.to_string() + } else { + let port = if use_tls { 443 } else { 80 }; + format!("{authority}:{port}") + }; + + let connect_timeout = options + .and_then(|r| r.connect_timeout) + .unwrap_or(Duration::from_secs(600)); + + let first_byte_timeout = options + .and_then(|r| r.first_byte_timeout) + .unwrap_or(Duration::from_secs(600)); + + let between_bytes_timeout = options + .and_then(|r| r.between_bytes_timeout) + .unwrap_or(Duration::from_secs(600)); + + let stream = match tokio::time::timeout(connect_timeout, async { + let addresses = policy.resolve(&uri).await?; + TcpStream::connect(addresses.as_slice()) + .await + .map_err(Error::Connect) + }) + .await + { + Ok(stream) => stream?, + Err(..) => return Err(Error::ConnectionTimeout), + }; + let stream = if use_tls { + // derived from https://github.com/rustls/rustls/blob/main/examples/src/bin/simpleclient.rs + let root_cert_store = rustls::RootCertStore { + roots: webpki_roots::TLS_SERVER_ROOTS.into(), + }; + let config = rustls::ClientConfig::builder() + .with_root_certificates(root_cert_store) + .with_no_client_auth(); + let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(config)); + let domain = tls_server_name(&authority)?; + let stream = connector + .connect(domain, stream) + .await + .map_err(Error::Tls)?; + stream.boxed() + } else { + stream.boxed() + }; + let (mut sender, conn) = tokio::time::timeout( + connect_timeout, + hyper::client::conn::http1::Builder::new().handshake(hyper_util::rt::TokioIo::new(stream)), + ) + .await + .map_err(|_| Error::ConnectionTimeout)??; + + // Send origin-form URIs, because this connection does not use a proxy. + *req.uri_mut() = http::Uri::builder() + .path_and_query( + req.uri() + .path_and_query() + .map(|p| p.as_str()) + .unwrap_or("/"), + ) + .build() + .expect("comes from valid request"); + + let send = async move { + use core::task::Context; + + /// Wrapper around [hyper::body::Incoming] used to + /// account for request option timeout configuration + struct IncomingResponseBody { + incoming: hyper::body::Incoming, + timeout: tokio::time::Interval, + } + impl http_body::Body for IncomingResponseBody { + type Data = ::Data; + type Error = Error; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + match Pin::new(&mut self.as_mut().incoming).poll_frame(cx) { + Poll::Ready(None) => Poll::Ready(None), + Poll::Ready(Some(Err(err))) => { + let err = if err.is_timeout() { + Error::HttpResponseTimeout + } else { + Error::from(err) + }; + Poll::Ready(Some(Err(err))) + } + Poll::Ready(Some(Ok(frame))) => { + self.timeout.reset(); + Poll::Ready(Some(Ok(frame))) + } + Poll::Pending => { + ready!(self.timeout.poll_tick(cx)); + Poll::Ready(Some(Err(Error::ConnectionReadTimeout))) + } + } + } + fn is_end_stream(&self) -> bool { + self.incoming.is_end_stream() + } + fn size_hint(&self) -> http_body::SizeHint { + self.incoming.size_hint() + } + } + + let res = tokio::time::timeout(first_byte_timeout, sender.send_request(req)) + .await + .map_err(|_| Error::ConnectionReadTimeout)? + .map_err(Error::from)?; + // Tokio panics on an `interval` with a zero duration, so clamp it to a + // 1ns timeout which is sort of effectively zero anyway. + let between_bytes_timeout = if between_bytes_timeout == Duration::ZERO { + Duration::new(0, 1) + } else { + between_bytes_timeout + }; + let mut timeout = tokio::time::interval(between_bytes_timeout); + timeout.reset(); + Ok(res.map(|incoming| IncomingResponseBody { incoming, timeout })) + }; + let mut send = pin!(send); + let mut conn = Some(conn); + // Wait for response while driving connection I/O + let res = poll_fn(|cx| match send.as_mut().poll(cx) { + Poll::Ready(Ok(res)) => Poll::Ready(Ok(res)), + Poll::Ready(Err(err)) => Poll::Ready(Err(err)), + Poll::Pending => { + // Response is not ready, poll `hyper` connection to drive I/O if it has not completed yet + let Some(fut) = conn.as_mut() else { + // `hyper` connection already completed + return Poll::Pending; + }; + let res = ready!(Pin::new(fut).poll(cx)); + // `hyper` connection completed, record that to prevent repeated poll + conn = None; + match res { + // `hyper` connection has successfully completed, optimistically poll for response + Ok(()) => send.as_mut().poll(cx), + // `hyper` connection has failed, return the error + Err(err) => Poll::Ready(Err(Error::from(err))), + } + } + }) + .await?; + Ok((res, async move { + let Some(conn) = conn.take() else { + // `hyper` connection has already completed + return Ok(()); + }; + if let Err(err) = conn.await { + if err.is_timeout() { + return Err(Error::HttpResponseTimeout); + } + return Err(err.into()); + } + Ok(()) + })) +} + +/// Resolve the rustls [`ServerName`] used for TLS certificate verification from +/// an outbound request `authority`. +/// +/// `authority` is in `host:port` form, where an IPv6 `host` is wrapped in +/// brackets (for example `[::1]:443`). An IP literal is recognized by parsing +/// the whole authority as a [`SocketAddr`]; this handles the bracketed IPv6 +/// form, which splitting on the first `:` would truncate. Anything else is +/// treated as a host name, with the port stripped off before it is handed to +/// rustls. +/// +/// [`ServerName`]: rustls::pki_types::ServerName +/// [`SocketAddr`]: std::net::SocketAddr +pub(crate) fn tls_server_name( + authority: &str, +) -> Result, rustls::pki_types::InvalidDnsNameError> { + use rustls::pki_types::ServerName; + + if let Ok(addr) = authority.parse::() { + return Ok(ServerName::from(addr.ip())); + } + let host = match authority.split_once(':') { + Some((host, _port)) => host, + None => authority, + }; + Ok(ServerName::try_from(host)?.to_owned()) +} + +#[cfg(test)] +mod tls_server_name_tests { + use super::tls_server_name; + use rustls::pki_types::ServerName; + + #[test] + fn resolves_server_name_from_authority() { + // Host names keep their host and drop the port. + assert_eq!( + tls_server_name("example.com:443").unwrap(), + ServerName::try_from("example.com").unwrap() + ); + assert_eq!( + tls_server_name("example.com").unwrap(), + ServerName::try_from("example.com").unwrap() + ); + + // IP literals resolve to an `IpAddress` server name. The bracketed IPv6 + // form must not be truncated at the first `:`. + assert_eq!( + tls_server_name("127.0.0.1:80").unwrap(), + ServerName::from(std::net::Ipv4Addr::LOCALHOST) + ); + assert_eq!( + tls_server_name("[::1]:443").unwrap(), + ServerName::from(std::net::Ipv6Addr::LOCALHOST) + ); + assert_eq!( + tls_server_name("[2001:db8::1]:8443").unwrap(), + ServerName::from("2001:db8::1".parse::().unwrap()) + ); + } +} diff --git a/host/src/process/mod.rs b/host/src/process/mod.rs index 2db6095..c995588 100644 --- a/host/src/process/mod.rs +++ b/host/src/process/mod.rs @@ -46,6 +46,12 @@ mod bindings { } use bindings::sausage::process::host::{Host, HostChild, Request, Stdio}; mod cache; +mod http; +pub mod net; +/// Temporary default until T28 supplies the bounded QuickJS evaluator. +pub fn default_script() -> Arc { + Arc::new(crate::cred::v8::V8) +} pub use cache::configure as configure_cache; mod services; mod stdio; @@ -84,6 +90,10 @@ pub struct Runtime { entries: Mutex>, next: AtomicU32, alive: Arc, + cred: Arc, + script: Arc, + router: http::Router, + pub network: Arc, } impl Drop for Runtime { fn drop(&mut self) { @@ -98,17 +108,18 @@ struct State { pid: u32, command: u32, group: u32, - conversation: Option, + conversations: Vec, stdio: Io, preview1: Option, http: wasmtime_wasi_http::WasiHttpCtx, + hooks: http::Hooks, } impl wasmtime_wasi_http::WasiHttpView for State { fn http(&mut self) -> wasmtime_wasi_http::WasiHttpCtxView<'_> { wasmtime_wasi_http::WasiHttpCtxView { ctx: &mut self.http, table: &mut self.table, - hooks: Default::default(), + hooks: &mut self.hooks, } } } @@ -148,6 +159,37 @@ impl Runtime { Self::with_compiler(world, cache::Compiler::new(directory)?) } fn with_compiler(world: World, compiler: Arc) -> anyhow::Result> { + let cred = Arc::new(crate::cred::Cred::open(&world.host_dir)?); + Self::with_hooks(world, compiler, cred, default_script(), http::whoami) + } + /// Share credentials and the host API router with the host's other subsystems. + pub fn new_with_hooks( + world: World, + cred: Arc, + script: Arc, + router: Option< + fn( + &crate::cred::Caller, + ::http::Request, + ) -> ::http::Response, + >, + ) -> anyhow::Result> { + Self::with_hooks( + world, + cache::shared()?, + cred, + script, + router.unwrap_or(http::whoami), + ) + } + fn with_hooks( + world: World, + compiler: Arc, + cred: Arc, + script: Arc, + router: http::Router, + ) -> anyhow::Result> { + let network = Arc::new(net::NetPolicy::open(&world.host_dir)?); let engine = compiler.engine.clone(); let alive = Arc::new(AtomicBool::new(true)); let ticking = alive.clone(); @@ -161,6 +203,10 @@ impl Runtime { let runtime = Arc::new(Self { world, engine, + cred, + script, + router, + network, alive, compiler, cache: Mutex::new(Default::default()), @@ -200,7 +246,7 @@ impl Runtime { for path in paths { for candidate in [path.clone(), format!("{path}.wasm")] { match self.world.root.stat(&candidate, true).await { - Ok(stat) if !stat.directory => return Ok(candidate), + Ok(stat) if !stat.directory => return self.canonical_program(candidate).await, Ok(_) => {} Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} Err(e) => return Err(e.into()), @@ -209,6 +255,32 @@ impl Runtime { } anyhow::bail!("program not found: {name}") } + async fn canonical_program(&self, mut path: String) -> anyhow::Result { + for _ in 0..40 { + let mut prefix = String::new(); + let components: Vec<_> = path.trim_start_matches('/').split('/').collect(); + let mut target = None; + for (index, component) in components.iter().enumerate() { + prefix.push('/'); + prefix.push_str(component); + if self.world.root.stat(&prefix, false).await?.symlink { + let link = self.world.root.readlink(&prefix).await?; + let parent = prefix.rsplit_once('/').unwrap().0; + let resolved = sausage_vfs::resolve(parent, &link); + target = Some(sausage_vfs::resolve( + &resolved, + &components[index + 1..].join("/"), + )); + break; + } + } + match target { + Some(target) => path = target, + None => return Ok(path), + } + } + anyhow::bail!("too many program symlinks") + } fn kill_command(&self, command: u32) { for entry in self.entries.lock().unwrap().values() { if entry.command == command { @@ -241,7 +313,7 @@ impl Runtime { parent: u32, command: u32, inherited_group: u32, - conversation: Option, + conversations: Vec, inherited_io: Option, ) -> anyhow::Result { anyhow::ensure!(!request.argv.is_empty(), "empty argv"); @@ -326,9 +398,14 @@ impl Runtime { .stdin(stdio.input.clone()) .stdout(AsyncStdoutStream::new(8192, stdio.output.clone())) .stderr(AsyncStdoutStream::new(8192, stdio.error.clone())) - .allow_tcp(false) + .allow_tcp(true) .allow_udp(false) - .allow_ip_name_lookup(false); + .allow_ip_name_lookup(true); + let policy = self.network.clone(); + wasi.socket_addr_check(move |address, _| { + let allowed = policy.allows(address); + Box::pin(async move { allowed }) + }); let mut store = Store::new( &self.engine, State { @@ -339,10 +416,21 @@ impl Runtime { pid, command, group, - conversation: conversation.clone(), + conversations: conversations.clone(), stdio, preview1, http: wasmtime_wasi_http::WasiHttpCtx::new(), + hooks: http::Hooks { + caller: crate::cred::Caller { + pid, + program: executable, + conversations: conversations.clone(), + }, + cred: self.cred.clone(), + script: self.script.clone(), + policy: self.network.clone(), + router: self.router, + }, }, ); store.epoch_deadline_async_yield_and_update(1); @@ -353,7 +441,7 @@ impl Runtime { argv: request.argv, cwd: request.cwd, state: "running".into(), - conversation, + conversation: conversations.first().cloned(), }, ); let (tx, code) = watch::channel(None); @@ -456,7 +544,7 @@ impl Host for State { self.pid, self.command, self.group, - self.conversation.clone(), + self.conversations.clone(), Some(self.stdio.clone()), ) .await @@ -594,7 +682,7 @@ impl Exec for Arc { 0, 0, 0, - options.conversation, + options.conversations, None, ); let mut child = if let Some(deadline) = deadline { diff --git a/host/src/process/net.rs b/host/src/process/net.rs new file mode 100644 index 0000000..52fa22b --- /dev/null +++ b/host/src/process/net.rs @@ -0,0 +1,85 @@ +//! Host-side network policy. `/host/network.toml` accepts `deny = ["address:port"]`. +//! Host listeners register their ports with `deny_host_port`. Guests cannot bind or connect there. +use std::{ + collections::HashSet, + net::{IpAddr, SocketAddr}, + path::Path, + sync::RwLock, +}; + +fn normalize(address: SocketAddr) -> SocketAddr { + match address { + SocketAddr::V6(v6) => v6 + .ip() + .to_ipv4_mapped() + .map(|ip| SocketAddr::new(ip.into(), v6.port())) + .unwrap_or(address), + _ => address, + } +} +pub struct NetPolicy { + denied: HashSet, + local: HashSet, + ports: RwLock>, +} +impl NetPolicy { + pub fn open(host_dir: &Path) -> anyhow::Result { + #[derive(serde::Deserialize, Default)] + #[serde(deny_unknown_fields)] + struct Config { + #[serde(default)] + deny: Vec, + } + let config = match std::fs::read_to_string(host_dir.join("network.toml")) { + Ok(text) => toml::from_str::(&text)?, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Config::default(), + Err(e) => return Err(e.into()), + }; + Ok(Self { + denied: config.deny.into_iter().map(normalize).collect(), + local: if_addrs::get_if_addrs()? + .into_iter() + .map(|a| a.ip()) + .collect(), + ports: RwLock::default(), + }) + } + pub fn deny_host_port(&self, port: u16) { + self.ports.write().unwrap().insert(port); + } + pub fn allows(&self, address: SocketAddr) -> bool { + let address = normalize(address); + let ip = address.ip(); + !self.denied.contains(&address) + && !(self.ports.read().unwrap().contains(&address.port()) + && (ip.is_loopback() || ip.is_unspecified() || self.local.contains(&ip))) + } + pub(super) async fn resolve( + &self, + uri: &http::Uri, + ) -> wasmtime_wasi_http::Result> { + use wasmtime_wasi_http::Error; + let host = uri + .host() + .ok_or(Error::HttpRequestUriInvalid)? + .trim_matches(['[', ']']); + let port = uri + .port_u16() + .unwrap_or(if uri.scheme_str() == Some("https") { + 443 + } else { + 80 + }); + let addresses: Vec<_> = tokio::net::lookup_host((host, port)) + .await + .map_err(|_| Error::DnsError { + rcode: None, + info_code: None, + })? + .collect(); + if addresses.is_empty() || addresses.iter().any(|a| !self.allows(*a)) { + return Err(Error::HttpRequestDenied); + } + Ok(addresses) + } +} diff --git a/host/src/process/services.rs b/host/src/process/services.rs index f61e065..9b814bb 100644 --- a/host/src/process/services.rs +++ b/host/src/process/services.rs @@ -52,7 +52,7 @@ impl Service { 0, 0, 0, - None, + Vec::new(), None, ) .await diff --git a/host/tests/hook.rs b/host/tests/hook.rs new file mode 100644 index 0000000..a25e407 --- /dev/null +++ b/host/tests/hook.rs @@ -0,0 +1,332 @@ +//! Real guest HTTP requests pass through caller identity, credential injection, and streaming transport. +mod hook_git; +mod hook_guest; +use sausage::{ + cred::{ + self, + items::{Field, Item, Rule}, + types::FieldType, + }, + exec::{Command, Exec, Options}, + process::Runtime, + world::World, +}; +use std::sync::Arc; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +struct Fake; +impl cred::script::Script for Fake { + fn evaluate( + &self, + _: &str, + input: &str, + _: Option<&str>, + _: cred::script::Limits, + ) -> Result { + let context: serde_json::Value = serde_json::from_str(input).unwrap(); + Ok(serde_json::json!({"headers":[{"name":"authorization","value":context["credential"]["token"]}]}).to_string()) + } +} +async fn run(runtime: &Arc, url: String, chain: Vec) -> String { + let mut out = Vec::new(); + let code = runtime + .exec( + Command::Argv(vec!["/bin/guest.wasm".into(), url, "GET".into()]), + Options { + cwd: "/".into(), + conversations: chain, + timeout: Some(std::time::Duration::from_secs(30)), + ..Default::default() + }, + &mut |_, b| out.extend_from_slice(b), + ) + .await + .unwrap(); + assert_eq!(code, 0); + String::from_utf8(out).unwrap() +} +#[tokio::test] +async fn identity_injection_denial_and_ambiguity() { + let guest = hook_guest::component(); + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + std::fs::create_dir_all(temp.path().join("world/bin")).unwrap(); + std::fs::copy(guest, temp.path().join("world/bin/guest.wasm")).unwrap(); + let c = Arc::new(cred::Cred::open(&world.host_dir).unwrap()); + let caller = cred::Caller::default(); + let item = c + .create_item( + Item { + id: String::new(), + name: "test item".into(), + version: 0, + free_use: true, + rules: vec![Rule { + scheme: "http".into(), + hosts: vec!["127.0.0.1".into()], + path_prefixes: vec!["/free".into()], + methods: None, + }], + validate_js: None, + transform_js: "inject".into(), + fields: vec![Field { + field_id: String::new(), + name: "token".into(), + public: false, + value: None, + needs_value: true, + field_type: FieldType::Text, + key_type: None, + }], + }, + &caller, + "test", + ) + .unwrap(); + let token = uuid::Uuid::new_v4().to_string(); + c.set_field(&item.id, &item.fields[0].field_id, &token, &caller, "test") + .unwrap(); + let runtime = Runtime::new_with_hooks(world, c.clone(), Arc::new(Fake), None).unwrap(); + let identity = run( + &runtime, + "http://host.sausage/whoami".into(), + vec!["child".into(), "root".into()], + ) + .await; + let who: serde_json::Value = + serde_json::from_str(identity.lines().find(|line| line.starts_with('{')).unwrap()).unwrap(); + assert_eq!(who["pid"], 1); + assert_eq!(who["program"], "/bin/guest.wasm"); + assert_eq!(who["conversations"], serde_json::json!(["child", "root"])); + std::os::unix::fs::symlink("guest.wasm", temp.path().join("world/bin/alias")).unwrap(); + let mut alias_out = Vec::new(); + let code = runtime + .exec( + Command::Argv(vec![ + "/bin/alias".into(), + "custom://host.sausage/whoami".into(), + "GET".into(), + ]), + Options { + cwd: "/".into(), + ..Default::default() + }, + &mut |_, b| alias_out.extend_from_slice(b), + ) + .await + .unwrap(); + assert_eq!(code, 0); + let alias_out = String::from_utf8(alias_out).unwrap(); + assert!(alias_out.contains("/bin/guest.wasm") && !alias_out.contains("/bin/alias")); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server_token = token.clone(); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut bytes = Vec::new(); + loop { + let mut b = [0; 1024]; + let n = socket.read(&mut b).await.unwrap(); + assert!(n > 0); + bytes.extend_from_slice(&b[..n]); + if bytes.ends_with(b"0\r\n\r\n") + || (bytes.ends_with(b"\r\n\r\n") + && !String::from_utf8_lossy(&bytes).contains("transfer-encoding: chunked")) + { + break; + } + } + let text = String::from_utf8(bytes).unwrap(); + assert!( + text.contains(&server_token), + "server did not receive the credential" + ); + socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + .await + .unwrap(); + }); + let url = format!("http://127.0.0.1:{port}/free"); + let out = run(&runtime, url.clone(), vec!["root".into()]).await; + assert!(out.contains("status=200") && !out.contains(&token)); + server.await.unwrap(); + let (redirect_url, redirect) = upstream(Some(token.clone()), 0, 302).await; + assert!( + run(&runtime, format!("{redirect_url}/free"), vec![]) + .await + .contains("status=302") + ); + redirect.await.unwrap(); + let (public_url, public) = upstream(None, 0, 200).await; + assert!( + run( + &runtime, + format!("{}/public", public_url.replace("127.0.0.1", "localhost")), + vec![] + ) + .await + .contains("status=200") + ); + public.await.unwrap(); + let mut changed = item.clone(); + changed.free_use = false; + // A second matching item must fail before any network request. + changed.id = String::new(); + changed.name = "second".into(); + changed.rules[0].path_prefixes = vec!["/free".into(), "/deny".into()]; + let second = c.create_item(changed, &caller, "test").unwrap(); + assert!( + run(&runtime, url.clone(), vec![]) + .await + .contains("ambiguous_item") + ); + let url = format!("http://127.0.0.1:{port}/deny"); + let out = run(&runtime, url.clone(), vec!["other".into()]).await; + assert!(out.contains("status=403") && out.contains("second") && out.contains("request")); + let again = run(&runtime, url, vec!["other".into()]).await; + assert!(again.contains("status=403")); + assert_eq!(c.request_list().unwrap().len(), 1); + c.set_field( + &second.id, + &second.fields[0].field_id, + &token, + &caller, + "test", + ) + .unwrap(); + c.approve( + &c.request_list().unwrap()[0].id, + "allow", + cred::requests::Overrides::default(), + Some("test"), + ) + .unwrap(); + let (url, task) = upstream(Some(token.clone()), 0, 200).await; + let out = run( + &runtime, + format!("{url}/deny"), + vec!["descendant".into(), "other".into()], + ) + .await; + assert!(out.contains("status=200") && !out.contains(&token)); + task.await.unwrap(); + let (child_url, task) = upstream(Some(token.clone()), 0, 200).await; + let mut child_out = Vec::new(); + let code = runtime + .exec( + Command::Argv(vec![ + "/bin/guest.wasm".into(), + "spawn".into(), + format!("{child_url}/deny"), + ]), + Options { + cwd: "/".into(), + conversations: vec!["descendant".into(), "other".into()], + timeout: Some(std::time::Duration::from_secs(30)), + ..Default::default() + }, + &mut |_, b| child_out.extend_from_slice(b), + ) + .await + .unwrap(); + assert_eq!(code, 0); + let child_out = String::from_utf8(child_out).unwrap(); + assert!(child_out.contains("status=200") && !child_out.contains(&token)); + task.await.unwrap(); + let denied = run(&runtime, format!("{url}/deny"), vec!["sibling".into()]).await; + assert!(denied.contains("status=403")); +} + +async fn upstream( + token: Option, + length: usize, + status: u16, +) -> (String, tokio::task::JoinHandle<()>) { + use http_body_util::{BodyExt, Full}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let service = + hyper::service::service_fn(move |request: http::Request| { + let token = token.clone(); + async move { + match token { + Some(token) => assert!( + request + .headers() + .get("authorization") + .is_some_and(|h| h.as_bytes() == token.as_bytes()), + "credential absent or incorrect" + ), + None => assert!(!request.headers().contains_key("authorization")), + } + let mut body = request.into_body(); + let mut received = 0; + while let Some(frame) = body.frame().await { + if let Ok(data) = frame.unwrap().into_data() { + received += data.len(); + } + } + assert_eq!(received, length); + Ok::<_, std::convert::Infallible>( + http::Response::builder() + .status(status) + .header("location", "http://localhost/public") + .body(Full::new(bytes::Bytes::from_static(b"ok"))) + .unwrap(), + ) + } + }); + hyper::server::conn::http1::Builder::new() + .keep_alive(false) + .serve_connection(hyper_util::rt::TokioIo::new(socket), service) + .await + .unwrap(); + }); + (url, task) +} + +#[tokio::test] +async fn streamed_upload_redirect_and_http_admin_denial() { + let guest = hook_guest::component(); + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + std::fs::create_dir_all(temp.path().join("world/bin")).unwrap(); + std::fs::copy(guest, temp.path().join("world/bin/guest.wasm")).unwrap(); + let runtime = Runtime::new(world).unwrap(); + let (url, task) = upstream(None, 4 * 1024 * 1024, 302).await; + let mut out = Vec::new(); + let code = runtime + .exec( + Command::Argv(vec![ + "/bin/guest.wasm".into(), + url, + "POST".into(), + (4 * 1024 * 1024).to_string(), + ]), + Options { + cwd: "/".into(), + timeout: Some(std::time::Duration::from_secs(30)), + ..Default::default() + }, + &mut |_, b| out.extend_from_slice(b), + ) + .await + .unwrap(); + assert_eq!(code, 0); + assert!(String::from_utf8(out).unwrap().contains("status=302")); + task.await.unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + runtime + .network + .deny_host_port(listener.local_addr().unwrap().port()); + assert!( + run( + &runtime, + format!("http://{}/", listener.local_addr().unwrap()), + vec![] + ) + .await + .contains("error") + ); +} diff --git a/host/tests/hook_git.rs b/host/tests/hook_git.rs new file mode 100644 index 0000000..463cce1 --- /dev/null +++ b/host/tests/hook_git.rs @@ -0,0 +1,208 @@ +//! A real git component pushes an incompressible commit through the hook to native git's HTTP backend. +use sausage::{ + exec::{Command, Exec, Options}, + process::Runtime, + world::World, +}; +use std::{ + io::Write, + path::Path, + process::Stdio, + sync::{Arc, OnceLock}, +}; +fn git(cwd: &Path, args: &[&str]) { + assert!( + std::process::Command::new("git") + .args(args) + .current_dir(cwd) + .status() + .unwrap() + .success() + ); +} +#[tokio::test] +async fn several_mb_git_push() { + static BUILT: OnceLock = OnceLock::new(); + let component = BUILT.get_or_init(|| { + let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("hook-git"); + let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland"); + if !userland.join("cwd-init/init.o").is_file() { + assert!( + std::process::Command::new("nu") + .args(["--no-config-file", "cwd-init/build.nu"]) + .current_dir(&userland) + .status() + .unwrap() + .success() + ); + } + let manifest = userland.join("Cargo.toml"); + assert!( + std::process::Command::new("cargo") + .args(["build", "--release", "-p", "git", "--manifest-path"]) + .arg(manifest) + .current_dir(&userland) + .args(["--target", "wasm32-wasip2", "--target-dir"]) + .arg(&target) + .args(["-j", "2", "--locked"]) + .status() + .unwrap() + .success() + ); + target.join("wasm32-wasip2/release/git.wasm") + }); + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + let root = temp.path().join("world"); + std::fs::create_dir_all(root.join("bin")).unwrap(); + std::fs::copy(component, root.join("bin/git.wasm")).unwrap(); + let repo = root.join("repo"); + std::fs::create_dir_all(&repo).unwrap(); + git(&repo, &["init", "-b", "main"]); + let mut data = vec![0; 4 * 1024 * 1024]; + openssl::rand::rand_bytes(&mut data).unwrap(); + std::fs::write(repo.join("large"), data).unwrap(); + git(&repo, &["add", "large"]); + git( + &repo, + &[ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.com", + "commit", + "-m", + "large commit", + ], + ); + let remote = temp.path().join("remote"); + std::fs::create_dir_all(&remote).unwrap(); + git(&remote, &["init", "--bare", "destination.git"]); + git( + &remote.join("destination.git"), + &["config", "http.receivepack", "true"], + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + git( + &repo, + &[ + "remote", + "add", + "origin", + &format!("http://127.0.0.1:{port}/destination.git"), + ], + ); + let uploaded = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let counter = uploaded.clone(); + let server = tokio::spawn(async move { + loop { + let (socket, _) = listener.accept().await.unwrap(); + let remote = remote.clone(); + let counter = counter.clone(); + let service = + hyper::service::service_fn(move |request: http::Request| { + let remote = remote.clone(); + let counter = counter.clone(); + async move { + use http_body_util::{BodyExt, Full}; + let (parts, body) = request.into_parts(); + let body = body.collect().await.unwrap().to_bytes(); + counter.fetch_add(body.len(), std::sync::atomic::Ordering::Relaxed); + let response = tokio::task::spawn_blocking(move || { + let mut process = std::process::Command::new("git") + .arg("http-backend") + .env("GIT_PROJECT_ROOT", remote) + .env("GIT_HTTP_EXPORT_ALL", "1") + .env("REQUEST_METHOD", parts.method.as_str()) + .env("PATH_INFO", parts.uri.path()) + .env("QUERY_STRING", parts.uri.query().unwrap_or("")) + .env( + "CONTENT_TYPE", + parts + .headers + .get("content-type") + .map(|h| h.to_str().unwrap()) + .unwrap_or(""), + ) + .env("CONTENT_LENGTH", body.len().to_string()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let mut stdin = process.stdin.take().unwrap(); + let output = std::thread::scope(|scope| { + scope.spawn(move || { + stdin.write_all(&body).unwrap(); + }); + process.wait_with_output().unwrap() + }); + assert!(output.status.success()); + let split = output + .stdout + .windows(4) + .position(|b| b == b"\r\n\r\n") + .unwrap(); + let headers = std::str::from_utf8(&output.stdout[..split]).unwrap(); + let mut response = http::Response::builder(); + for line in headers.split("\r\n") { + let (name, value) = line.split_once(':').unwrap(); + if name.eq_ignore_ascii_case("status") { + response = response.status( + value + .trim() + .split(' ') + .next() + .unwrap() + .parse::() + .unwrap(), + ); + } else { + response = response.header(name, value.trim()); + } + } + response + .body(Full::new(bytes::Bytes::copy_from_slice( + &output.stdout[split + 4..], + ))) + .unwrap() + }) + .await + .unwrap(); + Ok::<_, std::convert::Infallible>(response) + } + }); + hyper::server::conn::http1::Builder::new() + .keep_alive(false) + .serve_connection(hyper_util::rt::TokioIo::new(socket), service) + .await + .unwrap(); + } + }); + let runtime = Runtime::new(world).unwrap(); + let mut output = Vec::new(); + let code = runtime + .exec( + Command::Argv(vec![ + "/bin/git.wasm".into(), + "push".into(), + "origin".into(), + "main".into(), + ]), + Options { + cwd: "/repo".into(), + timeout: Some(std::time::Duration::from_secs(60)), + ..Default::default() + }, + &mut |_, b| output.extend_from_slice(b), + ) + .await + .unwrap(); + server.abort(); + assert_eq!(code, 0, "{}", String::from_utf8_lossy(&output)); + assert!(uploaded.load(std::sync::atomic::Ordering::Relaxed) > 3 * 1024 * 1024); + git( + &temp.path().join("remote/destination.git"), + &["cat-file", "-e", "refs/heads/main:large"], + ); +} diff --git a/host/tests/hook_guest.rs b/host/tests/hook_guest.rs new file mode 100644 index 0000000..0344d34 --- /dev/null +++ b/host/tests/hook_guest.rs @@ -0,0 +1,64 @@ +//! Persistent test component shared by HTTP and socket acceptance tests. +use std::{ + path::{Path, PathBuf}, + sync::OnceLock, +}; +fn write(path: PathBuf, text: &str) -> std::io::Result<()> { + if std::fs::read_to_string(&path).ok().as_deref() != Some(text) { + std::fs::write(path, text)?; + } + Ok(()) +} +pub fn component() -> &'static Path { + static PATH: OnceLock = OnceLock::new(); + PATH.get_or_init(|| { + let directory = Path::new(env!("CARGO_TARGET_TMPDIR")).join("hook-guest"); + std::fs::create_dir_all(directory.join("src")).unwrap(); + std::fs::create_dir_all(directory.join("wit/deps")).unwrap(); + write(directory.join("wit/process.wit"), include_str!("../wit/process.wit")).unwrap(); + write(directory.join("wit/deps/io.wit"), include_str!("../wit/deps/io.wit")).unwrap(); + write(directory.join("Cargo.toml"), "[package]\nname='hook-guest'\nversion='0.1.0'\nedition='2024'\n[workspace]\n[dependencies]\nwstd='=0.6.8'\nanyhow='1'\nfutures-lite='2'\nwit-bindgen='0.46'\n").unwrap(); + write(directory.join("src/main.rs"), r#" +use wstd::http::{Body, BodyExt, Client, Request}; +mod bindings { wit_bindgen::generate!({path:"wit",world:"program",generate_all}); } +#[wstd::main] +async fn main() -> anyhow::Result<()> { + let args: Vec<_> = std::env::args().collect(); + match args[1].as_str() { + "spawn" => { + use bindings::sausage::process::host::{self, Request, Stdio}; + let child=host::spawn(&Request {program:"/bin/guest.wasm".into(),argv:vec!["guest".into(),args[2].clone(),"GET".into()],cwd:"/".into(),env:vec![],group:None,stdin:Stdio::Null,stdout:Stdio::Inherit,stderr:Stdio::Inherit}).unwrap(); + println!("child={}",child.pid()); + assert_eq!(child.wait(),0); + }, + "listen" => { + let listener = std::net::TcpListener::bind(&args[2])?; + println!("ready"); + let (mut stream, _) = listener.accept()?; + use std::io::{Read, Write}; + let mut bytes = [0;4]; stream.read_exact(&mut bytes)?; + stream.write_all(&bytes)?; + }, + "connect" => { + use std::io::{Read, Write}; + match std::net::TcpStream::connect(&args[2]) { + Ok(mut stream) => { stream.write_all(b"ping")?; let mut bytes=[0;4]; stream.read_exact(&mut bytes)?; println!("{}", std::str::from_utf8(&bytes)?); }, + Err(_) => println!("denied"), + } + }, + _ => { + let body = if args.len() > 3 { Body::from_stream(futures_lite::stream::iter((0..args[3].parse::()?/8192).map(|_| vec![b'x';8192]))) } else { Body::empty() }; + let request = Request::builder().uri(&args[1]).method(args[2].as_str()).body(body)?; + match Client::new().send(request).await { + Ok(response) => { println!("status={}", response.status().as_u16()); let bytes=response.into_body().into_boxed_body().collect().await?.to_bytes(); println!("{}",String::from_utf8_lossy(&bytes)); }, + Err(_) => println!("error"), + } + } + } + Ok(()) +} +"#).unwrap(); + assert!(std::process::Command::new("cargo").args(["build", "--manifest-path"]).arg(directory.join("Cargo.toml")).args(["--target", "wasm32-wasip2", "--release", "-j", "2"]).env("CARGO_TARGET_DIR", directory.join("target")).status().unwrap().success(), "test guest build failed"); + directory.join("target/wasm32-wasip2/release/hook-guest.wasm") + }) +} diff --git a/host/tests/net.rs b/host/tests/net.rs new file mode 100644 index 0000000..b923957 --- /dev/null +++ b/host/tests/net.rs @@ -0,0 +1,125 @@ +//! Real components can accept TCP connections. Host port registration and host-only config deny access. +mod hook_guest; +use sausage::{ + exec::{Command, Exec, Options}, + process::{Runtime, net::NetPolicy}, + world::World, +}; +use std::{sync::Arc, time::Duration}; +#[test] +fn host_policy() { + let temp = tempfile::tempdir().unwrap(); + std::fs::write( + temp.path().join("network.toml"), + "deny = ['127.0.0.1:9001', '[::1]:9002']", + ) + .unwrap(); + let policy = NetPolicy::open(temp.path()).unwrap(); + policy.deny_host_port(9000); + for address in [ + "127.0.0.1:9000", + "0.0.0.0:9000", + "[::1]:9000", + "[::ffff:127.0.0.1]:9000", + "127.0.0.1:9001", + "[::1]:9002", + ] { + assert!(!policy.allows(address.parse().unwrap())); + } + assert!(policy.allows("0.0.0.0:8080".parse().unwrap())); + for interface in if_addrs::get_if_addrs().unwrap() { + assert!(!policy.allows(std::net::SocketAddr::new(interface.ip(), 9000))); + } + assert!(!policy.allows("[::ffff:127.0.0.1]:9001".parse().unwrap())); +} +async fn run(runtime: &Arc, mode: &str, address: String) -> String { + let mut output = Vec::new(); + let code = runtime + .exec( + Command::Argv(vec!["/bin/guest.wasm".into(), mode.into(), address]), + Options { + cwd: "/".into(), + timeout: Some(Duration::from_secs(30)), + ..Default::default() + }, + &mut |_, bytes| output.extend_from_slice(bytes), + ) + .await + .unwrap(); + assert_eq!(code, 0); + String::from_utf8(output).unwrap() +} +#[tokio::test] +async fn guest_tcp_and_host_denial() { + let guest = hook_guest::component(); + let temp = tempfile::tempdir().unwrap(); + let world = World::open(temp.path()).unwrap(); + std::fs::create_dir_all(temp.path().join("world/bin")).unwrap(); + std::fs::copy(guest, temp.path().join("world/bin/guest.wasm")).unwrap(); + let configured = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + std::fs::write( + world.host_dir.join("network.toml"), + format!("deny = ['{}']", configured.local_addr().unwrap()), + ) + .unwrap(); + let runtime = Runtime::new(world).unwrap(); + assert!( + run( + &runtime, + "connect", + configured.local_addr().unwrap().to_string() + ) + .await + .contains("denied") + ); + let native = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let denied = native.local_addr().unwrap(); + runtime.network.deny_host_port(denied.port()); + assert!( + run(&runtime, "connect", denied.to_string()) + .await + .contains("denied") + ); + let ephemeral = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = ephemeral.local_addr().unwrap(); + drop(ephemeral); + let server_runtime = runtime.clone(); + let (tx, rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + let mut tx = Some(tx); + let mut bytes = Vec::new(); + server_runtime + .exec( + Command::Argv(vec![ + "/bin/guest.wasm".into(), + "listen".into(), + address.to_string(), + ]), + Options { + cwd: "/".into(), + timeout: Some(Duration::from_secs(30)), + ..Default::default() + }, + &mut |_, chunk| { + bytes.extend_from_slice(chunk); + if bytes.windows(5).any(|w| w == b"ready") { + if let Some(tx) = tx.take() { + let _ = tx.send(()); + } + } + }, + ) + .await + .unwrap() + }); + tokio::time::timeout(Duration::from_secs(30), rx) + .await + .unwrap() + .unwrap(); + assert!( + run(&runtime, "connect", format!("localhost:{}", address.port())) + .await + .contains("ping") + ); + assert_eq!(server.await.unwrap(), 0); +} diff --git a/host/tests/process.rs b/host/tests/process.rs index d980b7a..c91961a 100644 --- a/host/tests/process.rs +++ b/host/tests/process.rs @@ -271,7 +271,7 @@ mod features { exec.exec( Command::Shell("sleep 60 | cat".into()), Options { - conversation: Some("c1".into()), + conversations: vec!["c1".into()], ..Default::default() }, &mut |_, _| {},