#[cfg(unix)]
use std::path::Path;
use std::path::PathBuf;
#[cfg(unix)]
use std::sync::atomic::Ordering;
use std::sync::{atomic::AtomicBool, Arc, Mutex};
use std::time::Instant;
#[cfg(unix)]
use crate::config::NetMode;
#[cfg(unix)]
use crate::events::Event;
use crate::events::EventBus;
use crate::multi::isolation::IsolationBarrier;
use crate::multi::{AgentSpec, Manifest, MultiAggregator, MultiEventStream, VirtualPortPool};
#[cfg(unix)]
use crate::policy;
use crate::report::stats::StatsCollector;
use crate::report::{self, storage::ReportStorage, ReportOptions};
use crate::sandbox::SandboxHandle;
#[cfg(unix)]
use crate::sandbox::{Backend, SpawnOptions, StdioMode};
use anyhow::{bail, Context, Result};
#[cfg(unix)]
use std::collections::HashMap;
#[cfg(unix)]
use std::io::Read;
#[cfg(target_os = "linux")]
use std::os::fd::IntoRawFd;
#[cfg(unix)]
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
#[cfg(unix)]
const OUTPUT_CAP: usize = 512 * 1024;
#[derive(Default)]
pub struct OutputBuffers {
pub stdout: Vec<u8>,
pub stderr: Vec<u8>,
}
impl OutputBuffers {
pub fn text(&self) -> String {
let mut bytes = self.stdout.clone();
bytes.extend_from_slice(&self.stderr);
String::from_utf8_lossy(&bytes).into_owned()
}
}
pub struct MultiSession {
pub spec: AgentSpec,
pub bus: EventBus,
pub stats: StatsCollector,
pub output: Arc<Mutex<OutputBuffers>>,
pub handle: Arc<Mutex<SandboxHandle>>,
pub finished: Arc<AtomicBool>,
pub started: Instant,
pub allocated_ports: Vec<u16>,
}
#[cfg(unix)]
struct PendingSession {
spec: AgentSpec,
net: NetMode,
tier: policy::Tier,
policy: policy::Policy,
bus: EventBus,
handle: SandboxHandle,
stdout_r: OwnedFd,
stderr_r: OwnedFd,
broker_ctrl_fd: Option<OwnedFd>,
notif_listener: Option<OwnedFd>,
allocated_ports: Vec<u16>,
}
#[cfg(unix)]
impl PendingSession {
fn terminate(&mut self) {
self.handle.terminate();
}
}
impl MultiSession {
pub fn pause(&self) {
if let Ok(mut handle) = self.handle.lock() {
handle.pause();
}
}
pub fn resume(&self) {
if let Ok(mut handle) = self.handle.lock() {
handle.resume();
}
}
pub fn terminate(&self) {
if let Ok(mut handle) = self.handle.lock() {
handle.terminate();
}
}
pub fn try_wait(&self) -> Option<i32> {
self.handle.lock().ok()?.try_wait()
}
pub fn output_text(&self) -> String {
self.output
.lock()
.map(|output| output.text())
.unwrap_or_default()
}
}
pub struct MultiRuntime {
pub manifest: Manifest,
pub sessions: Vec<MultiSession>,
pub stream: MultiEventStream,
pub aggregator: MultiAggregator,
pub port_pool: VirtualPortPool,
pub isolation_barrier: IsolationBarrier,
pub report_dir: Option<PathBuf>,
}
impl MultiRuntime {
#[cfg(unix)]
pub fn launch(manifest: Manifest, project: PathBuf, home: PathBuf) -> Result<Self> {
manifest.validate()?;
let _ = crate::multi::isolation::set_subreaper();
let port_pool = VirtualPortPool::default();
let isolation_barrier = IsolationBarrier::new();
let mut prepared = Vec::with_capacity(manifest.agents.len());
for (idx, spec) in manifest.agents.iter().enumerate() {
let net = crate::config::parse_net_mode(&spec.net)
.with_context(|| format!("agent '{}' network mode", spec.name))?;
let backend = Backend::detect(net.clone(), spec.observe_seccomp)
.with_context(|| format!("establish sandbox backend for agent '{}'", spec.name))?;
let tier = backend.tier().unwrap_or(policy::Tier::Full);
let policy =
policy::loader::load(&spec.profile, spec.policy.as_deref(), &project, &home, tier)
.with_context(|| format!("load policy for agent '{}'", spec.name))?;
let mut command = spec.command.clone();
command[0] = resolve_in_path(&command[0])
.with_context(|| format!("resolve command for agent '{}'", spec.name))?;
if !policy.in_read_scope(Path::new(&command[0])) {
tracing::warn!(
agent = %spec.name,
command = %command[0],
"agent executable is outside policy read scope; sandbox exec may be denied"
);
}
let allocated_ports = port_pool
.allocate_ports(&spec.name, 4)
.unwrap_or_else(|_| vec![port_pool.allocate_relay_port(idx)]);
prepared.push(Prepared {
spec: spec.clone(),
net,
backend: Some(backend),
policy,
command,
tier,
allocated_ports,
});
}
let mut pending = Vec::with_capacity(prepared.len());
for prepared in prepared {
match spawn_one(prepared, &project) {
Ok(session) => pending.push(session),
Err(error) => {
for session in &mut pending {
session.terminate();
}
bail!("multi-agent launch aborted; no unsandboxed fallback: {error:#}");
}
}
}
let stream = MultiEventStream::new();
let aggregator =
MultiAggregator::new(manifest.agents.iter().map(|agent| agent.name.clone()));
crate::multi::spawn_aggregator(&stream, aggregator.clone());
let mut sessions = Vec::with_capacity(pending.len());
for pending in pending {
let session = activate_pending(pending, &project, &stream, &isolation_barrier);
sessions.push(session);
}
Ok(Self {
report_dir: manifest.report_dir.clone(),
manifest,
sessions,
stream,
aggregator,
port_pool,
isolation_barrier,
})
}
#[cfg(not(unix))]
pub fn launch(_manifest: Manifest, _project: PathBuf, _home: PathBuf) -> Result<Self> {
bail!("multi-agent mode is unavailable on this platform; refusing to run unsandboxed")
}
pub fn terminate(&self, index: usize) -> Result<()> {
let session = self
.sessions
.get(index)
.ok_or_else(|| anyhow::anyhow!("unknown multi-agent pane {index}"))?;
session.terminate();
Ok(())
}
pub fn terminate_all(&self) {
for session in &self.sessions {
session.terminate();
}
}
pub fn combined_report(&self) -> serde_json::Value {
self.aggregator.report_json()
}
pub fn write_reports(&self) -> Result<Vec<PathBuf>> {
let mut written = Vec::new();
let rows = self.aggregator.snapshot();
for agent in &self.manifest.agents {
let dir = agent.report_path(self.report_dir.as_deref());
let options = ReportOptions {
report_dir: Some(dir),
auto_cleanup: false,
retention: None,
max_age_secs: None,
};
let storage = ReportStorage::new(&options)
.with_context(|| format!("prepare report directory for agent '{}'", agent.name))?;
let row = rows
.iter()
.find(|stats| stats.name == agent.name)
.cloned()
.unwrap_or_else(|| crate::multi::AgentStats::new(agent.name.clone()));
let mut value = serde_json::to_value(row).context("serialize agent report")?;
report::sanitize_json_strings(&mut value);
let text = serde_json::to_string_pretty(&value).context("render agent report")?;
let path = storage
.write("json", &text)
.with_context(|| format!("write report for agent '{}'", agent.name))?;
written.push(path);
}
let combined_dir = self
.report_dir
.clone()
.unwrap_or_else(|| PathBuf::from("."));
let options = ReportOptions {
report_dir: Some(combined_dir),
auto_cleanup: false,
retention: None,
max_age_secs: None,
};
let storage = ReportStorage::new(&options).context("prepare combined report directory")?;
let mut combined = self.combined_report();
report::sanitize_json_strings(&mut combined);
let combined = serde_json::to_string_pretty(&combined).context("render combined report")?;
let combined_path = storage
.write("json", &combined)
.context("write combined report")?;
written.push(combined_path);
Ok(written)
}
}
#[cfg(unix)]
struct Prepared {
spec: AgentSpec,
net: NetMode,
backend: Option<Backend>,
policy: policy::Policy,
command: Vec<String>,
tier: policy::Tier,
allocated_ports: Vec<u16>,
}
#[cfg(unix)]
fn spawn_one(prepared: Prepared, project: &Path) -> Result<PendingSession> {
let Prepared {
spec,
net,
backend,
policy,
command,
tier,
allocated_ports,
} = prepared;
let backend = backend.ok_or_else(|| anyhow::anyhow!("sandbox backend was consumed"))?;
let (stdout_r, stdout_w) = pipe2()?;
let (stderr_r, stderr_w) = pipe2()?;
let options = SpawnOptions {
agent_cmd: command,
cwd: project.to_path_buf(),
env_extra: relay_env(&net),
stdio: StdioMode::Captured {
stdout_w: stdout_w.as_raw_fd(),
stderr_w: stderr_w.as_raw_fd(),
},
};
let spawned = backend
.spawn(&policy, options)
.with_context(|| format!("spawn agent '{}' inside its sandbox", spec.name))?;
let crate::sandbox::Spawned {
handle,
broker_ctrl_fd,
relay_port: _relay_port,
notif_listener,
} = spawned;
drop(stdout_w);
drop(stderr_w);
Ok(PendingSession {
spec,
net,
tier,
policy,
bus: EventBus::new(),
handle,
stdout_r,
stderr_r,
broker_ctrl_fd,
notif_listener,
allocated_ports,
})
}
#[cfg(unix)]
fn activate_pending(
pending: PendingSession,
project: &Path,
stream: &MultiEventStream,
isolation_barrier: &IsolationBarrier,
) -> MultiSession {
#[cfg(not(target_os = "linux"))]
let _ = project;
let PendingSession {
spec,
net,
tier,
policy,
bus,
handle,
stdout_r,
stderr_r,
broker_ctrl_fd,
notif_listener,
allocated_ports,
} = pending;
let stats = StatsCollector::spawn(&bus);
let root_pid = handle.root_pid;
let is_full = tier == policy::Tier::Full;
isolation_barrier.register_agent(
&spec.name,
root_pid,
is_full,
is_full,
policy.limits.address_space_bytes,
);
stream.bridge_agent(spec.name.clone(), &bus);
bus.publish(Event::SessionStarted {
ts: crate::events::types::now(),
pid: root_pid,
tier: tier.label().to_string(),
net_mode: net.label(),
profile: policy.name.clone(),
});
#[cfg(target_os = "linux")]
{
if let Some(fd) = broker_ctrl_fd {
let broker_policy = match &net {
NetMode::Allowlist(domains) => {
crate::sandbox::linux::net_relay::BrokerPolicy::Allowlist(domains.clone())
}
NetMode::Strict(rules) => {
crate::sandbox::linux::net_relay::BrokerPolicy::Strict(rules.clone())
}
NetMode::Ask => crate::sandbox::linux::net_relay::BrokerPolicy::Ask,
NetMode::Off => {
crate::sandbox::linux::net_relay::BrokerPolicy::Allowlist(Vec::new())
}
};
let debug_config = spec
.debug_ports
.as_ref()
.map(|p| crate::sandbox::linux::debug_guard::DebugPortConfig {
isolate_devtools: p.isolate_devtools,
isolate_node_inspect: p.isolate_node_inspect,
isolate_debugpy: p.isolate_debugpy,
allowed_ports: p.allowed_ports.clone(),
})
.unwrap_or_default();
let debug_guard = crate::sandbox::linux::debug_guard::DebugPortGuard::new(debug_config);
let broker_config = crate::sandbox::linux::net_relay::BrokerConfig {
policy: broker_policy,
debug_guard: Some(debug_guard),
mode: crate::sandbox::linux::net_relay::RelayMode::NetNs,
allow_cidr: policy.allow_cidr.clone(),
quotas: policy.net_quota.clone(),
};
crate::sandbox::linux::net_relay::spawn_broker(
fd.into_raw_fd(),
broker_config,
bus.clone(),
);
}
if let Some(fd) = notif_listener {
crate::sandbox::linux::observe_seccomp::spawn_notifier(
fd,
bus.clone(),
Arc::new(policy.clone()),
project.to_path_buf(),
);
}
crate::sandbox::linux::visibility::spawn_poller(bus.clone(), vec![root_pid]);
}
#[cfg(all(unix, not(target_os = "linux")))]
{
let _ = (broker_ctrl_fd, notif_listener);
}
let output = Arc::new(Mutex::new(OutputBuffers::default()));
spawn_pipe_reader(stdout_r, Arc::clone(&output), true);
spawn_pipe_reader(stderr_r, Arc::clone(&output), false);
let handle = Arc::new(Mutex::new(handle));
let finished = Arc::new(AtomicBool::new(false));
let wait_handle = Arc::clone(&handle);
let wait_finished = Arc::clone(&finished);
let wait_bus = bus.clone();
let agent_name = spec.name.clone();
let barrier_clone = isolation_barrier.clone();
std::thread::Builder::new()
.name(format!("vetto-multi-wait-{}", spec.name))
.spawn(move || {
let code = wait_handle
.lock()
.map(|mut handle| handle.wait())
.unwrap_or(-1);
wait_bus.publish(Event::SessionEnded {
ts: crate::events::types::now(),
exit_code: code,
duration_secs: 0,
});
barrier_clone.unregister_agent(&agent_name);
wait_finished.store(true, Ordering::SeqCst);
})
.expect("spawn multi wait thread");
MultiSession {
spec,
bus,
stats,
output,
handle,
finished,
started: Instant::now(),
allocated_ports,
}
}
#[cfg(unix)]
fn spawn_pipe_reader(fd: OwnedFd, output: Arc<Mutex<OutputBuffers>>, stdout: bool) {
std::thread::Builder::new()
.name("vetto-multi-output".into())
.spawn(move || {
let mut file: std::fs::File = fd.into();
let mut chunk = [0u8; 8192];
loop {
match file.read(&mut chunk) {
Ok(0) | Err(_) => break,
Ok(n) => {
if let Ok(mut output) = output.lock() {
let target = if stdout {
&mut output.stdout
} else {
&mut output.stderr
};
target.extend_from_slice(&chunk[..n]);
if target.len() > OUTPUT_CAP {
let excess = target.len() - OUTPUT_CAP;
target.drain(..excess);
}
}
}
}
}
})
.expect("spawn multi output reader");
}
#[cfg(unix)]
fn pipe2() -> Result<(OwnedFd, OwnedFd)> {
let mut fds = [0 as libc::c_int; 2];
if unsafe { libc::pipe(fds.as_mut_ptr()) } != 0 {
bail!("pipe: {}", std::io::Error::last_os_error());
}
for fd in fds {
let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
if flags < 0 {
let error = std::io::Error::last_os_error();
unsafe {
libc::close(fds[0]);
libc::close(fds[1]);
}
bail!("fcntl(F_GETFD): {error}");
}
if unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) } < 0 {
let error = std::io::Error::last_os_error();
unsafe {
libc::close(fds[0]);
libc::close(fds[1]);
}
bail!("fcntl(F_SETFD): {error}");
}
}
Ok((unsafe { OwnedFd::from_raw_fd(fds[0]) }, unsafe {
OwnedFd::from_raw_fd(fds[1])
}))
}
#[cfg(target_os = "linux")]
fn relay_env(net: &NetMode) -> HashMap<String, String> {
let mut env = HashMap::new();
if net.uses_relay() {
for (key, value) in crate::sandbox::linux::net_relay::build_proxy_env(
crate::sandbox::linux::net_relay::RELAY_PORT_BASE,
) {
env.insert(key, value);
}
}
env
}
#[cfg(all(unix, not(target_os = "linux")))]
fn relay_env(_net: &NetMode) -> HashMap<String, String> {
HashMap::new()
}
#[cfg(unix)]
fn resolve_in_path(command: &str) -> Result<String> {
if command.contains('/') {
return Ok(command.to_string());
}
for dir in std::env::var_os("PATH")
.unwrap_or_default()
.to_string_lossy()
.split(':')
{
if dir.is_empty() {
continue;
}
let candidate = Path::new(dir).join(command);
if std::fs::metadata(&candidate)
.map(|meta| meta.is_file())
.unwrap_or(false)
{
return Ok(candidate.to_string_lossy().into_owned());
}
}
bail!("agent command '{command}' not found in PATH")
}
#[cfg(test)]
mod tests {
#[cfg(unix)]
use super::*;
use crate::multi::parse_manifest_str;
use std::path::Path;
#[test]
fn report_directory_is_per_agent() {
let manifest = parse_manifest_str(
r#"
[[agents]]
name = "one"
command = ["one"]
[[agents]]
name = "two"
command = ["two"]
"#,
)
.expect("manifest");
let root = Path::new("reports");
assert_ne!(
manifest.agents[0].report_path(Some(root)),
manifest.agents[1].report_path(Some(root))
);
}
#[cfg(unix)]
#[test]
fn write_reports_refuses_symlinked_agent_directory() {
use std::os::unix::fs::symlink;
use std::time::{SystemTime, UNIX_EPOCH};
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock after epoch")
.as_nanos();
let root = std::env::temp_dir().join(format!("vetto-multi-storage-{nonce}"));
let real = root.join("real");
let link = root.join("link");
std::fs::create_dir_all(&real).expect("create real report directory");
symlink(&real, &link).expect("create report directory symlink");
let manifest = Manifest {
version: 1,
agents: vec![AgentSpec {
name: "one".into(),
command: vec!["agent".into()],
profile: "default".into(),
policy: None,
net: "off".into(),
observe_seccomp: false,
report_dir: Some(link.clone()),
debug_ports: None,
}],
report_dir: Some(root.join("combined")),
};
let runtime = MultiRuntime {
manifest,
sessions: Vec::new(),
stream: MultiEventStream::new(),
aggregator: MultiAggregator::new(["one".to_string()]),
port_pool: VirtualPortPool::default(),
isolation_barrier: IsolationBarrier::new(),
report_dir: Some(root.join("combined")),
};
assert!(runtime.write_reports().is_err());
assert!(real
.read_dir()
.expect("read real directory")
.next()
.is_none());
std::fs::remove_file(&link).expect("remove report directory symlink");
std::fs::remove_dir(&real).expect("remove real report directory");
std::fs::remove_dir(&root).expect("remove report root");
}
}