diff --git a/Cargo.lock b/Cargo.lock index 864042d9..3f545e36 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1413,6 +1413,25 @@ dependencies = [ "stable_deref_trait", ] +[[package]] +name = "h2" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap 2.13.0", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -1550,6 +1569,7 @@ dependencies = [ "bytes", "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", @@ -2117,6 +2137,8 @@ dependencies = [ "async-trait", "axum", "blake3", + "bollard", + "bytes", "chrono", "clap", "crossterm", @@ -2125,6 +2147,9 @@ dependencies = [ "dotenvy", "futures", "hkdf", + "http-body-util", + "hyper", + "hyper-util", "open", "pgvector", "postgres-types", diff --git a/Cargo.toml b/Cargo.toml index 8b05502f..8af7ba00 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -89,6 +89,15 @@ sha2 = "0.10" blake3 = "1" rand = "0.8" +# Docker sandbox +bollard = "0.18" + +# HTTP proxy for sandboxed network access +hyper = { version = "1.5", features = ["server", "http1", "http2"] } +hyper-util = { version = "0.1", features = ["server", "tokio", "http1", "http2"] } +http-body-util = "0.1" +bytes = "1" + [dev-dependencies] tokio-test = "0.4" testcontainers-modules = { version = "0.11", features = ["postgres"] } diff --git a/channels/whatsapp/Cargo.lock b/channels/whatsapp/Cargo.lock new file mode 100644 index 00000000..0e55d1e5 --- /dev/null +++ b/channels/whatsapp/Cargo.lock @@ -0,0 +1,401 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + +[[package]] +name = "anyhow" +version = "1.0.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" + +[[package]] +name = "bitflags" +version = "2.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + +[[package]] +name = "indexmap" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017" +dependencies = [ + "equivalent", + "hashbrown 0.16.1", + "serde", + "serde_core", +] + +[[package]] +name = "itoa" +version = "1.0.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" + +[[package]] +name = "leb128" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "884e2677b40cc8c339eaefcb701c32ef1fd2493d71118dc0ca4b6a736c93bd67" + +[[package]] +name = "log" +version = "0.4.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" + +[[package]] +name = "memchr" +version = "2.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" + +[[package]] +name = "once_cell" +version = "1.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" + +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21b2ebcf727b7760c461f091f9f0f539b77b8e87f2fd88131e7f1b433b3cece4" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "semver" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2" + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.149" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "smallvec" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" + +[[package]] +name = "spdx" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3e17e880bafaeb362a7b751ec46bdc5b61445a188f80e0606e68167cd540fa3" +dependencies = [ + "smallvec", +] + +[[package]] +name = "syn" +version = "2.0.114" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4d107df263a3013ef9b1879b0df87d706ff80f65a86ea879bd9c31f9b307c2a" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5" + +[[package]] +name = "unicode-xid" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasm-encoder" +version = "0.220.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e913f9242315ca39eff82aee0e19ee7a372155717ff0eb082c741e435ce25ed1" +dependencies = [ + "leb128", + "wasmparser", +] + +[[package]] +name = "wasm-metadata" +version = "0.220.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "185dfcd27fa5db2e6a23906b54c28199935f71d9a27a1a27b3a88d6fee2afae7" +dependencies = [ + "anyhow", + "indexmap", + "serde", + "serde_derive", + "serde_json", + "spdx", + "wasm-encoder", + "wasmparser", +] + +[[package]] +name = "wasmparser" +version = "0.220.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d07b6a3b550fefa1a914b6d54fc175dd11c3392da11eee604e6ffc759805d25" +dependencies = [ + "ahash", + "bitflags", + "hashbrown 0.14.5", + "indexmap", + "semver", +] + +[[package]] +name = "whatsapp-channel" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "wit-bindgen", +] + +[[package]] +name = "wit-bindgen" +version = "0.36.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a2b3e15cd6068f233926e7d8c7c588b2ec4fb7cc7bf3824115e7c7e2a8485a3" +dependencies = [ + "wit-bindgen-rt", + "wit-bindgen-rust-macro", +] + +[[package]] +name = "wit-bindgen-core" +version = "0.36.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b632a5a0fa2409489bd49c9e6d99fcc61bb3d4ce9d1907d44662e75a28c71172" +dependencies = [ + "anyhow", + "heck", + "wit-parser", +] + +[[package]] +name = "wit-bindgen-rt" +version = "0.36.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7947d0131c7c9da3f01dfde0ab8bd4c4cf3c5bd49b6dba0ae640f1fa752572ea" +dependencies = [ + "bitflags", +] + +[[package]] +name = "wit-bindgen-rust" +version = "0.36.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4329de4186ee30e2ef30a0533f9b3c123c019a237a7c82d692807bf1b3ee2697" +dependencies = [ + "anyhow", + "heck", + "indexmap", + "prettyplease", + "syn", + "wasm-metadata", + "wit-bindgen-core", + "wit-component", +] + +[[package]] +name = "wit-bindgen-rust-macro" +version = "0.36.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "177fb7ee1484d113b4792cc480b1ba57664bbc951b42a4beebe573502135b1fc" +dependencies = [ + "anyhow", + "prettyplease", + "proc-macro2", + "quote", + "syn", + "wit-bindgen-core", + "wit-bindgen-rust", +] + +[[package]] +name = "wit-component" +version = "0.220.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b505603761ed400c90ed30261f44a768317348e49f1864e82ecdc3b2744e5627" +dependencies = [ + "anyhow", + "bitflags", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder", + "wasm-metadata", + "wasmparser", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.220.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae2a7999ed18efe59be8de2db9cb2b7f84d88b27818c79353dfc53131840fe1a" +dependencies = [ + "anyhow", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-xid", + "wasmparser", +] + +[[package]] +name = "zerocopy" +version = "0.8.38" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57cf3aa6855b23711ee9852dfc97dfaa51c45feaba5b645d0c777414d494a961" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.38" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a616990af1a287837c4fe6596ad77ef57948f787e46ce28e166facc0cc1cb75" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "zmij" +version = "1.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445" diff --git a/channels/whatsapp/Cargo.toml b/channels/whatsapp/Cargo.toml new file mode 100644 index 00000000..6e91d1f3 --- /dev/null +++ b/channels/whatsapp/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "whatsapp-channel" +version = "0.1.0" +edition = "2021" +description = "WhatsApp channel for near-agent using the Cloud API" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +wit-bindgen = "0.36" +serde = { version = "1", features = ["derive"] } +serde_json = "1" + +[profile.release] +opt-level = "s" +lto = true +strip = true diff --git a/channels/whatsapp/src/lib.rs b/channels/whatsapp/src/lib.rs new file mode 100644 index 00000000..3fb1d875 --- /dev/null +++ b/channels/whatsapp/src/lib.rs @@ -0,0 +1,266 @@ +//! WhatsApp Channel for near-agent +//! +//! Implements the channel interface for WhatsApp Cloud API. +//! Handles incoming webhooks and sends responses via the API. + +use serde::{Deserialize, Serialize}; + +// Generate bindings from the WIT file +wit_bindgen::generate!({ + world: "sandboxed-channel", + path: "../../wit/channel.wit", +}); + +use exports::near::agent::channel::*; +use near::agent::channel_host::*; + +struct WhatsAppChannel; + +impl Guest for WhatsAppChannel { + fn on_start(config_json: String) -> Result { + log(LogLevel::Info, &format!("WhatsApp channel starting with config: {}", config_json)); + + Ok(ChannelConfig { + display_name: "WhatsApp".to_string(), + http_endpoints: vec![ + HttpEndpointConfig { + path: "/webhook/whatsapp".to_string(), + methods: vec!["GET".to_string(), "POST".to_string()], + require_secret: true, + }, + ], + poll: None, // WhatsApp uses webhooks, not polling + }) + } + + fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse { + log(LogLevel::Debug, &format!("Received {} request to {}", req.method, req.path)); + + // Handle webhook verification (GET request) + if req.method == "GET" { + return handle_verification(&req); + } + + // Handle incoming messages (POST request) + if req.method == "POST" { + return handle_incoming_message(&req); + } + + // Method not allowed + OutgoingHttpResponse { + status: 405, + headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), + body: b"Method not allowed".to_vec(), + } + } + + fn on_poll() { + // WhatsApp uses webhooks, no polling needed + } + + fn on_respond(response: AgentResponse) -> Result<(), String> { + log(LogLevel::Info, &format!("Sending response to WhatsApp: {}", response.message_id)); + + // Parse metadata to get phone number + let metadata: ResponseMetadata = serde_json::from_str(&response.metadata_json) + .map_err(|e| format!("Failed to parse metadata: {}", e))?; + + // Build WhatsApp API request + let phone_number_id = metadata.phone_number_id.ok_or("Missing phone_number_id")?; + let recipient = metadata.recipient.ok_or("Missing recipient")?; + + let api_url = format!( + "https://graph.facebook.com/v18.0/{}/messages", + phone_number_id + ); + + let request_body = serde_json::json!({ + "messaging_product": "whatsapp", + "to": recipient, + "type": "text", + "text": { + "body": response.content + } + }); + + let body_bytes = serde_json::to_vec(&request_body) + .map_err(|e| format!("Failed to serialize request: {}", e))?; + + // Make API request (host will inject the access token) + let result = http_request( + "POST", + &api_url, + r#"{"Content-Type": "application/json"}"#, + Some(&body_bytes), + ); + + match result { + Ok(resp) if resp.status >= 200 && resp.status < 300 => { + log(LogLevel::Info, "Message sent successfully"); + Ok(()) + } + Ok(resp) => { + let body_str = String::from_utf8_lossy(&resp.body); + Err(format!("WhatsApp API error {}: {}", resp.status, body_str)) + } + Err(e) => Err(format!("HTTP request failed: {}", e)), + } + } + + fn on_shutdown() { + log(LogLevel::Info, "WhatsApp channel shutting down"); + } +} + +/// Handle WhatsApp webhook verification request +fn handle_verification(req: &IncomingHttpRequest) -> OutgoingHttpResponse { + // Parse query parameters + let query: serde_json::Value = serde_json::from_str(&req.query_json) + .unwrap_or(serde_json::Value::Null); + + let mode = query.get("hub.mode").and_then(|v| v.as_str()); + let challenge = query.get("hub.challenge").and_then(|v| v.as_str()); + + // WhatsApp sends hub.mode=subscribe for verification + if mode == Some("subscribe") { + if let Some(challenge) = challenge { + log(LogLevel::Info, "Webhook verification successful"); + return OutgoingHttpResponse { + status: 200, + headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), + body: challenge.as_bytes().to_vec(), + }; + } + } + + OutgoingHttpResponse { + status: 403, + headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), + body: b"Verification failed".to_vec(), + } +} + +/// Handle incoming WhatsApp message +fn handle_incoming_message(req: &IncomingHttpRequest) -> OutgoingHttpResponse { + // Parse webhook payload + let payload: WebhookPayload = match serde_json::from_slice(&req.body) { + Ok(p) => p, + Err(e) => { + log(LogLevel::Warn, &format!("Failed to parse webhook payload: {}", e)); + return OutgoingHttpResponse { + status: 400, + headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), + body: b"Invalid payload".to_vec(), + }; + } + }; + + // Process each entry + for entry in payload.entry.iter() { + for change in entry.changes.iter() { + if change.field != "messages" { + continue; + } + + let value = &change.value; + let phone_number_id = value.metadata.phone_number_id.clone(); + + // Process messages + for message in value.messages.iter() { + // Only handle text messages for now + if message.r#type != "text" { + continue; + } + + let text = message.text.as_ref().map(|t| t.body.clone()).unwrap_or_default(); + let from = message.from.clone(); + let msg_id = message.id.clone(); + + // Build metadata for response routing + let metadata = MessageMetadata { + phone_number_id: phone_number_id.clone(), + message_id: msg_id.clone(), + timestamp: message.timestamp.clone(), + }; + + // Emit message to the agent + emit_message(&EmittedMessage { + user_id: from.clone(), + user_name: None, // Could look up contact name + content: text, + thread_id: None, + metadata_json: serde_json::to_string(&metadata).unwrap_or_default(), + }); + + log(LogLevel::Info, &format!("Emitted message from {}", from)); + } + } + } + + // Acknowledge receipt + OutgoingHttpResponse { + status: 200, + headers_json: r#"{"Content-Type": "text/plain"}"#.to_string(), + body: b"OK".to_vec(), + } +} + +// ==================== WhatsApp API Types ==================== + +#[derive(Debug, Deserialize)] +struct WebhookPayload { + entry: Vec, +} + +#[derive(Debug, Deserialize)] +struct WebhookEntry { + changes: Vec, +} + +#[derive(Debug, Deserialize)] +struct WebhookChange { + field: String, + value: WebhookValue, +} + +#[derive(Debug, Deserialize)] +struct WebhookValue { + metadata: WhatsAppMetadata, + #[serde(default)] + messages: Vec, +} + +#[derive(Debug, Deserialize)] +struct WhatsAppMetadata { + phone_number_id: String, +} + +#[derive(Debug, Deserialize)] +struct WhatsAppMessage { + id: String, + from: String, + timestamp: String, + r#type: String, + text: Option, +} + +#[derive(Debug, Deserialize)] +struct WhatsAppText { + body: String, +} + +#[derive(Debug, Serialize)] +struct MessageMetadata { + phone_number_id: String, + message_id: String, + timestamp: String, +} + +#[derive(Debug, Deserialize)] +struct ResponseMetadata { + phone_number_id: Option, + recipient: Option, +} + +// Export the channel implementation +export!(WhatsAppChannel); diff --git a/docker/sandbox.Dockerfile b/docker/sandbox.Dockerfile new file mode 100644 index 00000000..69c4da15 --- /dev/null +++ b/docker/sandbox.Dockerfile @@ -0,0 +1,25 @@ +FROM rust:1.86-slim-bookworm + +# Install build dependencies +RUN apt-get update && apt-get install -y --no-install-recommends \ + bash \ + ca-certificates \ + curl \ + git \ + pkg-config \ + && rm -rf /var/lib/apt/lists/* + +# Install WASM targets +RUN rustup target add wasm32-wasip2 wasm32-unknown-unknown + +# Install wasm-tools for component manipulation +RUN cargo install wasm-tools --locked + +# Create non-root user for sandbox +RUN useradd -m -u 1000 sandbox +USER sandbox + +WORKDIR /workspace + +# Default command +CMD ["bash"] diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 41d13235..ca143690 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -64,13 +64,18 @@ pub struct Agent { impl Agent { /// Create a new agent. + /// + /// Optionally accepts a pre-created `ContextManager` for sharing with job tools. + /// If not provided, creates a new one. pub fn new( config: AgentConfig, deps: AgentDeps, channels: ChannelManager, heartbeat_config: Option, + context_manager: Option>, ) -> Self { - let context_manager = Arc::new(ContextManager::new(config.max_parallel_jobs)); + let context_manager = context_manager + .unwrap_or_else(|| Arc::new(ContextManager::new(config.max_parallel_jobs))); let scheduler = Arc::new(Scheduler::new( config.clone(), @@ -388,25 +393,21 @@ impl Agent { } } - // Route for job commands (bypass turn system) - // Build a temporary message with the content to route + // Handle explicit commands (starting with /) directly + // Everything else goes through the normal agentic loop with tools let temp_message = IncomingMessage { content: content.to_string(), ..message.clone() }; - let intent = self.router.route(&temp_message); - match &intent { - MessageIntent::CreateJob { .. } - | MessageIntent::CheckJobStatus { .. } - | MessageIntent::CancelJob { .. } - | MessageIntent::ListJobs { .. } - | MessageIntent::HelpJob { .. } - | MessageIntent::Command { .. } => { - return self.handle_job_or_command(intent, message).await; - } - _ => {} + + if let Some(intent) = self.router.route_command(&temp_message) { + // Explicit command like /status, /job, /list - handle directly + return self.handle_job_or_command(intent, message).await; } + // Natural language goes through the agentic loop + // Job tools (create_job, list_jobs, etc.) are in the tool registry + // Auto-compact if needed BEFORE adding new turn { let mut sess = session.lock().await; diff --git a/src/agent/router.rs b/src/agent/router.rs index 273daa9b..dccb4771 100644 --- a/src/agent/router.rs +++ b/src/agent/router.rs @@ -1,4 +1,8 @@ //! Message routing to appropriate handlers. +//! +//! The router handles explicit commands (starting with `/`). +//! Natural language intent classification is handled by `IntentClassifier` +//! which uses LLM + tools instead of brittle pattern matching. use crate::channels::IncomingMessage; @@ -27,7 +31,9 @@ pub enum MessageIntent { Unknown, } -/// Routes messages to appropriate handlers based on intent. +/// Routes messages to appropriate handlers based on explicit commands. +/// +/// For natural language messages, use `IntentClassifier` instead. pub struct Router { /// Command prefix (e.g., "/" or "!") command_prefix: String, @@ -47,17 +53,23 @@ impl Router { self } - /// Route a message to determine its intent. - pub fn route(&self, message: &IncomingMessage) -> MessageIntent { + /// Check if a message is an explicit command. + pub fn is_command(&self, message: &IncomingMessage) -> bool { + message.content.trim().starts_with(&self.command_prefix) + } + + /// Route an explicit command to determine its intent. + /// + /// Returns `None` if the message is not a command. + /// For non-commands, use `IntentClassifier::classify()` instead. + pub fn route_command(&self, message: &IncomingMessage) -> Option { let content = message.content.trim(); - // Check for commands if content.starts_with(&self.command_prefix) { - return self.parse_command(content); + Some(self.parse_command(content)) + } else { + None } - - // Try to extract intent from natural language - self.extract_intent(content) } fn parse_command(&self, content: &str) -> MessageIntent { @@ -111,61 +123,6 @@ impl Router { None => MessageIntent::Unknown, } } - - fn extract_intent(&self, content: &str) -> MessageIntent { - let lower = content.to_lowercase(); - - // Job creation patterns - must be explicit about creating a job - // More specific patterns to avoid capturing general conversation - let is_job_creation = lower.starts_with("create job ") - || lower.starts_with("new job ") - || lower.starts_with("schedule job ") - || lower.starts_with("run job ") - || (lower.contains("create") && lower.contains("job")); - - if is_job_creation { - return MessageIntent::CreateJob { - title: extract_title(content), - description: content.to_string(), - category: extract_category(content), - }; - } - - // Status check patterns - if lower.contains("status") - || lower.contains("how is") - || lower.contains("progress") - || lower.starts_with("check ") - { - return MessageIntent::CheckJobStatus { - job_id: extract_job_id(content), - }; - } - - // Cancel patterns - if lower.contains("cancel") || lower.contains("stop") || lower.contains("abort") { - if let Some(job_id) = extract_job_id(content) { - return MessageIntent::CancelJob { job_id }; - } - } - - // List patterns - if lower.starts_with("list") || lower.contains("show jobs") || lower.contains("my jobs") { - return MessageIntent::ListJobs { filter: None }; - } - - // Help patterns - if lower.contains("stuck") || lower.contains("not working") || lower.contains("fix") { - if let Some(job_id) = extract_job_id(content) { - return MessageIntent::HelpJob { job_id }; - } - } - - // Default to chat - MessageIntent::Chat { - content: content.to_string(), - } - } } impl Default for Router { @@ -174,53 +131,6 @@ impl Default for Router { } } -/// Extract a title from content. -fn extract_title(content: &str) -> String { - // Take first sentence or first N characters - let first_sentence = content.split('.').next().unwrap_or(content); - let title = first_sentence.chars().take(100).collect::(); - if title.len() < first_sentence.len() { - format!("{}...", title) - } else { - title - } -} - -/// Extract a category from content. -fn extract_category(content: &str) -> Option { - let lower = content.to_lowercase(); - - let categories = [ - ("code", "development"), - ("program", "development"), - ("website", "web"), - ("api", "development"), - ("data", "data"), - ("write", "writing"), - ("design", "design"), - ("research", "research"), - ]; - - for (keyword, category) in categories { - if lower.contains(keyword) { - return Some(category.to_string()); - } - } - - None -} - -/// Extract a job ID from content. -fn extract_job_id(content: &str) -> Option { - // Look for UUID patterns - let uuid_regex = regex::Regex::new( - r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", - ) - .ok()?; - - uuid_regex.find(content).map(|m| m.as_str().to_string()) -} - #[cfg(test)] mod tests { use super::*; @@ -230,35 +140,61 @@ mod tests { let router = Router::new(); let msg = IncomingMessage::new("test", "user", "/status abc-123"); - let intent = router.route(&msg); + let intent = router.route_command(&msg); - assert!(matches!(intent, MessageIntent::CheckJobStatus { .. })); + assert!(matches!(intent, Some(MessageIntent::CheckJobStatus { .. }))); } #[test] - fn test_natural_language_routing() { + fn test_is_command() { let router = Router::new(); + let cmd_msg = IncomingMessage::new("test", "user", "/status"); + assert!(router.is_command(&cmd_msg)); + + let chat_msg = IncomingMessage::new("test", "user", "Hello there"); + assert!(!router.is_command(&chat_msg)); + } + + #[test] + fn test_non_command_returns_none() { + let router = Router::new(); + + // Natural language messages return None - they should use IntentClassifier let msg = IncomingMessage::new("test", "user", "Can you create a website for me?"); - let intent = router.route(&msg); + assert!(router.route_command(&msg).is_none()); - assert!(matches!(intent, MessageIntent::CreateJob { .. })); + let msg2 = IncomingMessage::new("test", "user", "Hello, how are you?"); + assert!(router.route_command(&msg2).is_none()); } #[test] - fn test_chat_fallback() { + fn test_command_create_job() { let router = Router::new(); - let msg = IncomingMessage::new("test", "user", "Hello, how are you?"); - let intent = router.route(&msg); + let msg = IncomingMessage::new("test", "user", "/job build a website"); + let intent = router.route_command(&msg); - assert!(matches!(intent, MessageIntent::Chat { .. })); + match intent { + Some(MessageIntent::CreateJob { title, .. }) => { + assert_eq!(title, "build a website"); + } + _ => panic!("Expected CreateJob intent"), + } } #[test] - fn test_extract_job_id() { - let content = "Check status of job 550e8400-e29b-41d4-a716-446655440000"; - let id = extract_job_id(content); - assert_eq!(id, Some("550e8400-e29b-41d4-a716-446655440000".to_string())); + fn test_command_list_jobs() { + let router = Router::new(); + + let msg = IncomingMessage::new("test", "user", "/list active"); + let intent = router.route_command(&msg); + + match intent { + Some(MessageIntent::ListJobs { filter }) => { + assert_eq!(filter, Some("active".to_string())); + } + _ => panic!("Expected ListJobs intent"), + } } } diff --git a/src/config.rs b/src/config.rs index e561fe73..81f0e1d8 100644 --- a/src/config.rs +++ b/src/config.rs @@ -21,6 +21,7 @@ pub struct Config { pub secrets: SecretsConfig, pub builder: BuilderModeConfig, pub heartbeat: HeartbeatConfig, + pub sandbox: SandboxModeConfig, } impl Config { @@ -41,6 +42,7 @@ impl Config { secrets: SecretsConfig::from_env()?, builder: BuilderModeConfig::from_env()?, heartbeat: HeartbeatConfig::from_env()?, + sandbox: SandboxModeConfig::from_env()?, }) } } @@ -599,7 +601,7 @@ pub struct BuilderModeConfig { impl Default for BuilderModeConfig { fn default() -> Self { Self { - enabled: false, + enabled: true, // Builder enabled by default build_dir: None, max_iterations: 20, timeout_secs: 600, @@ -618,7 +620,7 @@ impl BuilderModeConfig { key: "BUILDER_ENABLED".to_string(), message: format!("must be 'true' or 'false': {e}"), })? - .unwrap_or(false), + .unwrap_or(true), // Builder enabled by default build_dir: optional_env("BUILDER_DIR")?.map(PathBuf::from), max_iterations: parse_optional_env("BUILDER_MAX_ITERATIONS", 20)?, timeout_secs: parse_optional_env("BUILDER_TIMEOUT_SECS", 600)?, @@ -690,6 +692,99 @@ impl HeartbeatConfig { } } +/// Docker sandbox configuration. +#[derive(Debug, Clone)] +pub struct SandboxModeConfig { + /// Whether the Docker sandbox is enabled. + pub enabled: bool, + /// Sandbox policy: "readonly", "workspace_write", or "full_access". + pub policy: String, + /// Command timeout in seconds. + pub timeout_secs: u64, + /// Memory limit in megabytes. + pub memory_limit_mb: u64, + /// CPU shares (relative weight). + pub cpu_shares: u32, + /// Docker image for the sandbox. + pub image: String, + /// Whether to auto-pull the image if not found. + pub auto_pull_image: bool, + /// Additional domains to allow through the network proxy. + pub extra_allowed_domains: Vec, +} + +impl Default for SandboxModeConfig { + fn default() -> Self { + Self { + enabled: false, // Disabled by default + policy: "readonly".to_string(), + timeout_secs: 120, + memory_limit_mb: 2048, + cpu_shares: 1024, + image: "ghcr.io/nearai/sandbox:latest".to_string(), + auto_pull_image: true, + extra_allowed_domains: Vec::new(), + } + } +} + +impl SandboxModeConfig { + fn from_env() -> Result { + let extra_domains = optional_env("SANDBOX_EXTRA_DOMAINS")? + .map(|s| s.split(',').map(|d| d.trim().to_string()).collect()) + .unwrap_or_default(); + + Ok(Self { + enabled: optional_env("SANDBOX_ENABLED")? + .map(|s| s.parse()) + .transpose() + .map_err(|e| ConfigError::InvalidValue { + key: "SANDBOX_ENABLED".to_string(), + message: format!("must be 'true' or 'false': {e}"), + })? + .unwrap_or(true), + policy: optional_env("SANDBOX_POLICY")?.unwrap_or_else(|| "readonly".to_string()), + timeout_secs: parse_optional_env("SANDBOX_TIMEOUT_SECS", 120)?, + memory_limit_mb: parse_optional_env("SANDBOX_MEMORY_LIMIT_MB", 2048)?, + cpu_shares: parse_optional_env("SANDBOX_CPU_SHARES", 1024)?, + image: optional_env("SANDBOX_IMAGE")? + .unwrap_or_else(|| "ghcr.io/nearai/sandbox:latest".to_string()), + auto_pull_image: optional_env("SANDBOX_AUTO_PULL")? + .map(|s| s.parse()) + .transpose() + .map_err(|e| ConfigError::InvalidValue { + key: "SANDBOX_AUTO_PULL".to_string(), + message: format!("must be 'true' or 'false': {e}"), + })? + .unwrap_or(true), + extra_allowed_domains: extra_domains, + }) + } + + /// Convert to SandboxConfig for the sandbox module. + pub fn to_sandbox_config(&self) -> crate::sandbox::SandboxConfig { + use crate::sandbox::SandboxPolicy; + use std::time::Duration; + + let policy = self.policy.parse().unwrap_or(SandboxPolicy::ReadOnly); + + let mut allowlist = crate::sandbox::default_allowlist(); + allowlist.extend(self.extra_allowed_domains.clone()); + + crate::sandbox::SandboxConfig { + enabled: self.enabled, + policy, + timeout: Duration::from_secs(self.timeout_secs), + memory_limit_mb: self.memory_limit_mb, + cpu_shares: self.cpu_shares, + network_allowlist: allowlist, + image: self.image.clone(), + auto_pull_image: self.auto_pull_image, + proxy_port: 0, // Auto-assign + } + } +} + // Helper functions fn required_env(key: &str) -> Result { diff --git a/src/lib.rs b/src/lib.rs index 3541d8a5..7cabc3f5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -49,6 +49,7 @@ pub mod evaluation; pub mod history; pub mod llm; pub mod safety; +pub mod sandbox; pub mod secrets; pub mod settings; pub mod setup; diff --git a/src/main.rs b/src/main.rs index f40e39fc..720a40f6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,7 @@ use near_agent::{ }, cli::{Cli, Command, run_tool_command}, config::Config, + context::ContextManager, history::Store, llm::{SessionConfig, create_llm_provider, create_session_manager}, safety::SafetyLayer, @@ -509,6 +510,12 @@ async fn main() -> anyhow::Result<()> { } } + // Create context manager (shared between job tools and agent) + let context_manager = Arc::new(ContextManager::new(config.agent.max_parallel_jobs)); + + // Register job tools + tools.register_job_tools(Arc::clone(&context_manager)); + // Create and run the agent let deps = AgentDeps { store, @@ -522,6 +529,7 @@ async fn main() -> anyhow::Result<()> { deps, channels, Some(config.heartbeat.clone()), + Some(context_manager), ); tracing::info!("Agent initialized, starting main loop..."); diff --git a/src/sandbox/config.rs b/src/sandbox/config.rs new file mode 100644 index 00000000..fa01ddc0 --- /dev/null +++ b/src/sandbox/config.rs @@ -0,0 +1,258 @@ +//! Configuration for the Docker execution sandbox. + +use std::time::Duration; + +/// Configuration for the sandbox system. +#[derive(Debug, Clone)] +pub struct SandboxConfig { + /// Whether the sandbox is enabled. + pub enabled: bool, + /// Security policy for sandbox execution. + pub policy: SandboxPolicy, + /// Default timeout for command execution. + pub timeout: Duration, + /// Memory limit in megabytes. + pub memory_limit_mb: u64, + /// CPU shares (relative weight, default 1024). + pub cpu_shares: u32, + /// Network allowlist for proxied requests. + pub network_allowlist: Vec, + /// Docker image to use for the sandbox. + pub image: String, + /// Whether to auto-pull the image if not found. + pub auto_pull_image: bool, + /// Port for the HTTP proxy (0 = auto-assign). + pub proxy_port: u16, +} + +impl Default for SandboxConfig { + fn default() -> Self { + Self { + enabled: false, // Disabled by default until Docker is confirmed available + policy: SandboxPolicy::ReadOnly, + timeout: Duration::from_secs(120), + memory_limit_mb: 2048, + cpu_shares: 1024, + network_allowlist: default_allowlist(), + image: "ghcr.io/nearai/sandbox:latest".to_string(), + auto_pull_image: true, + proxy_port: 0, + } + } +} + +/// Security policy for sandbox execution. +/// +/// ```text +/// ┌─────────────────────────────────────────────────────────────────────┐ +/// │ Sandbox Policies │ +/// ├─────────────────┬──────────────────┬────────────────────────────────┤ +/// │ Policy │ Filesystem │ Network │ +/// ├─────────────────┼──────────────────┼────────────────────────────────┤ +/// │ ReadOnly │ /workspace (ro) │ Proxied (allowlist only) │ +/// │ WorkspaceWrite │ /workspace (rw) │ Proxied (allowlist only) │ +/// │ FullAccess │ Full host │ Full network (DANGER) │ +/// └─────────────────┴──────────────────┴────────────────────────────────┘ +/// ``` +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum SandboxPolicy { + /// Read-only access to workspace, proxied network. + /// Use for: exploring code, fetching docs, read-only operations. + #[default] + ReadOnly, + + /// Read/write access to workspace, proxied network. + /// Use for: building software, running tests, generating files. + WorkspaceWrite, + + /// Full access (no sandbox). Use with extreme caution. + /// This bypasses all isolation and runs directly on host. + FullAccess, +} + +impl SandboxPolicy { + /// Returns true if filesystem writes are allowed. + pub fn allows_writes(&self) -> bool { + matches!( + self, + SandboxPolicy::WorkspaceWrite | SandboxPolicy::FullAccess + ) + } + + /// Returns true if network requests bypass the proxy. + pub fn has_full_network(&self) -> bool { + matches!(self, SandboxPolicy::FullAccess) + } + + /// Returns true if running in a container. + pub fn is_sandboxed(&self) -> bool { + !matches!(self, SandboxPolicy::FullAccess) + } +} + +impl std::str::FromStr for SandboxPolicy { + type Err = String; + + fn from_str(s: &str) -> Result { + match s.to_lowercase().as_str() { + "readonly" | "read_only" | "ro" => Ok(SandboxPolicy::ReadOnly), + "workspacewrite" | "workspace_write" | "rw" => Ok(SandboxPolicy::WorkspaceWrite), + "fullaccess" | "full_access" | "full" | "none" => Ok(SandboxPolicy::FullAccess), + _ => Err(format!( + "invalid sandbox policy '{}', expected 'readonly', 'workspace_write', or 'full_access'", + s + )), + } + } +} + +/// Resource limits for container execution. +#[derive(Debug, Clone)] +pub struct ResourceLimits { + /// Maximum memory in bytes. + pub memory_bytes: u64, + /// CPU shares (relative weight). + pub cpu_shares: u32, + /// Maximum execution time. + pub timeout: Duration, + /// Maximum output size in bytes. + pub max_output_bytes: usize, +} + +impl Default for ResourceLimits { + fn default() -> Self { + Self { + memory_bytes: 2 * 1024 * 1024 * 1024, // 2 GB + cpu_shares: 1024, + timeout: Duration::from_secs(120), + max_output_bytes: 64 * 1024, // 64 KB + } + } +} + +/// Default network allowlist for common development operations. +pub fn default_allowlist() -> Vec { + vec![ + // Package registries + "crates.io".to_string(), + "static.crates.io".to_string(), + "index.crates.io".to_string(), + "registry.npmjs.org".to_string(), + "proxy.golang.org".to_string(), + "pypi.org".to_string(), + "files.pythonhosted.org".to_string(), + // Documentation + "docs.rs".to_string(), + "doc.rust-lang.org".to_string(), + "nodejs.org".to_string(), + "go.dev".to_string(), + "docs.python.org".to_string(), + // Version control (read-only) + "github.com".to_string(), + "raw.githubusercontent.com".to_string(), + "api.github.com".to_string(), + "codeload.github.com".to_string(), + // Common APIs (credentials will be injected by proxy) + "api.openai.com".to_string(), + "api.anthropic.com".to_string(), + "api.near.ai".to_string(), + ] +} + +/// Credential injection configuration. +#[derive(Debug, Clone)] +pub struct CredentialMapping { + /// Domain this credential applies to. + pub domain: String, + /// Name of the secret to inject. + pub secret_name: String, + /// Where to inject the credential. + pub location: CredentialLocation, +} + +/// Where to inject a credential in an HTTP request. +#[derive(Debug, Clone)] +pub enum CredentialLocation { + /// Inject as Authorization: Bearer + AuthorizationBearer, + /// Inject as a custom header. + Header(String), + /// Inject as a query parameter. + QueryParam(String), +} + +impl Default for CredentialMapping { + fn default() -> Self { + Self { + domain: String::new(), + secret_name: String::new(), + location: CredentialLocation::AuthorizationBearer, + } + } +} + +/// Default credential mappings for common APIs. +pub fn default_credential_mappings() -> Vec { + vec![ + CredentialMapping { + domain: "api.openai.com".to_string(), + secret_name: "OPENAI_API_KEY".to_string(), + location: CredentialLocation::AuthorizationBearer, + }, + CredentialMapping { + domain: "api.anthropic.com".to_string(), + secret_name: "ANTHROPIC_API_KEY".to_string(), + location: CredentialLocation::Header("x-api-key".to_string()), + }, + CredentialMapping { + domain: "api.near.ai".to_string(), + secret_name: "NEARAI_API_KEY".to_string(), + location: CredentialLocation::AuthorizationBearer, + }, + ] +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_policy_parsing() { + assert_eq!( + "readonly".parse::().unwrap(), + SandboxPolicy::ReadOnly + ); + assert_eq!( + "workspace_write".parse::().unwrap(), + SandboxPolicy::WorkspaceWrite + ); + assert_eq!( + "full_access".parse::().unwrap(), + SandboxPolicy::FullAccess + ); + assert!("invalid".parse::().is_err()); + } + + #[test] + fn test_policy_properties() { + assert!(!SandboxPolicy::ReadOnly.allows_writes()); + assert!(SandboxPolicy::WorkspaceWrite.allows_writes()); + assert!(SandboxPolicy::FullAccess.allows_writes()); + + assert!(!SandboxPolicy::ReadOnly.has_full_network()); + assert!(!SandboxPolicy::WorkspaceWrite.has_full_network()); + assert!(SandboxPolicy::FullAccess.has_full_network()); + + assert!(SandboxPolicy::ReadOnly.is_sandboxed()); + assert!(SandboxPolicy::WorkspaceWrite.is_sandboxed()); + assert!(!SandboxPolicy::FullAccess.is_sandboxed()); + } + + #[test] + fn test_default_allowlist_has_common_registries() { + let allowlist = default_allowlist(); + assert!(allowlist.contains(&"crates.io".to_string())); + assert!(allowlist.contains(&"registry.npmjs.org".to_string())); + assert!(allowlist.contains(&"github.com".to_string())); + } +} diff --git a/src/sandbox/container.rs b/src/sandbox/container.rs new file mode 100644 index 00000000..29289667 --- /dev/null +++ b/src/sandbox/container.rs @@ -0,0 +1,519 @@ +//! Docker container lifecycle management. +//! +//! Handles creating, running, and cleaning up containers for sandboxed execution. +//! +//! # Container Setup +//! +//! ```text +//! ┌────────────────────────────────────────────────────────────────────────┐ +//! │ Docker Container │ +//! │ │ +//! │ Environment: │ +//! │ http_proxy=http://host.docker.internal:PORT │ +//! │ https_proxy=http://host.docker.internal:PORT │ +//! │ (No secrets or credentials) │ +//! │ │ +//! │ Mounts: │ +//! │ /workspace ─▶ Host working directory (ro or rw based on policy) │ +//! │ /output ─▶ Output directory for artifacts (rw) │ +//! │ │ +//! │ Limits: │ +//! │ Memory: 2GB (default) │ +//! │ CPU: 1024 shares │ +//! │ No privileged mode │ +//! │ Non-root user (UID 1000) │ +//! └────────────────────────────────────────────────────────────────────────┘ +//! ``` + +use std::collections::HashMap; +use std::path::Path; +use std::time::Duration; + +use bollard::Docker; +use bollard::container::{ + Config, CreateContainerOptions, LogOutput, LogsOptions, RemoveContainerOptions, + StartContainerOptions, WaitContainerOptions, +}; +use bollard::exec::{CreateExecOptions, StartExecResults}; +use bollard::models::HostConfig; +use futures::StreamExt; + +use crate::sandbox::config::{ResourceLimits, SandboxPolicy}; +use crate::sandbox::error::{Result, SandboxError}; + +/// Output from container execution. +#[derive(Debug, Clone)] +pub struct ContainerOutput { + /// Exit code from the command. + pub exit_code: i64, + /// Standard output. + pub stdout: String, + /// Standard error. + pub stderr: String, + /// How long the command ran. + pub duration: Duration, + /// Whether output was truncated. + pub truncated: bool, +} + +/// Manages Docker container lifecycle. +pub struct ContainerRunner { + docker: Docker, + image: String, + proxy_port: u16, +} + +impl ContainerRunner { + /// Create a new container runner. + pub fn new(docker: Docker, image: String, proxy_port: u16) -> Self { + Self { + docker, + image, + proxy_port, + } + } + + /// Check if the Docker daemon is available. + pub async fn is_available(&self) -> bool { + self.docker.ping().await.is_ok() + } + + /// Check if the sandbox image exists locally. + pub async fn image_exists(&self) -> bool { + self.docker.inspect_image(&self.image).await.is_ok() + } + + /// Pull the sandbox image. + pub async fn pull_image(&self) -> Result<()> { + use bollard::image::CreateImageOptions; + + tracing::info!("Pulling sandbox image: {}", self.image); + + let options = CreateImageOptions { + from_image: self.image.clone(), + ..Default::default() + }; + + let mut stream = self.docker.create_image(Some(options), None, None); + + while let Some(result) = stream.next().await { + match result { + Ok(info) => { + if let Some(status) = info.status { + tracing::debug!("Pull status: {}", status); + } + } + Err(e) => { + return Err(SandboxError::ContainerCreationFailed { + reason: format!("image pull failed: {}", e), + }); + } + } + } + + tracing::info!("Successfully pulled image: {}", self.image); + Ok(()) + } + + /// Execute a command in a new container. + pub async fn execute( + &self, + command: &str, + working_dir: &Path, + policy: SandboxPolicy, + limits: &ResourceLimits, + env: HashMap, + ) -> Result { + let start_time = std::time::Instant::now(); + + // Create the container + let container_id = self + .create_container(command, working_dir, policy, limits, env) + .await?; + + // Start the container + self.docker + .start_container(&container_id, None::>) + .await + .map_err(|e| SandboxError::ContainerStartFailed { + reason: e.to_string(), + })?; + + // Wait for completion with timeout + let result = tokio::time::timeout(limits.timeout, async { + self.wait_for_container(&container_id, limits.max_output_bytes) + .await + }) + .await; + + // Always clean up the container + let _ = self + .docker + .remove_container( + &container_id, + Some(RemoveContainerOptions { + force: true, + ..Default::default() + }), + ) + .await; + + match result { + Ok(Ok(mut output)) => { + output.duration = start_time.elapsed(); + Ok(output) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(SandboxError::Timeout(limits.timeout)), + } + } + + /// Execute a command in an existing container using exec. + pub async fn exec_in_container( + &self, + container_id: &str, + command: &str, + working_dir: &str, + limits: &ResourceLimits, + ) -> Result { + let start_time = std::time::Instant::now(); + + let exec = self + .docker + .create_exec( + container_id, + CreateExecOptions { + cmd: Some(vec!["sh", "-c", command]), + attach_stdout: Some(true), + attach_stderr: Some(true), + working_dir: Some(working_dir), + ..Default::default() + }, + ) + .await + .map_err(|e| SandboxError::ExecutionFailed { + reason: format!("exec create failed: {}", e), + })?; + + let result = tokio::time::timeout( + limits.timeout, + self.run_exec(&exec.id, limits.max_output_bytes), + ) + .await; + + match result { + Ok(Ok(mut output)) => { + output.duration = start_time.elapsed(); + Ok(output) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(SandboxError::Timeout(limits.timeout)), + } + } + + /// Create a container with the appropriate configuration. + async fn create_container( + &self, + command: &str, + working_dir: &Path, + policy: SandboxPolicy, + limits: &ResourceLimits, + env: HashMap, + ) -> Result { + let working_dir_str = working_dir.display().to_string(); + + // Build environment variables + let mut env_vec: Vec = env + .into_iter() + .map(|(k, v)| format!("{}={}", k, v)) + .collect(); + + // Add proxy environment (uses host.docker.internal for Mac/Windows, 172.17.0.1 for Linux) + let proxy_host = if cfg!(target_os = "linux") { + "172.17.0.1" + } else { + "host.docker.internal" + }; + + if self.proxy_port > 0 && policy.is_sandboxed() { + env_vec.push(format!( + "http_proxy=http://{}:{}", + proxy_host, self.proxy_port + )); + env_vec.push(format!( + "https_proxy=http://{}:{}", + proxy_host, self.proxy_port + )); + env_vec.push(format!( + "HTTP_PROXY=http://{}:{}", + proxy_host, self.proxy_port + )); + env_vec.push(format!( + "HTTPS_PROXY=http://{}:{}", + proxy_host, self.proxy_port + )); + } + + // Build volume mounts based on policy + let binds = match policy { + SandboxPolicy::ReadOnly => { + vec![format!("{}:/workspace:ro", working_dir_str)] + } + SandboxPolicy::WorkspaceWrite => { + vec![format!("{}:/workspace:rw", working_dir_str)] + } + SandboxPolicy::FullAccess => { + // Full access - mount more of the host + vec![ + format!("{}:/workspace:rw", working_dir_str), + "/tmp:/tmp:rw".to_string(), + ] + } + }; + + let host_config = HostConfig { + binds: Some(binds), + memory: Some((limits.memory_bytes) as i64), + cpu_shares: Some(limits.cpu_shares as i64), + auto_remove: Some(true), + network_mode: Some("bridge".to_string()), + // Security: drop all capabilities and add back only what's needed + cap_drop: Some(vec!["ALL".to_string()]), + cap_add: Some(vec![ + "CHOWN".to_string(), + "SETUID".to_string(), + "SETGID".to_string(), + ]), + // Prevent privilege escalation + security_opt: Some(vec!["no-new-privileges:true".to_string()]), + // Read-only root filesystem (workspace is still writable if policy allows) + readonly_rootfs: Some(policy == SandboxPolicy::ReadOnly), + // Tmpfs mounts for /tmp and cargo cache + tmpfs: Some( + [ + ("/tmp".to_string(), "size=512M".to_string()), + ( + "/home/sandbox/.cargo/registry".to_string(), + "size=1G".to_string(), + ), + ] + .into_iter() + .collect(), + ), + ..Default::default() + }; + + let config = Config { + image: Some(self.image.clone()), + cmd: Some(vec![ + "sh".to_string(), + "-c".to_string(), + command.to_string(), + ]), + working_dir: Some("/workspace".to_string()), + env: Some(env_vec), + host_config: Some(host_config), + user: Some("1000:1000".to_string()), // Non-root user + ..Default::default() + }; + + let options = CreateContainerOptions { + name: format!("sandbox-{}", uuid::Uuid::new_v4()), + ..Default::default() + }; + + let response = self + .docker + .create_container(Some(options), config) + .await + .map_err(|e| SandboxError::ContainerCreationFailed { + reason: e.to_string(), + })?; + + Ok(response.id) + } + + /// Wait for a container to complete and collect output. + async fn wait_for_container( + &self, + container_id: &str, + max_output: usize, + ) -> Result { + // Wait for the container to finish + let mut wait_stream = self.docker.wait_container( + container_id, + Some(WaitContainerOptions { + condition: "not-running", + }), + ); + + let exit_code = match wait_stream.next().await { + Some(Ok(response)) => response.status_code, + Some(Err(e)) => { + return Err(SandboxError::ExecutionFailed { + reason: format!("wait failed: {}", e), + }); + } + None => { + return Err(SandboxError::ExecutionFailed { + reason: "container wait stream ended unexpectedly".to_string(), + }); + } + }; + + // Collect logs + let (stdout, stderr, truncated) = self.collect_logs(container_id, max_output).await?; + + Ok(ContainerOutput { + exit_code, + stdout, + stderr, + duration: Duration::ZERO, // Will be set by caller + truncated, + }) + } + + /// Collect stdout and stderr from a container. + async fn collect_logs( + &self, + container_id: &str, + max_output: usize, + ) -> Result<(String, String, bool)> { + let options = LogsOptions:: { + stdout: true, + stderr: true, + follow: false, + ..Default::default() + }; + + let mut stream = self.docker.logs(container_id, Some(options)); + + let mut stdout = String::new(); + let mut stderr = String::new(); + let mut truncated = false; + let half_max = max_output / 2; + + while let Some(result) = stream.next().await { + match result { + Ok(LogOutput::StdOut { message }) => { + let text = String::from_utf8_lossy(&message); + if stdout.len() + text.len() > half_max { + truncated = true; + let remaining = half_max.saturating_sub(stdout.len()); + stdout.push_str(&text[..remaining.min(text.len())]); + } else { + stdout.push_str(&text); + } + } + Ok(LogOutput::StdErr { message }) => { + let text = String::from_utf8_lossy(&message); + if stderr.len() + text.len() > half_max { + truncated = true; + let remaining = half_max.saturating_sub(stderr.len()); + stderr.push_str(&text[..remaining.min(text.len())]); + } else { + stderr.push_str(&text); + } + } + Ok(_) => {} + Err(e) => { + tracing::warn!("Error reading container logs: {}", e); + } + } + } + + Ok((stdout, stderr, truncated)) + } + + /// Run an exec and collect output. + async fn run_exec(&self, exec_id: &str, max_output: usize) -> Result { + let start_result = self.docker.start_exec(exec_id, None).await.map_err(|e| { + SandboxError::ExecutionFailed { + reason: format!("exec start failed: {}", e), + } + })?; + + let mut stdout = String::new(); + let mut stderr = String::new(); + let mut truncated = false; + let half_max = max_output / 2; + + if let StartExecResults::Attached { mut output, .. } = start_result { + while let Some(result) = output.next().await { + match result { + Ok(LogOutput::StdOut { message }) => { + let text = String::from_utf8_lossy(&message); + if stdout.len() < half_max { + let remaining = half_max.saturating_sub(stdout.len()); + stdout.push_str(&text[..remaining.min(text.len())]); + if text.len() > remaining { + truncated = true; + } + } + } + Ok(LogOutput::StdErr { message }) => { + let text = String::from_utf8_lossy(&message); + if stderr.len() < half_max { + let remaining = half_max.saturating_sub(stderr.len()); + stderr.push_str(&text[..remaining.min(text.len())]); + if text.len() > remaining { + truncated = true; + } + } + } + Ok(_) => {} + Err(e) => { + tracing::warn!("Error reading exec output: {}", e); + } + } + } + } + + // Get exec exit code + let inspect = + self.docker + .inspect_exec(exec_id) + .await + .map_err(|e| SandboxError::ExecutionFailed { + reason: format!("exec inspect failed: {}", e), + })?; + + let exit_code = inspect.exit_code.unwrap_or(-1); + + Ok(ContainerOutput { + exit_code, + stdout, + stderr, + duration: Duration::ZERO, + truncated, + }) + } +} + +/// Connect to the Docker daemon. +pub async fn connect_docker() -> Result { + Docker::connect_with_local_defaults().map_err(|e| SandboxError::DockerNotAvailable { + reason: e.to_string(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_docker_connection() { + // This test requires Docker to be running + let result = connect_docker().await; + // Don't fail if Docker isn't available, just skip + if result.is_err() { + eprintln!("Skipping Docker test: Docker not available"); + return; + } + + let docker = result.unwrap(); + let runner = ContainerRunner::new(docker, "alpine:latest".to_string(), 0); + // Just check that we can query Docker (result doesn't matter for CI) + let _available = runner.is_available().await; + } +} diff --git a/src/sandbox/error.rs b/src/sandbox/error.rs new file mode 100644 index 00000000..f9bd4fff --- /dev/null +++ b/src/sandbox/error.rs @@ -0,0 +1,58 @@ +//! Error types for the Docker execution sandbox. + +use std::time::Duration; + +/// Errors that can occur in the sandbox system. +#[derive(Debug, thiserror::Error)] +pub enum SandboxError { + /// Docker daemon is not available or not running. + #[error("Docker not available: {reason}")] + DockerNotAvailable { reason: String }, + + /// Failed to create container. + #[error("Container creation failed: {reason}")] + ContainerCreationFailed { reason: String }, + + /// Failed to start container. + #[error("Container start failed: {reason}")] + ContainerStartFailed { reason: String }, + + /// Command execution failed inside container. + #[error("Execution failed: {reason}")] + ExecutionFailed { reason: String }, + + /// Command timed out. + #[error("Command timed out after {0:?}")] + Timeout(Duration), + + /// Container resource limit exceeded. + #[error("Resource limit exceeded: {resource} limit of {limit}")] + ResourceLimitExceeded { resource: String, limit: String }, + + /// Network proxy error. + #[error("Proxy error: {reason}")] + ProxyError { reason: String }, + + /// Network request blocked by policy. + #[error("Network request blocked: {reason}")] + NetworkBlocked { reason: String }, + + /// Credential injection failed. + #[error("Credential injection failed for {domain}: {reason}")] + CredentialInjectionFailed { domain: String, reason: String }, + + /// Docker API error. + #[error("Docker API error: {0}")] + Docker(#[from] bollard::errors::Error), + + /// I/O error. + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + + /// Configuration error. + #[error("Configuration error: {reason}")] + Config { reason: String }, +} + +/// Result type for sandbox operations. +pub type Result = std::result::Result; diff --git a/src/sandbox/manager.rs b/src/sandbox/manager.rs new file mode 100644 index 00000000..3b07eafe --- /dev/null +++ b/src/sandbox/manager.rs @@ -0,0 +1,474 @@ +//! Main sandbox manager coordinating proxy and containers. +//! +//! The `SandboxManager` is the primary entry point for sandboxed execution. +//! It coordinates: +//! - Docker container creation and lifecycle +//! - HTTP proxy for network access control +//! - Credential injection for API calls +//! - Resource limits and timeouts +//! +//! # Architecture +//! +//! ```text +//! ┌───────────────────────────────────────────────────────────────────────────┐ +//! │ SandboxManager │ +//! │ │ +//! │ execute(cmd, cwd, policy) │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────────┐ │ +//! │ │ Start Proxy │────▶│ Create │────▶│ Execute & Collect Output │ │ +//! │ │ (if needed) │ │ Container │ │ │ │ +//! │ └──────────────┘ └──────────────┘ └──────────────────────────┘ │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌──────────────────────────┐ │ +//! │ │ Cleanup Container │ │ +//! │ └──────────────────────────┘ │ +//! └───────────────────────────────────────────────────────────────────────────┘ +//! ``` + +use std::collections::HashMap; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::RwLock; + +use crate::sandbox::config::{ResourceLimits, SandboxConfig, SandboxPolicy}; +use crate::sandbox::container::{ContainerOutput, ContainerRunner, connect_docker}; +use crate::sandbox::error::{Result, SandboxError}; +use crate::sandbox::proxy::{HttpProxy, NetworkProxyBuilder}; + +/// Output from sandbox execution. +#[derive(Debug, Clone)] +pub struct ExecOutput { + /// Exit code from the command. + pub exit_code: i64, + /// Standard output. + pub stdout: String, + /// Standard error. + pub stderr: String, + /// Combined output (stdout + stderr). + pub output: String, + /// How long the command ran. + pub duration: Duration, + /// Whether output was truncated. + pub truncated: bool, +} + +impl From for ExecOutput { + fn from(c: ContainerOutput) -> Self { + let output = if c.stderr.is_empty() { + c.stdout.clone() + } else if c.stdout.is_empty() { + c.stderr.clone() + } else { + format!("{}\n\n--- stderr ---\n{}", c.stdout, c.stderr) + }; + + Self { + exit_code: c.exit_code, + stdout: c.stdout, + stderr: c.stderr, + output, + duration: c.duration, + truncated: c.truncated, + } + } +} + +/// Main sandbox manager. +pub struct SandboxManager { + config: SandboxConfig, + proxy: Arc>>, + runner: Arc>>, + initialized: std::sync::atomic::AtomicBool, +} + +impl SandboxManager { + /// Create a new sandbox manager. + pub fn new(config: SandboxConfig) -> Self { + Self { + config, + proxy: Arc::new(RwLock::new(None)), + runner: Arc::new(RwLock::new(None)), + initialized: std::sync::atomic::AtomicBool::new(false), + } + } + + /// Create with default configuration. + pub fn with_defaults() -> Self { + Self::new(SandboxConfig::default()) + } + + /// Check if the sandbox is available (Docker running, etc.). + pub async fn is_available(&self) -> bool { + if !self.config.enabled { + return false; + } + + match connect_docker().await { + Ok(docker) => docker.ping().await.is_ok(), + Err(_) => false, + } + } + + /// Initialize the sandbox (connect to Docker, start proxy). + pub async fn initialize(&self) -> Result<()> { + if self.initialized.load(std::sync::atomic::Ordering::SeqCst) { + return Ok(()); + } + + if !self.config.enabled { + return Err(SandboxError::Config { + reason: "sandbox is disabled".to_string(), + }); + } + + // Connect to Docker + let docker = connect_docker().await?; + + // Check if Docker is responsive + docker + .ping() + .await + .map_err(|e| SandboxError::DockerNotAvailable { + reason: e.to_string(), + })?; + + // Create container runner + let runner = + ContainerRunner::new(docker, self.config.image.clone(), self.config.proxy_port); + + // Check for / pull image + if !runner.image_exists().await { + if self.config.auto_pull_image { + runner.pull_image().await?; + } else { + return Err(SandboxError::ContainerCreationFailed { + reason: format!( + "image {} not found and auto_pull is disabled", + self.config.image + ), + }); + } + } + + *self.runner.write().await = Some(runner); + + // Start the network proxy if we're using a sandboxed policy + if self.config.policy.is_sandboxed() { + let proxy = NetworkProxyBuilder::from_config(&self.config) + .build_and_start(self.config.proxy_port) + .await?; + + *self.proxy.write().await = Some(proxy); + } + + self.initialized + .store(true, std::sync::atomic::Ordering::SeqCst); + + tracing::info!("Sandbox initialized"); + Ok(()) + } + + /// Shutdown the sandbox (stop proxy, clean up). + pub async fn shutdown(&self) { + if let Some(proxy) = self.proxy.write().await.take() { + proxy.stop().await; + } + + self.initialized + .store(false, std::sync::atomic::Ordering::SeqCst); + + tracing::info!("Sandbox shut down"); + } + + /// Execute a command in the sandbox. + pub async fn execute( + &self, + command: &str, + cwd: &Path, + env: HashMap, + ) -> Result { + self.execute_with_policy(command, cwd, self.config.policy, env) + .await + } + + /// Execute a command with a specific policy. + pub async fn execute_with_policy( + &self, + command: &str, + cwd: &Path, + policy: SandboxPolicy, + env: HashMap, + ) -> Result { + // FullAccess policy bypasses the sandbox entirely + if policy == SandboxPolicy::FullAccess { + return self.execute_direct(command, cwd, env).await; + } + + // Ensure we're initialized + if !self.initialized.load(std::sync::atomic::Ordering::SeqCst) { + self.initialize().await?; + } + + // Get proxy port if running + let proxy_port = if let Some(proxy) = self.proxy.read().await.as_ref() { + proxy.addr().await.map(|a| a.port()).unwrap_or(0) + } else { + 0 + }; + + // Create a runner with the current proxy port + let docker = connect_docker().await?; + let runner = ContainerRunner::new(docker, self.config.image.clone(), proxy_port); + + let limits = ResourceLimits { + memory_bytes: self.config.memory_limit_mb * 1024 * 1024, + cpu_shares: self.config.cpu_shares, + timeout: self.config.timeout, + max_output_bytes: 64 * 1024, + }; + + let container_output = runner.execute(command, cwd, policy, &limits, env).await?; + + Ok(container_output.into()) + } + + /// Execute a command directly on the host (no sandbox). + async fn execute_direct( + &self, + command: &str, + cwd: &Path, + env: HashMap, + ) -> Result { + use tokio::process::Command; + + let start = std::time::Instant::now(); + + let mut cmd = if cfg!(target_os = "windows") { + let mut c = Command::new("cmd"); + c.args(["/C", command]); + c + } else { + let mut c = Command::new("sh"); + c.args(["-c", command]); + c + }; + + cmd.current_dir(cwd); + cmd.envs(env); + + let output = tokio::time::timeout(self.config.timeout, cmd.output()) + .await + .map_err(|_| SandboxError::Timeout(self.config.timeout))? + .map_err(|e| SandboxError::ExecutionFailed { + reason: e.to_string(), + })?; + + let stdout = String::from_utf8_lossy(&output.stdout).to_string(); + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + let combined = if stderr.is_empty() { + stdout.clone() + } else if stdout.is_empty() { + stderr.clone() + } else { + format!("{}\n\n--- stderr ---\n{}", stdout, stderr) + }; + + Ok(ExecOutput { + exit_code: output.status.code().unwrap_or(-1) as i64, + stdout, + stderr, + output: combined, + duration: start.elapsed(), + truncated: false, + }) + } + + /// Execute a build command (convenience method using WorkspaceWrite policy). + pub async fn build( + &self, + command: &str, + project_dir: &Path, + env: HashMap, + ) -> Result { + self.execute_with_policy(command, project_dir, SandboxPolicy::WorkspaceWrite, env) + .await + } + + /// Get the current configuration. + pub fn config(&self) -> &SandboxConfig { + &self.config + } + + /// Check if the sandbox is initialized. + pub fn is_initialized(&self) -> bool { + self.initialized.load(std::sync::atomic::Ordering::SeqCst) + } + + /// Get the proxy port if running. + pub async fn proxy_port(&self) -> Option { + if let Some(proxy) = self.proxy.read().await.as_ref() { + proxy.addr().await.map(|a| a.port()) + } else { + None + } + } +} + +impl Drop for SandboxManager { + fn drop(&mut self) { + // Note: async cleanup should be done via shutdown() before dropping + if self.initialized.load(std::sync::atomic::Ordering::SeqCst) { + tracing::warn!("SandboxManager dropped without shutdown(), resources may leak"); + } + } +} + +/// Builder for creating a sandbox manager. +pub struct SandboxManagerBuilder { + config: SandboxConfig, +} + +impl SandboxManagerBuilder { + /// Create a new builder. + pub fn new() -> Self { + Self { + config: SandboxConfig::default(), + } + } + + /// Enable the sandbox. + pub fn enabled(mut self, enabled: bool) -> Self { + self.config.enabled = enabled; + self + } + + /// Set the sandbox policy. + pub fn policy(mut self, policy: SandboxPolicy) -> Self { + self.config.policy = policy; + self + } + + /// Set the command timeout. + pub fn timeout(mut self, timeout: Duration) -> Self { + self.config.timeout = timeout; + self + } + + /// Set the memory limit in MB. + pub fn memory_limit_mb(mut self, mb: u64) -> Self { + self.config.memory_limit_mb = mb; + self + } + + /// Set the Docker image. + pub fn image(mut self, image: &str) -> Self { + self.config.image = image.to_string(); + self + } + + /// Add domains to the network allowlist. + pub fn allow_domains(mut self, domains: Vec) -> Self { + self.config.network_allowlist.extend(domains); + self + } + + /// Build the sandbox manager. + pub fn build(self) -> SandboxManager { + SandboxManager::new(self.config) + } + + /// Build and initialize the sandbox manager. + pub async fn build_and_init(self) -> Result { + let manager = self.build(); + manager.initialize().await?; + Ok(manager) + } +} + +impl Default for SandboxManagerBuilder { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_exec_output_from_container_output() { + let container = ContainerOutput { + exit_code: 0, + stdout: "hello".to_string(), + stderr: String::new(), + duration: Duration::from_secs(1), + truncated: false, + }; + + let exec: ExecOutput = container.into(); + assert_eq!(exec.exit_code, 0); + assert_eq!(exec.output, "hello"); + } + + #[test] + fn test_exec_output_combined() { + let container = ContainerOutput { + exit_code: 1, + stdout: "out".to_string(), + stderr: "err".to_string(), + duration: Duration::from_secs(1), + truncated: false, + }; + + let exec: ExecOutput = container.into(); + assert!(exec.output.contains("out")); + assert!(exec.output.contains("err")); + assert!(exec.output.contains("stderr")); + } + + #[test] + fn test_builder_defaults() { + let manager = SandboxManagerBuilder::new().build(); + assert!(!manager.config.enabled); // Disabled by default + } + + #[test] + fn test_builder_custom() { + let manager = SandboxManagerBuilder::new() + .enabled(true) + .policy(SandboxPolicy::WorkspaceWrite) + .timeout(Duration::from_secs(60)) + .memory_limit_mb(1024) + .image("custom:latest") + .build(); + + assert!(manager.config.enabled); + assert_eq!(manager.config.policy, SandboxPolicy::WorkspaceWrite); + assert_eq!(manager.config.timeout, Duration::from_secs(60)); + assert_eq!(manager.config.memory_limit_mb, 1024); + assert_eq!(manager.config.image, "custom:latest"); + } + + #[tokio::test] + async fn test_direct_execution() { + let manager = SandboxManager::new(SandboxConfig { + enabled: true, + policy: SandboxPolicy::FullAccess, + ..Default::default() + }); + + let result = manager + .execute("echo hello", Path::new("."), HashMap::new()) + .await; + + // This should work even without Docker since FullAccess runs directly + assert!(result.is_ok()); + let output = result.unwrap(); + assert!(output.stdout.contains("hello")); + } +} diff --git a/src/sandbox/mod.rs b/src/sandbox/mod.rs new file mode 100644 index 00000000..ca6e9f9f --- /dev/null +++ b/src/sandbox/mod.rs @@ -0,0 +1,113 @@ +//! Docker execution sandbox for secure command execution. +//! +//! This module provides a complete sandboxing solution for running untrusted commands: +//! - **Container isolation**: Commands run in ephemeral Docker containers +//! - **Network proxy**: All network traffic goes through a validating proxy +//! - **Credential injection**: Secrets are injected by the proxy, never exposed in containers +//! - **Resource limits**: Memory, CPU, and timeout enforcement +//! +//! # Architecture +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────────────────────┐ +//! │ Sandbox System │ +//! │ │ +//! │ ┌─────────────────────────────────────────────────────────────────────┐ │ +//! │ │ SandboxManager │ │ +//! │ │ │ │ +//! │ │ • Coordinates container creation and execution │ │ +//! │ │ • Manages proxy lifecycle │ │ +//! │ │ • Enforces resource limits │ │ +//! │ └─────────────────────────────────────────────────────────────────────┘ │ +//! │ │ │ │ +//! │ ▼ ▼ │ +//! │ ┌──────────────────┐ ┌───────────────────┐ │ +//! │ │ Container │ │ Network Proxy │ │ +//! │ │ Runner │ │ │ │ +//! │ │ │ │ • Allowlist │ │ +//! │ │ • Create │◀────────▶│ • Credentials │ │ +//! │ │ • Execute │ │ • Logging │ │ +//! │ │ • Cleanup │ │ │ │ +//! │ └──────────────────┘ └───────────────────┘ │ +//! │ │ │ │ +//! │ ▼ ▼ │ +//! │ ┌──────────────────┐ ┌───────────────────┐ │ +//! │ │ Docker │ │ Internet │ │ +//! │ │ │ │ (allowed hosts) │ │ +//! │ └──────────────────┘ └───────────────────┘ │ +//! └─────────────────────────────────────────────────────────────────────────────┘ +//! ``` +//! +//! # Sandbox Policies +//! +//! | Policy | Filesystem | Network | Use Case | +//! |--------|------------|---------|----------| +//! | `ReadOnly` | Read workspace | Proxied | Explore code, fetch docs | +//! | `WorkspaceWrite` | Read/write workspace | Proxied | Build software, run tests | +//! | `FullAccess` | Full host | Full | Direct execution (no sandbox) | +//! +//! # Example +//! +//! ```rust,no_run +//! use near_agent::sandbox::{SandboxManager, SandboxManagerBuilder, SandboxPolicy}; +//! use std::collections::HashMap; +//! use std::path::Path; +//! +//! # async fn example() -> Result<(), Box> { +//! let manager = SandboxManagerBuilder::new() +//! .enabled(true) +//! .policy(SandboxPolicy::WorkspaceWrite) +//! .build(); +//! +//! manager.initialize().await?; +//! +//! let result = manager.execute( +//! "cargo build --release", +//! Path::new("/workspace/my-project"), +//! HashMap::new(), +//! ).await?; +//! +//! println!("Exit code: {}", result.exit_code); +//! println!("Output: {}", result.output); +//! +//! manager.shutdown().await; +//! # Ok(()) +//! # } +//! ``` +//! +//! # Security Properties +//! +//! - **No credentials in containers**: Environment variables with secrets never enter containers +//! - **Network isolation**: All traffic routes through the proxy (validated domains only) +//! - **Non-root execution**: Containers run as UID 1000 +//! - **Read-only root**: Container filesystem is read-only (except workspace mount) +//! - **Capability dropping**: All Linux capabilities dropped, only essential ones added back +//! - **Auto-cleanup**: Containers are removed after execution (--rm + explicit cleanup) +//! - **Timeout enforcement**: Commands are killed after the timeout + +pub mod config; +pub mod container; +pub mod error; +pub mod manager; +pub mod proxy; + +pub use config::{ + CredentialLocation, CredentialMapping, ResourceLimits, SandboxConfig, SandboxPolicy, +}; +pub use container::{ContainerOutput, ContainerRunner, connect_docker}; +pub use error::{Result, SandboxError}; +pub use manager::{ExecOutput, SandboxManager, SandboxManagerBuilder}; +pub use proxy::{ + CredentialResolver, DefaultPolicyDecider, DomainAllowlist, EnvCredentialResolver, HttpProxy, + NetworkDecision, NetworkPolicyDecider, NetworkProxyBuilder, NetworkRequest, +}; + +/// Default allowlist getter (re-export for convenience). +pub fn default_allowlist() -> Vec { + config::default_allowlist() +} + +/// Default credential mappings getter (re-export for convenience). +pub fn default_credential_mappings() -> Vec { + config::default_credential_mappings() +} diff --git a/src/sandbox/proxy/allowlist.rs b/src/sandbox/proxy/allowlist.rs new file mode 100644 index 00000000..207aa272 --- /dev/null +++ b/src/sandbox/proxy/allowlist.rs @@ -0,0 +1,251 @@ +//! Domain allowlist for the network proxy. +//! +//! Validates that HTTP requests only go to allowed domains. +//! Supports exact matches and wildcard patterns. + +use std::fmt; + +/// Pattern for matching allowed domains. +#[derive(Debug, Clone)] +pub struct DomainPattern { + /// The domain pattern (e.g., "api.example.com" or "*.example.com"). + pattern: String, + /// Whether this is a wildcard pattern. + is_wildcard: bool, + /// The base domain for wildcard matching. + base_domain: String, +} + +impl DomainPattern { + /// Create a new domain pattern. + pub fn new(pattern: &str) -> Self { + let is_wildcard = pattern.starts_with("*."); + let base_domain = if is_wildcard { + pattern[2..].to_lowercase() + } else { + pattern.to_lowercase() + }; + + Self { + pattern: pattern.to_string(), + is_wildcard, + base_domain, + } + } + + /// Check if a host matches this pattern. + pub fn matches(&self, host: &str) -> bool { + let host_lower = host.to_lowercase(); + + if self.is_wildcard { + // *.example.com matches foo.example.com, bar.baz.example.com, example.com + host_lower == self.base_domain + || host_lower.ends_with(&format!(".{}", self.base_domain)) + } else { + host_lower == self.base_domain + } + } + + /// Get the pattern string. + pub fn pattern(&self) -> &str { + &self.pattern + } +} + +impl fmt::Display for DomainPattern { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.pattern) + } +} + +/// Result of domain validation. +#[derive(Debug, Clone)] +pub enum DomainValidationResult { + /// Domain is allowed. + Allowed, + /// Domain is denied with a reason. + Denied(String), +} + +impl DomainValidationResult { + pub fn is_allowed(&self) -> bool { + matches!(self, DomainValidationResult::Allowed) + } +} + +/// Validates domains against an allowlist. +#[derive(Debug, Clone)] +pub struct DomainAllowlist { + patterns: Vec, +} + +impl DomainAllowlist { + /// Create a new allowlist from domain strings. + pub fn new(domains: &[String]) -> Self { + Self { + patterns: domains.iter().map(|d| DomainPattern::new(d)).collect(), + } + } + + /// Create an empty allowlist (denies everything). + pub fn empty() -> Self { + Self { patterns: vec![] } + } + + /// Add a domain pattern to the allowlist. + pub fn add(&mut self, pattern: &str) { + self.patterns.push(DomainPattern::new(pattern)); + } + + /// Check if a domain is allowed. + pub fn is_allowed(&self, host: &str) -> DomainValidationResult { + if self.patterns.is_empty() { + return DomainValidationResult::Denied("empty allowlist".to_string()); + } + + for pattern in &self.patterns { + if pattern.matches(host) { + return DomainValidationResult::Allowed; + } + } + + DomainValidationResult::Denied(format!( + "host '{}' not in allowlist: [{}]", + host, + self.patterns + .iter() + .map(|p| p.pattern()) + .collect::>() + .join(", ") + )) + } + + /// Get all patterns in the allowlist. + pub fn patterns(&self) -> &[DomainPattern] { + &self.patterns + } + + /// Check if the allowlist is empty. + pub fn is_empty(&self) -> bool { + self.patterns.is_empty() + } + + /// Get the number of patterns. + pub fn len(&self) -> usize { + self.patterns.len() + } +} + +impl Default for DomainAllowlist { + fn default() -> Self { + Self::new(&crate::sandbox::config::default_allowlist()) + } +} + +/// Parse host from a URL string. +pub fn extract_host(url: &str) -> Option { + // Determine scheme and extract the rest + let rest = if let Some(stripped) = url.strip_prefix("https://") { + stripped + } else if let Some(stripped) = url.strip_prefix("http://") { + stripped + } else { + return None; + }; + + // Find the end of the host (start of path, query, or end of string) + let host_end = rest.find('/').unwrap_or(rest.len()); + let host_and_port = &rest[..host_end]; + + // Remove port if present + let host = if let Some(bracket_idx) = host_and_port.find('[') { + // IPv6 address + let close_bracket = host_and_port.find(']')?; + &host_and_port[bracket_idx + 1..close_bracket] + } else if let Some(colon_idx) = host_and_port.rfind(':') { + // Check if this is a port (all digits after colon) + let after_colon = &host_and_port[colon_idx + 1..]; + if after_colon.chars().all(|c| c.is_ascii_digit()) { + &host_and_port[..colon_idx] + } else { + host_and_port + } + } else { + host_and_port + }; + + if host.is_empty() { + None + } else { + Some(host.to_lowercase()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_exact_match() { + let pattern = DomainPattern::new("api.example.com"); + assert!(pattern.matches("api.example.com")); + assert!(pattern.matches("API.EXAMPLE.COM")); + assert!(!pattern.matches("foo.api.example.com")); + assert!(!pattern.matches("example.com")); + } + + #[test] + fn test_wildcard_match() { + let pattern = DomainPattern::new("*.example.com"); + assert!(pattern.matches("api.example.com")); + assert!(pattern.matches("foo.bar.example.com")); + assert!(pattern.matches("example.com")); // Base domain also matches + assert!(!pattern.matches("exampleXcom")); + assert!(!pattern.matches("other.com")); + } + + #[test] + fn test_allowlist_allows() { + let allowlist = + DomainAllowlist::new(&["crates.io".to_string(), "*.github.com".to_string()]); + + assert!(allowlist.is_allowed("crates.io").is_allowed()); + assert!(allowlist.is_allowed("api.github.com").is_allowed()); + assert!( + !allowlist + .is_allowed("raw.githubusercontent.com") + .is_allowed() + ); + } + + #[test] + fn test_allowlist_denies() { + let allowlist = DomainAllowlist::new(&["crates.io".to_string()]); + + let result = allowlist.is_allowed("evil.com"); + assert!(!result.is_allowed()); + } + + #[test] + fn test_empty_allowlist() { + let allowlist = DomainAllowlist::empty(); + assert!(!allowlist.is_allowed("anything.com").is_allowed()); + } + + #[test] + fn test_extract_host() { + assert_eq!( + extract_host("https://api.example.com/v1/endpoint"), + Some("api.example.com".to_string()) + ); + assert_eq!( + extract_host("http://localhost:8080/api"), + Some("localhost".to_string()) + ); + assert_eq!( + extract_host("https://EXAMPLE.COM"), + Some("example.com".to_string()) + ); + assert_eq!(extract_host("not-a-url"), None); + } +} diff --git a/src/sandbox/proxy/http.rs b/src/sandbox/proxy/http.rs new file mode 100644 index 00000000..6b0a3e82 --- /dev/null +++ b/src/sandbox/proxy/http.rs @@ -0,0 +1,444 @@ +//! HTTP proxy server for sandboxed network access. +//! +//! This proxy runs on the host and handles all network requests from containers. +//! It validates requests against the allowlist and injects credentials when needed. +//! +//! ```text +//! Container ──► http_proxy=host.docker.internal:PORT ──► This Proxy ──► Internet +//! │ +//! ├─► Validate domain +//! ├─► Inject credentials +//! └─► Log requests +//! ``` + +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::Arc; + +use bytes::Bytes; +use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody}; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::{Method, Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use tokio::net::TcpListener; +use tokio::sync::RwLock; + +use crate::sandbox::config::CredentialLocation; +use crate::sandbox::error::{Result, SandboxError}; +use crate::sandbox::proxy::policy::{NetworkDecision, NetworkPolicyDecider, NetworkRequest}; + +/// State shared across proxy connections. +struct ProxyState { + /// Policy decider for network requests. + decider: Arc, + /// Credential resolver (maps secret names to values). + credential_resolver: Arc, + /// Request counter for logging. + request_count: std::sync::atomic::AtomicU64, + /// Whether the proxy is running. + running: std::sync::atomic::AtomicBool, +} + +/// Resolves secret names to their values. +#[async_trait::async_trait] +pub trait CredentialResolver: Send + Sync { + /// Get the value of a secret by name. + async fn resolve(&self, name: &str) -> Option; +} + +/// A credential resolver that uses environment variables. +pub struct EnvCredentialResolver; + +#[async_trait::async_trait] +impl CredentialResolver for EnvCredentialResolver { + async fn resolve(&self, name: &str) -> Option { + std::env::var(name).ok() + } +} + +/// A credential resolver that returns nothing (for testing). +pub struct NoCredentialResolver; + +#[async_trait::async_trait] +impl CredentialResolver for NoCredentialResolver { + async fn resolve(&self, _name: &str) -> Option { + None + } +} + +/// HTTP proxy server. +pub struct HttpProxy { + state: Arc, + addr: RwLock>, + shutdown_tx: RwLock>>, +} + +impl HttpProxy { + /// Create a new HTTP proxy. + pub fn new( + decider: Arc, + credential_resolver: Arc, + ) -> Self { + Self { + state: Arc::new(ProxyState { + decider, + credential_resolver, + request_count: std::sync::atomic::AtomicU64::new(0), + running: std::sync::atomic::AtomicBool::new(false), + }), + addr: RwLock::new(None), + shutdown_tx: RwLock::new(None), + } + } + + /// Start the proxy server on the given port (0 for auto-assign). + pub async fn start(&self, port: u16) -> Result { + let listener = TcpListener::bind(format!("127.0.0.1:{}", port)) + .await + .map_err(|e| SandboxError::ProxyError { + reason: format!("failed to bind: {}", e), + })?; + + let addr = listener + .local_addr() + .map_err(|e| SandboxError::ProxyError { + reason: format!("failed to get local addr: {}", e), + })?; + + *self.addr.write().await = Some(addr); + + let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel(); + *self.shutdown_tx.write().await = Some(shutdown_tx); + + self.state + .running + .store(true, std::sync::atomic::Ordering::SeqCst); + + let state = self.state.clone(); + + tokio::spawn(async move { + tracing::info!("Sandbox proxy started on {}", addr); + + loop { + tokio::select! { + accept_result = listener.accept() => { + match accept_result { + Ok((stream, _)) => { + let io = TokioIo::new(stream); + let state = state.clone(); + + tokio::spawn(async move { + let service = service_fn(move |req| { + let state = state.clone(); + async move { handle_request(req, state).await } + }); + + if let Err(e) = http1::Builder::new() + .preserve_header_case(true) + .title_case_headers(true) + .serve_connection(io, service) + .with_upgrades() + .await + { + tracing::debug!("Proxy connection error: {}", e); + } + }); + } + Err(e) => { + tracing::error!("Proxy accept error: {}", e); + } + } + } + _ = &mut shutdown_rx => { + tracing::info!("Sandbox proxy shutting down"); + break; + } + } + } + + state + .running + .store(false, std::sync::atomic::Ordering::SeqCst); + }); + + Ok(addr) + } + + /// Stop the proxy server. + pub async fn stop(&self) { + if let Some(tx) = self.shutdown_tx.write().await.take() { + let _ = tx.send(()); + } + } + + /// Get the address the proxy is listening on. + pub async fn addr(&self) -> Option { + *self.addr.read().await + } + + /// Check if the proxy is running. + pub fn is_running(&self) -> bool { + self.state.running.load(std::sync::atomic::Ordering::SeqCst) + } + + /// Get the number of requests handled. + pub fn request_count(&self) -> u64 { + self.state + .request_count + .load(std::sync::atomic::Ordering::SeqCst) + } +} + +/// Handle an incoming proxy request. +async fn handle_request( + req: Request, + state: Arc, +) -> std::result::Result>, Infallible> { + state + .request_count + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + + // Handle CONNECT method for HTTPS tunneling + if req.method() == Method::CONNECT { + return Ok(handle_connect(req, state).await); + } + + // For HTTP requests, validate and forward + let uri = req.uri().to_string(); + let method = req.method().to_string(); + + let network_req = match NetworkRequest::from_url(&method, &uri) { + Some(r) => r, + None => { + tracing::warn!("Proxy: invalid URL: {}", uri); + return Ok(error_response( + StatusCode::BAD_REQUEST, + "Invalid URL".to_string(), + )); + } + }; + + // Make policy decision + let decision = state.decider.decide(&network_req).await; + + match decision { + NetworkDecision::Deny { reason } => { + tracing::info!("Proxy: blocked {} {} - {}", method, uri, reason); + Ok(error_response(StatusCode::FORBIDDEN, reason)) + } + NetworkDecision::Allow | NetworkDecision::AllowWithCredentials { .. } => { + // Forward the request + forward_request(req, decision, state).await + } + } +} + +/// Handle CONNECT method for HTTPS tunneling. +async fn handle_connect( + req: Request, + state: Arc, +) -> Response> { + // Extract host from CONNECT target + let host = req.uri().authority().map(|a| a.host().to_string()); + + let host = match host { + Some(h) => h, + None => { + return error_response(StatusCode::BAD_REQUEST, "Missing host".to_string()); + } + }; + + // Check if host is allowed + let network_req = NetworkRequest { + method: "CONNECT".to_string(), + url: format!("https://{}", host), + host: host.clone(), + path: "/".to_string(), + }; + + let decision = state.decider.decide(&network_req).await; + + if !decision.is_allowed() { + if let NetworkDecision::Deny { reason } = decision { + tracing::info!("Proxy: blocked CONNECT {} - {}", host, reason); + return error_response(StatusCode::FORBIDDEN, reason); + } + } + + tracing::debug!("Proxy: allowing CONNECT to {}", host); + + // For CONNECT, we return 200 OK and the client will upgrade to TLS + // The actual TLS connection goes directly to the target, we just act as a tunnel + Response::builder() + .status(StatusCode::OK) + .body(empty_body()) + .unwrap() +} + +/// Forward a request to the target server. +async fn forward_request( + req: Request, + decision: NetworkDecision, + state: Arc, +) -> std::result::Result>, Infallible> { + let method = req.method().clone(); + let uri = req.uri().clone(); + + // Build the forwarded request + let client = reqwest::Client::new(); + let mut builder = client.request( + reqwest::Method::from_bytes(method.as_str().as_bytes()).unwrap_or(reqwest::Method::GET), + uri.to_string(), + ); + + // Copy headers (except hop-by-hop headers) + for (name, value) in req.headers() { + if !is_hop_by_hop_header(name.as_str()) { + if let Ok(v) = value.to_str() { + builder = builder.header(name.as_str(), v); + } + } + } + + // Inject credentials if needed + if let NetworkDecision::AllowWithCredentials { + secret_name, + location, + } = decision + { + if let Some(credential) = state.credential_resolver.resolve(&secret_name).await { + builder = match location { + CredentialLocation::AuthorizationBearer => { + builder.header("Authorization", format!("Bearer {}", credential)) + } + CredentialLocation::Header(header_name) => builder.header(header_name, credential), + CredentialLocation::QueryParam(param_name) => { + builder.query(&[(param_name, credential)]) + } + }; + tracing::debug!("Proxy: injected credential for {}", secret_name); + } else { + tracing::warn!("Proxy: credential {} not found", secret_name); + } + } + + // Copy body + let body_bytes = match req.collect().await { + Ok(collected) => collected.to_bytes(), + Err(e) => { + tracing::error!("Proxy: failed to read request body: {}", e); + return Ok(error_response( + StatusCode::INTERNAL_SERVER_ERROR, + "Failed to read body".to_string(), + )); + } + }; + + if !body_bytes.is_empty() { + builder = builder.body(body_bytes.to_vec()); + } + + // Send the request + match builder.send().await { + Ok(response) => { + let status = response.status(); + let headers = response.headers().clone(); + + match response.bytes().await { + Ok(body) => { + let mut builder = Response::builder().status(status.as_u16()); + + for (name, value) in headers.iter() { + if !is_hop_by_hop_header(name.as_str()) { + builder = builder.header(name.as_str(), value.as_bytes()); + } + } + + Ok(builder.body(full_body(body)).unwrap()) + } + Err(e) => { + tracing::error!("Proxy: failed to read response body: {}", e); + Ok(error_response( + StatusCode::BAD_GATEWAY, + "Failed to read response".to_string(), + )) + } + } + } + Err(e) => { + tracing::error!("Proxy: request failed: {}", e); + Ok(error_response( + StatusCode::BAD_GATEWAY, + format!("Request failed: {}", e), + )) + } + } +} + +/// Check if a header is hop-by-hop (should not be forwarded). +fn is_hop_by_hop_header(name: &str) -> bool { + matches!( + name.to_lowercase().as_str(), + "connection" + | "keep-alive" + | "proxy-authenticate" + | "proxy-authorization" + | "te" + | "trailers" + | "transfer-encoding" + | "upgrade" + ) +} + +/// Create an error response. +fn error_response(status: StatusCode, message: String) -> Response> { + Response::builder() + .status(status) + .header("Content-Type", "text/plain") + .body(full_body(Bytes::from(message))) + .unwrap() +} + +/// Create an empty body. +fn empty_body() -> BoxBody { + Empty::::new().map_err(|_| unreachable!()).boxed() +} + +/// Create a body from bytes. +fn full_body(bytes: Bytes) -> BoxBody { + Full::new(bytes).map_err(|_| unreachable!()).boxed() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sandbox::proxy::allowlist::DomainAllowlist; + use crate::sandbox::proxy::policy::DefaultPolicyDecider; + + #[tokio::test] + async fn test_proxy_starts_and_stops() { + let allowlist = DomainAllowlist::new(&["example.com".to_string()]); + let decider = Arc::new(DefaultPolicyDecider::new(allowlist, vec![])); + let resolver = Arc::new(NoCredentialResolver); + + let proxy = HttpProxy::new(decider, resolver); + + let addr = proxy.start(0).await.unwrap(); + assert!(proxy.is_running()); + assert!(addr.port() > 0); + + proxy.stop().await; + // Give it a moment to shut down + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + + #[test] + fn test_hop_by_hop_headers() { + assert!(is_hop_by_hop_header("connection")); + assert!(is_hop_by_hop_header("Connection")); + assert!(is_hop_by_hop_header("transfer-encoding")); + assert!(!is_hop_by_hop_header("content-type")); + assert!(!is_hop_by_hop_header("authorization")); + } +} diff --git a/src/sandbox/proxy/mod.rs b/src/sandbox/proxy/mod.rs new file mode 100644 index 00000000..2feb460c --- /dev/null +++ b/src/sandbox/proxy/mod.rs @@ -0,0 +1,164 @@ +//! Network proxy for sandboxed container access. +//! +//! The proxy provides: +//! - Domain allowlist validation +//! - Credential injection for API calls +//! - Request logging and monitoring +//! +//! # Architecture +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────────┐ +//! │ Network Proxy │ +//! │ │ +//! │ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │ +//! │ │ HTTP Proxy │───▶│ Policy │───▶│ Credential Resolver │ │ +//! │ │ Server │ │ Decider │ │ │ │ +//! │ └─────────────┘ └─────────────┘ └─────────────────────┘ │ +//! │ │ │ │ +//! │ │ ▼ │ +//! │ │ ┌─────────────┐ │ +//! │ │ │ Allowlist │ │ +//! │ │ │ Validator │ │ +//! │ │ └─────────────┘ │ +//! │ ▼ │ +//! │ ┌──────────────────────────────────────────────────────────┐ │ +//! │ │ Internet │ │ +//! │ └──────────────────────────────────────────────────────────┘ │ +//! └─────────────────────────────────────────────────────────────────┘ +//! ``` + +pub mod allowlist; +pub mod http; +pub mod policy; + +pub use allowlist::{DomainAllowlist, DomainPattern, DomainValidationResult}; +pub use http::{CredentialResolver, EnvCredentialResolver, HttpProxy, NoCredentialResolver}; +pub use policy::{ + AllowAllDecider, DefaultPolicyDecider, DenyAllDecider, NetworkDecision, NetworkPolicyDecider, + NetworkRequest, +}; + +use std::sync::Arc; + +use crate::sandbox::config::{ + CredentialMapping, SandboxConfig, SandboxPolicy, default_credential_mappings, +}; +use crate::sandbox::error::Result; + +/// Creates a configured network proxy from sandbox config. +pub struct NetworkProxyBuilder { + allowlist: Vec, + credential_mappings: Vec, + credential_resolver: Arc, + policy: SandboxPolicy, +} + +impl NetworkProxyBuilder { + /// Create a new builder with default settings. + pub fn new() -> Self { + Self { + allowlist: crate::sandbox::config::default_allowlist(), + credential_mappings: default_credential_mappings(), + credential_resolver: Arc::new(EnvCredentialResolver), + policy: SandboxPolicy::ReadOnly, + } + } + + /// Create from a sandbox config. + pub fn from_config(config: &SandboxConfig) -> Self { + Self { + allowlist: config.network_allowlist.clone(), + credential_mappings: default_credential_mappings(), + credential_resolver: Arc::new(EnvCredentialResolver), + policy: config.policy, + } + } + + /// Set the domain allowlist. + pub fn with_allowlist(mut self, domains: Vec) -> Self { + self.allowlist = domains; + self + } + + /// Add a domain to the allowlist. + pub fn allow_domain(mut self, domain: &str) -> Self { + self.allowlist.push(domain.to_string()); + self + } + + /// Set credential mappings. + pub fn with_credentials(mut self, mappings: Vec) -> Self { + self.credential_mappings = mappings; + self + } + + /// Set the credential resolver. + pub fn with_credential_resolver(mut self, resolver: Arc) -> Self { + self.credential_resolver = resolver; + self + } + + /// Set the sandbox policy. + pub fn with_policy(mut self, policy: SandboxPolicy) -> Self { + self.policy = policy; + self + } + + /// Build the HTTP proxy. + pub fn build(self) -> HttpProxy { + let decider: Arc = if self.policy.has_full_network() { + Arc::new(AllowAllDecider) + } else { + Arc::new(DefaultPolicyDecider::new( + DomainAllowlist::new(&self.allowlist), + self.credential_mappings, + )) + }; + + HttpProxy::new(decider, self.credential_resolver) + } + + /// Build and start the proxy on the given port. + pub async fn build_and_start(self, port: u16) -> Result { + let proxy = self.build(); + proxy.start(port).await?; + Ok(proxy) + } +} + +impl Default for NetworkProxyBuilder { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_builder_default() { + let builder = NetworkProxyBuilder::new(); + assert!(!builder.allowlist.is_empty()); + } + + #[test] + fn test_builder_with_custom_allowlist() { + let builder = NetworkProxyBuilder::new() + .with_allowlist(vec!["custom.com".to_string()]) + .allow_domain("another.com"); + + assert!(builder.allowlist.contains(&"custom.com".to_string())); + assert!(builder.allowlist.contains(&"another.com".to_string())); + } + + #[tokio::test] + async fn test_builder_builds_proxy() { + let proxy = NetworkProxyBuilder::new() + .with_policy(SandboxPolicy::ReadOnly) + .build(); + + assert!(!proxy.is_running()); + } +} diff --git a/src/sandbox/proxy/policy.rs b/src/sandbox/proxy/policy.rs new file mode 100644 index 00000000..f3d967a0 --- /dev/null +++ b/src/sandbox/proxy/policy.rs @@ -0,0 +1,229 @@ +//! Network policy decision making. +//! +//! Determines whether network requests should be allowed, denied, +//! or allowed with credential injection. + +use async_trait::async_trait; + +use crate::sandbox::config::{CredentialLocation, CredentialMapping}; +use crate::sandbox::proxy::allowlist::DomainAllowlist; + +/// A network request to be evaluated. +#[derive(Debug, Clone)] +pub struct NetworkRequest { + /// HTTP method (GET, POST, etc.). + pub method: String, + /// Full URL being requested. + pub url: String, + /// Host extracted from URL. + pub host: String, + /// Path portion of the URL. + pub path: String, +} + +impl NetworkRequest { + /// Create from a URL string. + pub fn from_url(method: &str, url: &str) -> Option { + let host = crate::sandbox::proxy::allowlist::extract_host(url)?; + let path = extract_path(url); + + Some(Self { + method: method.to_uppercase(), + url: url.to_string(), + host, + path, + }) + } +} + +/// Extract path from a URL. +fn extract_path(url: &str) -> String { + // Find the start of the path (after ://) + if let Some(idx) = url.find("://") { + let rest = &url[idx + 3..]; + if let Some(path_start) = rest.find('/') { + return rest[path_start..].to_string(); + } + } + "/".to_string() +} + +/// Decision for a network request. +#[derive(Debug, Clone)] +pub enum NetworkDecision { + /// Allow the request as-is. + Allow, + /// Allow with credential injection. + AllowWithCredentials { + /// Name of the secret to look up. + secret_name: String, + /// Where to inject the credential. + location: CredentialLocation, + }, + /// Deny the request. + Deny { + /// Reason for denial. + reason: String, + }, +} + +impl NetworkDecision { + pub fn is_allowed(&self) -> bool { + !matches!(self, NetworkDecision::Deny { .. }) + } +} + +/// Trait for making network policy decisions. +#[async_trait] +pub trait NetworkPolicyDecider: Send + Sync { + /// Decide whether a request should be allowed. + async fn decide(&self, request: &NetworkRequest) -> NetworkDecision; +} + +/// Default policy decider that uses allowlist and credential mappings. +pub struct DefaultPolicyDecider { + allowlist: DomainAllowlist, + credential_mappings: Vec, +} + +impl DefaultPolicyDecider { + /// Create a new policy decider. + pub fn new(allowlist: DomainAllowlist, credential_mappings: Vec) -> Self { + Self { + allowlist, + credential_mappings, + } + } + + /// Find credential mapping for a domain. + fn find_credential(&self, host: &str) -> Option<&CredentialMapping> { + let host_lower = host.to_lowercase(); + self.credential_mappings + .iter() + .find(|m| m.domain.to_lowercase() == host_lower) + } +} + +#[async_trait] +impl NetworkPolicyDecider for DefaultPolicyDecider { + async fn decide(&self, request: &NetworkRequest) -> NetworkDecision { + // First check if the domain is allowed + let validation = self.allowlist.is_allowed(&request.host); + if !validation.is_allowed() { + if let crate::sandbox::proxy::allowlist::DomainValidationResult::Denied(reason) = + validation + { + return NetworkDecision::Deny { reason }; + } + } + + // Check if we need to inject credentials + if let Some(mapping) = self.find_credential(&request.host) { + return NetworkDecision::AllowWithCredentials { + secret_name: mapping.secret_name.clone(), + location: mapping.location.clone(), + }; + } + + NetworkDecision::Allow + } +} + +/// A policy decider that allows everything (use with FullAccess policy). +pub struct AllowAllDecider; + +#[async_trait] +impl NetworkPolicyDecider for AllowAllDecider { + async fn decide(&self, _request: &NetworkRequest) -> NetworkDecision { + NetworkDecision::Allow + } +} + +/// A policy decider that denies everything. +pub struct DenyAllDecider { + reason: String, +} + +impl DenyAllDecider { + pub fn new(reason: &str) -> Self { + Self { + reason: reason.to_string(), + } + } +} + +#[async_trait] +impl NetworkPolicyDecider for DenyAllDecider { + async fn decide(&self, _request: &NetworkRequest) -> NetworkDecision { + NetworkDecision::Deny { + reason: self.reason.clone(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_network_request_from_url() { + let req = NetworkRequest::from_url("GET", "https://api.example.com/v1/data").unwrap(); + assert_eq!(req.method, "GET"); + assert_eq!(req.host, "api.example.com"); + assert_eq!(req.path, "/v1/data"); + } + + #[test] + fn test_extract_path() { + assert_eq!( + extract_path("https://example.com/api/v1"), + "/api/v1".to_string() + ); + assert_eq!(extract_path("https://example.com"), "/".to_string()); + assert_eq!(extract_path("https://example.com/"), "/".to_string()); + } + + #[tokio::test] + async fn test_default_policy_allows_listed_domain() { + let allowlist = DomainAllowlist::new(&["crates.io".to_string()]); + let decider = DefaultPolicyDecider::new(allowlist, vec![]); + + let req = NetworkRequest::from_url("GET", "https://crates.io/api/v1/crates").unwrap(); + let decision = decider.decide(&req).await; + + assert!(decision.is_allowed()); + } + + #[tokio::test] + async fn test_default_policy_denies_unlisted_domain() { + let allowlist = DomainAllowlist::new(&["crates.io".to_string()]); + let decider = DefaultPolicyDecider::new(allowlist, vec![]); + + let req = NetworkRequest::from_url("GET", "https://evil.com/steal").unwrap(); + let decision = decider.decide(&req).await; + + assert!(!decision.is_allowed()); + } + + #[tokio::test] + async fn test_credential_injection() { + let allowlist = DomainAllowlist::new(&["api.openai.com".to_string()]); + let credentials = vec![CredentialMapping { + domain: "api.openai.com".to_string(), + secret_name: "OPENAI_API_KEY".to_string(), + location: CredentialLocation::AuthorizationBearer, + }]; + let decider = DefaultPolicyDecider::new(allowlist, credentials); + + let req = + NetworkRequest::from_url("POST", "https://api.openai.com/v1/chat/completions").unwrap(); + let decision = decider.decide(&req).await; + + match decision { + NetworkDecision::AllowWithCredentials { secret_name, .. } => { + assert_eq!(secret_name, "OPENAI_API_KEY"); + } + _ => panic!("Expected AllowWithCredentials"), + } + } +} diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs new file mode 100644 index 00000000..354fe14f --- /dev/null +++ b/src/tools/builtin/job.rs @@ -0,0 +1,417 @@ +//! Job management tools. +//! +//! These tools allow the LLM to manage jobs: +//! - Create new jobs/tasks +//! - List existing jobs +//! - Check job status +//! - Cancel running jobs + +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::context::{ContextManager, JobContext, JobState}; +use crate::tools::tool::{Tool, ToolError, ToolOutput}; + +/// Tool for creating a new job. +pub struct CreateJobTool { + context_manager: Arc, +} + +impl CreateJobTool { + pub fn new(context_manager: Arc) -> Self { + Self { context_manager } + } +} + +#[async_trait] +impl Tool for CreateJobTool { + fn name(&self) -> &str { + "create_job" + } + + fn description(&self) -> &str { + "Create a new job or task for the agent to work on. Use this when the user wants \ + you to do something substantial that should be tracked as a separate job." + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": { + "title": { + "type": "string", + "description": "A short title for the job (max 100 chars)" + }, + "description": { + "type": "string", + "description": "Full description of what needs to be done" + } + }, + "required": ["title", "description"] + }) + } + + async fn execute( + &self, + params: serde_json::Value, + _ctx: &JobContext, + ) -> Result { + let start = std::time::Instant::now(); + + let title = params + .get("title") + .and_then(|v| v.as_str()) + .ok_or_else(|| ToolError::InvalidParameters("missing 'title' parameter".into()))?; + + let description = params + .get("description") + .and_then(|v| v.as_str()) + .ok_or_else(|| { + ToolError::InvalidParameters("missing 'description' parameter".into()) + })?; + + match self.context_manager.create_job(title, description).await { + Ok(job_id) => { + let result = serde_json::json!({ + "job_id": job_id.to_string(), + "title": title, + "status": "pending", + "message": format!("Created job '{}'", title) + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + Err(e) => { + let result = serde_json::json!({ + "error": e.to_string() + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + } + } + + fn requires_sanitization(&self) -> bool { + false + } +} + +/// Tool for listing jobs. +pub struct ListJobsTool { + context_manager: Arc, +} + +impl ListJobsTool { + pub fn new(context_manager: Arc) -> Self { + Self { context_manager } + } +} + +#[async_trait] +impl Tool for ListJobsTool { + fn name(&self) -> &str { + "list_jobs" + } + + fn description(&self) -> &str { + "List all jobs or filter by status. Shows job IDs, titles, and current status." + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": { + "filter": { + "type": "string", + "description": "Filter by status: 'active', 'completed', 'failed', 'all' (default: 'all')", + "enum": ["active", "completed", "failed", "all"] + } + } + }) + } + + async fn execute( + &self, + params: serde_json::Value, + _ctx: &JobContext, + ) -> Result { + let start = std::time::Instant::now(); + + let filter = params + .get("filter") + .and_then(|v| v.as_str()) + .unwrap_or("all"); + + let job_ids = match filter { + "active" => self.context_manager.active_jobs().await, + _ => self.context_manager.all_jobs().await, + }; + + let mut jobs = Vec::new(); + for job_id in job_ids { + if let Ok(ctx) = self.context_manager.get_context(job_id).await { + let include = match filter { + "completed" => ctx.state == JobState::Completed, + "failed" => ctx.state == JobState::Failed, + "active" => ctx.state.is_active(), + _ => true, + }; + + if include { + jobs.push(serde_json::json!({ + "job_id": job_id.to_string(), + "title": ctx.title, + "status": format!("{:?}", ctx.state), + "created_at": ctx.created_at.to_rfc3339() + })); + } + } + } + + let summary = self.context_manager.summary().await; + + let result = serde_json::json!({ + "jobs": jobs, + "summary": { + "total": summary.total, + "pending": summary.pending, + "in_progress": summary.in_progress, + "completed": summary.completed, + "failed": summary.failed + } + }); + + Ok(ToolOutput::success(result, start.elapsed())) + } + + fn requires_sanitization(&self) -> bool { + false + } +} + +/// Tool for checking job status. +pub struct JobStatusTool { + context_manager: Arc, +} + +impl JobStatusTool { + pub fn new(context_manager: Arc) -> Self { + Self { context_manager } + } +} + +#[async_trait] +impl Tool for JobStatusTool { + fn name(&self) -> &str { + "job_status" + } + + fn description(&self) -> &str { + "Check the status and details of a specific job by its ID." + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": { + "job_id": { + "type": "string", + "description": "The UUID of the job to check" + } + }, + "required": ["job_id"] + }) + } + + async fn execute( + &self, + params: serde_json::Value, + _ctx: &JobContext, + ) -> Result { + let start = std::time::Instant::now(); + + let job_id_str = params + .get("job_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| ToolError::InvalidParameters("missing 'job_id' parameter".into()))?; + + let job_id = Uuid::parse_str(job_id_str).map_err(|_| { + ToolError::InvalidParameters(format!("invalid job ID format: {}", job_id_str)) + })?; + + match self.context_manager.get_context(job_id).await { + Ok(ctx) => { + let result = serde_json::json!({ + "job_id": job_id.to_string(), + "title": ctx.title, + "description": ctx.description, + "status": format!("{:?}", ctx.state), + "created_at": ctx.created_at.to_rfc3339(), + "started_at": ctx.started_at.map(|t| t.to_rfc3339()), + "completed_at": ctx.completed_at.map(|t| t.to_rfc3339()), + "actual_cost": ctx.actual_cost.to_string() + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + Err(e) => { + let result = serde_json::json!({ + "error": format!("Job not found: {}", e) + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + } + } + + fn requires_sanitization(&self) -> bool { + false + } +} + +/// Tool for canceling a job. +pub struct CancelJobTool { + context_manager: Arc, +} + +impl CancelJobTool { + pub fn new(context_manager: Arc) -> Self { + Self { context_manager } + } +} + +#[async_trait] +impl Tool for CancelJobTool { + fn name(&self) -> &str { + "cancel_job" + } + + fn description(&self) -> &str { + "Cancel a running or pending job. The job will be marked as cancelled and stopped." + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": { + "job_id": { + "type": "string", + "description": "The UUID of the job to cancel" + } + }, + "required": ["job_id"] + }) + } + + async fn execute( + &self, + params: serde_json::Value, + _ctx: &JobContext, + ) -> Result { + let start = std::time::Instant::now(); + + let job_id_str = params + .get("job_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| ToolError::InvalidParameters("missing 'job_id' parameter".into()))?; + + let job_id = Uuid::parse_str(job_id_str).map_err(|_| { + ToolError::InvalidParameters(format!("invalid job ID format: {}", job_id_str)) + })?; + + // Transition to cancelled state + match self + .context_manager + .update_context(job_id, |ctx| { + ctx.transition_to(JobState::Cancelled, Some("Cancelled by user".to_string())) + }) + .await + { + Ok(Ok(())) => { + let result = serde_json::json!({ + "job_id": job_id.to_string(), + "status": "cancelled", + "message": "Job cancelled successfully" + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + Ok(Err(reason)) => { + let result = serde_json::json!({ + "error": format!("Cannot cancel job: {}", reason) + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + Err(e) => { + let result = serde_json::json!({ + "error": format!("Job not found: {}", e) + }); + Ok(ToolOutput::success(result, start.elapsed())) + } + } + } + + fn requires_approval(&self) -> bool { + true // Canceling a job should require approval + } + + fn requires_sanitization(&self) -> bool { + false + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_create_job_tool() { + let manager = Arc::new(ContextManager::new(5)); + let tool = CreateJobTool::new(manager.clone()); + + let params = serde_json::json!({ + "title": "Test Job", + "description": "A test job description" + }); + + let ctx = JobContext::default(); + let result = tool.execute(params, &ctx).await.unwrap(); + + let job_id = result.result.get("job_id").unwrap().as_str().unwrap(); + assert!(!job_id.is_empty()); + } + + #[tokio::test] + async fn test_list_jobs_tool() { + let manager = Arc::new(ContextManager::new(5)); + + // Create some jobs + manager.create_job("Job 1", "Desc 1").await.unwrap(); + manager.create_job("Job 2", "Desc 2").await.unwrap(); + + let tool = ListJobsTool::new(manager); + + let params = serde_json::json!({}); + let ctx = JobContext::default(); + let result = tool.execute(params, &ctx).await.unwrap(); + + let jobs = result.result.get("jobs").unwrap().as_array().unwrap(); + assert_eq!(jobs.len(), 2); + } + + #[tokio::test] + async fn test_job_status_tool() { + let manager = Arc::new(ContextManager::new(5)); + let job_id = manager.create_job("Test Job", "Description").await.unwrap(); + + let tool = JobStatusTool::new(manager); + + let params = serde_json::json!({ + "job_id": job_id.to_string() + }); + let ctx = JobContext::default(); + let result = tool.execute(params, &ctx).await.unwrap(); + + assert_eq!( + result.result.get("title").unwrap().as_str().unwrap(), + "Test Job" + ); + } +} diff --git a/src/tools/builtin/mod.rs b/src/tools/builtin/mod.rs index 6d9c8c64..b922ca66 100644 --- a/src/tools/builtin/mod.rs +++ b/src/tools/builtin/mod.rs @@ -4,6 +4,7 @@ mod echo; mod ecommerce; mod file; mod http; +mod job; mod json; mod marketplace; mod memory; @@ -16,6 +17,7 @@ pub use echo::EchoTool; pub use ecommerce::EcommerceTool; pub use file::{ApplyPatchTool, ListDirTool, ReadFileTool, WriteFileTool}; pub use http::HttpTool; +pub use job::{CancelJobTool, CreateJobTool, JobStatusTool, ListJobsTool}; pub use json::JsonTool; pub use marketplace::MarketplaceTool; pub use memory::{MemoryReadTool, MemorySearchTool, MemoryTreeTool, MemoryWriteTool}; diff --git a/src/tools/builtin/shell.rs b/src/tools/builtin/shell.rs index 6da5ec0b..ca59dad6 100644 --- a/src/tools/builtin/shell.rs +++ b/src/tools/builtin/shell.rs @@ -1,15 +1,27 @@ //! Shell execution tool for running commands in a sandboxed environment. //! //! Provides controlled command execution with: +//! - Docker sandbox isolation (when enabled) //! - Working directory isolation //! - Timeout enforcement //! - Output capture and truncation //! - Blocked command patterns for safety +//! +//! # Execution Modes +//! +//! When sandbox is available and enabled: +//! - Commands run inside ephemeral Docker containers +//! - Network traffic goes through a validating proxy +//! - Credentials are injected by the proxy, never exposed to commands +//! +//! When sandbox is unavailable: +//! - Commands run directly on host with basic protections +//! - Blocked command patterns are still enforced use std::collections::HashSet; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::Stdio; -use std::sync::LazyLock; +use std::sync::{Arc, LazyLock}; use std::time::Duration; use async_trait::async_trait; @@ -17,6 +29,7 @@ use tokio::io::AsyncReadExt; use tokio::process::Command; use crate::context::JobContext; +use crate::sandbox::{SandboxManager, SandboxPolicy}; use crate::tools::tool::{Tool, ToolError, ToolOutput}; /// Maximum output size before truncation (64KB). @@ -62,7 +75,6 @@ static DANGEROUS_PATTERNS: LazyLock> = LazyLock::new(|| { }); /// Shell command execution tool. -#[derive(Debug)] pub struct ShellTool { /// Working directory for commands (if None, uses job's working dir or cwd). working_dir: Option, @@ -70,6 +82,22 @@ pub struct ShellTool { timeout: Duration, /// Whether to allow potentially dangerous commands (requires explicit approval). allow_dangerous: bool, + /// Optional sandbox manager for Docker execution. + sandbox: Option>, + /// Sandbox policy to use when sandbox is available. + sandbox_policy: SandboxPolicy, +} + +impl std::fmt::Debug for ShellTool { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ShellTool") + .field("working_dir", &self.working_dir) + .field("timeout", &self.timeout) + .field("allow_dangerous", &self.allow_dangerous) + .field("sandbox", &self.sandbox.is_some()) + .field("sandbox_policy", &self.sandbox_policy) + .finish() + } } impl ShellTool { @@ -79,6 +107,8 @@ impl ShellTool { working_dir: None, timeout: DEFAULT_TIMEOUT, allow_dangerous: false, + sandbox: None, + sandbox_policy: SandboxPolicy::ReadOnly, } } @@ -94,6 +124,18 @@ impl ShellTool { self } + /// Enable sandbox execution with the given manager. + pub fn with_sandbox(mut self, sandbox: Arc) -> Self { + self.sandbox = Some(sandbox); + self + } + + /// Set the sandbox policy. + pub fn with_sandbox_policy(mut self, policy: SandboxPolicy) -> Self { + self.sandbox_policy = policy; + self + } + /// Check if a command is blocked. fn is_blocked(&self, cmd: &str) -> Option<&'static str> { let normalized = cmd.to_lowercase(); @@ -115,28 +157,44 @@ impl ShellTool { None } - /// Execute a command and capture output. - async fn execute_command( + /// Execute a command through the sandbox. + async fn execute_sandboxed( + &self, + sandbox: &SandboxManager, + cmd: &str, + workdir: &Path, + timeout: Duration, + ) -> Result<(String, i64), ToolError> { + // Override sandbox config timeout if needed + let result = tokio::time::timeout(timeout, async { + sandbox + .execute_with_policy( + cmd, + workdir, + self.sandbox_policy, + std::collections::HashMap::new(), + ) + .await + }) + .await; + + match result { + Ok(Ok(output)) => { + let combined = truncate_output(&output.output); + Ok((combined, output.exit_code)) + } + Ok(Err(e)) => Err(ToolError::ExecutionFailed(format!("Sandbox error: {}", e))), + Err(_) => Err(ToolError::Timeout(timeout)), + } + } + + /// Execute a command directly (fallback when sandbox unavailable). + async fn execute_direct( &self, cmd: &str, - workdir: Option<&str>, - timeout: Option, + workdir: &PathBuf, + timeout: Duration, ) -> Result<(String, i32), ToolError> { - // Check for blocked commands - if let Some(reason) = self.is_blocked(cmd) { - return Err(ToolError::NotAuthorized(format!( - "{}: {}", - reason, - truncate_for_error(cmd) - ))); - } - - // Determine working directory - let cwd = workdir - .map(PathBuf::from) - .or_else(|| self.working_dir.clone()) - .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); - // Build command let mut command = if cfg!(target_os = "windows") { let mut c = Command::new("cmd"); @@ -149,7 +207,7 @@ impl ShellTool { }; command - .current_dir(&cwd) + .current_dir(workdir) .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); @@ -159,11 +217,8 @@ impl ShellTool { .spawn() .map_err(|e| ToolError::ExecutionFailed(format!("Failed to spawn command: {}", e)))?; - // Determine timeout - let timeout_duration = timeout.map(Duration::from_secs).unwrap_or(self.timeout); - // Wait with timeout - let result = tokio::time::timeout(timeout_duration, async { + let result = tokio::time::timeout(timeout, async { let status = child.wait().await?; // Read stdout @@ -204,10 +259,56 @@ impl ShellTool { Err(_) => { // Timeout - try to kill the process let _ = child.kill().await; - Err(ToolError::Timeout(timeout_duration)) + Err(ToolError::Timeout(timeout)) } } } + + /// Execute a command, using sandbox if available. + async fn execute_command( + &self, + cmd: &str, + workdir: Option<&str>, + timeout: Option, + ) -> Result<(String, i64), ToolError> { + // Check for blocked commands + if let Some(reason) = self.is_blocked(cmd) { + return Err(ToolError::NotAuthorized(format!( + "{}: {}", + reason, + truncate_for_error(cmd) + ))); + } + + // Determine working directory + let cwd = workdir + .map(PathBuf::from) + .or_else(|| self.working_dir.clone()) + .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); + + // Determine timeout + let timeout_duration = timeout.map(Duration::from_secs).unwrap_or(self.timeout); + + // Try sandbox execution if available + if let Some(ref sandbox) = self.sandbox { + if sandbox.is_initialized() || sandbox.config().enabled { + match self + .execute_sandboxed(sandbox, cmd, &cwd, timeout_duration) + .await + { + Ok((output, code)) => return Ok((output, code)), + Err(e) => { + // Log sandbox failure and fall through to direct execution + tracing::warn!("Sandbox execution failed, falling back to direct: {}", e); + } + } + } + } + + // Fallback to direct execution + let (output, code) = self.execute_direct(cmd, &cwd, timeout_duration).await?; + Ok((output, code as i64)) + } } impl Default for ShellTool { @@ -224,7 +325,8 @@ impl Tool for ShellTool { fn description(&self) -> &str { "Execute shell commands. Use for running builds, tests, git operations, and other CLI tasks. \ - Commands run in a subprocess with captured output. Long-running commands have a timeout." + Commands run in a subprocess with captured output. Long-running commands have a timeout. \ + When Docker sandbox is enabled, commands run in isolated containers for security." } fn parameters_schema(&self) -> serde_json::Value { @@ -265,10 +367,13 @@ impl Tool for ShellTool { let (output, exit_code) = self.execute_command(command, workdir, timeout).await?; let duration = start.elapsed(); + let sandboxed = self.sandbox.is_some(); + let result = serde_json::json!({ "output": output, "exit_code": exit_code, - "success": exit_code == 0 + "success": exit_code == 0, + "sandboxed": sandboxed }); Ok(ToolOutput::success(result, duration)) @@ -348,4 +453,14 @@ mod tests { assert!(matches!(result, Err(ToolError::Timeout(_)))); } + + #[test] + fn test_sandbox_policy_builder() { + let tool = ShellTool::new() + .with_sandbox_policy(SandboxPolicy::WorkspaceWrite) + .with_timeout(Duration::from_secs(60)); + + assert_eq!(tool.sandbox_policy, SandboxPolicy::WorkspaceWrite); + assert_eq!(tool.timeout, Duration::from_secs(60)); + } } diff --git a/src/tools/registry.rs b/src/tools/registry.rs index 805b29fb..9e87b751 100644 --- a/src/tools/registry.rs +++ b/src/tools/registry.rs @@ -5,12 +5,14 @@ use std::sync::Arc; use tokio::sync::RwLock; +use crate::context::ContextManager; use crate::llm::{LlmProvider, ToolDefinition}; use crate::safety::SafetyLayer; use crate::tools::builder::{BuildSoftwareTool, BuilderConfig, LlmSoftwareBuilder}; use crate::tools::builtin::{ - ApplyPatchTool, EchoTool, HttpTool, JsonTool, ListDirTool, MemoryReadTool, MemorySearchTool, - MemoryTreeTool, MemoryWriteTool, ReadFileTool, ShellTool, TimeTool, WriteFileTool, + ApplyPatchTool, CancelJobTool, CreateJobTool, EchoTool, HttpTool, JobStatusTool, JsonTool, + ListDirTool, ListJobsTool, MemoryReadTool, MemorySearchTool, MemoryTreeTool, MemoryWriteTool, + ReadFileTool, ShellTool, TimeTool, WriteFileTool, }; use crate::tools::tool::Tool; use crate::tools::wasm::{ @@ -144,6 +146,19 @@ impl ToolRegistry { tracing::info!("Registered 4 memory tools"); } + /// Register job management tools. + /// + /// Job tools allow the LLM to create, list, check status, and cancel jobs. + /// These enable natural language job management without hardcoded intent parsing. + pub fn register_job_tools(&self, context_manager: Arc) { + self.register_sync(Arc::new(CreateJobTool::new(Arc::clone(&context_manager)))); + self.register_sync(Arc::new(ListJobsTool::new(Arc::clone(&context_manager)))); + self.register_sync(Arc::new(JobStatusTool::new(Arc::clone(&context_manager)))); + self.register_sync(Arc::new(CancelJobTool::new(context_manager))); + + tracing::info!("Registered 4 job management tools"); + } + /// Register the software builder tool. /// /// The builder tool allows the agent to create new software including WASM tools,