Allow-listed egress over vsock: network "egress" { allow "host"... } ea81796bb757

//! The guest's half of allow-listed egress (oxygenflow#55): a proxy on

//! 127.0.0.1:3128 that relays each connection, unread, over vsock to the

//! runner (host CID 2), which decides what may pass. The VM has no network

//! device; this is its only way out.

use std::fs::File;

use std::io::{Read, Write};

use std::net::{TcpListener, TcpStream};

use std::os::fd::{AsRawFd, FromRawFd};

/// Where the job's tools reach the proxy, and the vsock port behind it.

const LISTEN: &str = "127.0.0.1:3128";

const VSOCK_PORT: u32 = 3128;

/// Start relaying in the background (before the job starts, so its first

/// connection finds the proxy listening).

pub fn start() -> Result<(), String> {

    let listener = TcpListener::bind(LISTEN).map_err(|e| format!("egress proxy on {LISTEN}: {e}"))?;

    std::thread::spawn(move || {

        for conn in listener.incoming().flatten() {

            std::thread::spawn(move || relay(conn));

        }

    });

    Ok(())

}

fn vsock_to_host(port: u32) -> std::io::Result<File> {

    let fd = unsafe { libc::socket(libc::AF_VSOCK, libc::SOCK_STREAM | libc::SOCK_CLOEXEC, 0) };

    if fd < 0 {

        return Err(std::io::Error::last_os_error());

    }

    // Owned from here, so it is closed on every path.

    let file = unsafe { File::from_raw_fd(fd) };

    let mut addr: libc::sockaddr_vm = unsafe { std::mem::zeroed() };

    addr.svm_family = libc::AF_VSOCK as libc::sa_family_t;

    addr.svm_cid = libc::VMADDR_CID_HOST;

    addr.svm_port = port;

    let rc = unsafe {

        libc::connect(

            fd,

            &addr as *const libc::sockaddr_vm as *const libc::sockaddr,

            std::mem::size_of::<libc::sockaddr_vm>() as libc::socklen_t,

        )

    };

    if rc != 0 {

        return Err(std::io::Error::last_os_error());

    }

    Ok(file)

}

fn relay(tcp: TcpStream) {

    let Ok(host) = vsock_to_host(VSOCK_PORT) else {

        let mut tcp = tcp;

        let _ = tcp.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\nthe runner's egress proxy did not answer\n");

        return;

    };

    let (Ok(mut tcp_r), Ok(mut host_w)) = (tcp.try_clone(), host.try_clone()) else { return };

    let up = std::thread::spawn(move || {

        copy(&mut tcp_r, &mut host_w);

        unsafe { libc::shutdown(host_w.as_raw_fd(), libc::SHUT_WR) };

    });

    let (mut host_r, mut tcp_w) = (host, tcp);

    copy(&mut host_r, &mut tcp_w);

    let _ = tcp_w.shutdown(std::net::Shutdown::Write);

    let _ = up.join();

}

fn copy(from: &mut impl Read, to: &mut impl Write) {

    let mut buf = [0u8; 16 * 1024];

    loop {

        match from.read(&mut buf) {

            Ok(0) | Err(_) => return,

            Ok(n) => {

                if to.write_all(&buf[..n]).is_err() {

                    return;

                }

            }

        }

    }

}

//! Allow-listed egress for a microVM job (oxygenflow#55). The VM has no

//! network device: `flow-init` serves a proxy on the guest's 127.0.0.1

//! and relays each connection over vsock to this end, a Unix socket that

//! Firecracker connects for the guest. Here each `CONNECT host:443` is

//! checked against the sandbox's allow-list, the host is resolved (the

//! guest never resolves), any address that is not public is refused, the

//! decision goes into the job log, and the bytes are relayed untouched (no

//! TLS is opened here).

use std::io::{Read, Write};

use std::net::{IpAddr, SocketAddr, TcpStream, ToSocketAddrs};

use std::os::unix::net::{UnixListener, UnixStream};

use std::path::Path;

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

use std::sync::{mpsc, Arc};

use std::time::Duration;

/// The vsock port the guest's relay connects to on the host (CID 2).

pub const VSOCK_PORT: u32 = 3128;

/// Where the job's tools find the proxy inside the guest.

pub const GUEST_PROXY: &str = "http://127.0.0.1:3128";

/// The only port relayed: TLS.

const PORT: u16 = 443;

/// Connections at once, per job.

const MAX_OPEN: usize = 64;

const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// The longest request head read.

const MAX_HEAD: usize = 8 * 1024;

/// What a CONNECT asked for, or why it was refused.

#[derive(Debug, PartialEq, Eq)]

pub enum Ask {

    Connect { host: String, port: u16 },

    Refused(String),

}

/// Read a request head: `CONNECT host:port HTTP/1.x`, and only that.

pub fn parse_head(head: &str) -> Ask {

    let line = head.lines().next().unwrap_or("");

    let mut parts = line.split_whitespace();

    let (method, target, version) = (parts.next().unwrap_or(""), parts.next().unwrap_or(""), parts.next().unwrap_or(""));

    if method != "CONNECT" {

        return Ask::Refused(format!("only https is relayed (CONNECT), not {method:?}"));

    }

    if !version.starts_with("HTTP/1.") {

        return Ask::Refused("not an HTTP/1 request".into());

    }

    let Some((host, port)) = target.rsplit_once(':') else { return Ask::Refused(format!("CONNECT {target:?} names no port")) };

    let Ok(port) = port.parse::<u16>() else { return Ask::Refused(format!("CONNECT {target:?} has no valid port")) };

    let host = host.trim_start_matches('[').trim_end_matches(']').trim_end_matches('.').to_ascii_lowercase();

    if host.is_empty() {

        return Ask::Refused("CONNECT names no host".into());

    }

    Ask::Connect { host, port }

}

/// Whether the job may go to `host:port`: the port is 443, the host is a

/// name (an address is refused: the allow-list names hosts) the list allows.

pub fn decide(allow: &[String], host: &str, port: u16) -> Result<(), String> {

    if port != PORT {

        return Err(format!("only port {PORT} is relayed"));

    }

    if host.parse::<IpAddr>().is_ok() {

        return Err("an address, not a name: the allow-list names hosts".into());

    }

    if !flow_core::egress_allows(allow, host) {

        return Err("not in the sandbox's egress allow-list".into());

    }

    Ok(())

}

/// Whether an address is on the public internet: not loopback, private,

/// shared (CGNAT), link-local (cloud metadata among them), multicast,

/// unspecified or documentation, nor an IPv6 local range or a mapped form.

pub fn public(ip: IpAddr) -> bool {

    match ip {

        IpAddr::V4(v4) => {

            let o = v4.octets();

            !(v4.is_loopback()

                || v4.is_private()

                || v4.is_link_local()

                || v4.is_multicast()

                || v4.is_broadcast()

                || v4.is_unspecified()

                || v4.is_documentation()

                || o[0] == 0

                || (o[0] == 100 && (o[1] & 0xc0) == 64)

                || (o[0] == 192 && o[1] == 0 && o[2] == 0)

                || o[0] >= 240)

        }

        IpAddr::V6(v6) => {

            if let Some(v4) = v6.to_ipv4_mapped() {

                return public(IpAddr::V4(v4));

            }

            let s = v6.segments();

            !(v6.is_loopback()

                || v6.is_unspecified()

                || v6.is_multicast()

                || (s[0] & 0xfe00) == 0xfc00

                || (s[0] & 0xffc0) == 0xfe80

                || (s[0] == 0x64 && s[1] == 0xff9b)

                || (s[0] == 0x2001 && s[1] == 0x0db8))

        }

    }

}

/// The proxy for one job: listening until dropped.

pub struct Proxy {

    stop: Arc<AtomicBool>,

    path: std::path::PathBuf,

}

impl Drop for Proxy {

    fn drop(&mut self) {

        self.stop.store(true, Ordering::SeqCst);

        // Wake the accept loop so it sees the stop.

        let _ = UnixStream::connect(&self.path);

        let _ = std::fs::remove_file(&self.path);

    }

}

/// Listen on `socket` (Firecracker's `uds_path` and `_3128`) for the

/// guest's connections; each decision is sent to `log` as a line.

pub fn start(socket: &Path, allow: Vec<String>, log: mpsc::Sender<String>) -> std::io::Result<Proxy> {

    let _ = std::fs::remove_file(socket);

    let listener = UnixListener::bind(socket)?;

    let stop = Arc::new(AtomicBool::new(false));

    let open = Arc::new(AtomicUsize::new(0));

    let allow = Arc::new(allow);

    let s = stop.clone();

    std::thread::spawn(move || {

        for conn in listener.incoming() {

            if s.load(Ordering::SeqCst) {

                break;

            }

            let Ok(conn) = conn else { continue };

            if open.load(Ordering::SeqCst) >= MAX_OPEN {

                let _ = refuse(conn, "too many connections at once");

                let _ = log.send(format!("egress: refused a connection: {MAX_OPEN} already open"));

                continue;

            }

            open.fetch_add(1, Ordering::SeqCst);

            let (allow, log, open) = (allow.clone(), log.clone(), open.clone());

            std::thread::spawn(move || {

                serve(conn, &allow, &log);

                open.fetch_sub(1, Ordering::SeqCst);

            });

        }

    });

    Ok(Proxy { stop, path: socket.to_path_buf() })

}

fn refuse(mut conn: UnixStream, why: &str) -> std::io::Result<()> {

    let body = format!("{why}\n");

    write!(conn, "HTTP/1.1 403 Forbidden\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len())

}

/// Read up to the blank line that ends a request head.

fn read_head(conn: &mut UnixStream) -> std::io::Result<String> {

    let mut head = Vec::new();

    let mut byte = [0u8; 1];

    while !head.ends_with(b"\r\n\r\n") {

        if head.len() >= MAX_HEAD || conn.read(&mut byte)? == 0 {

            break;

        }

        head.push(byte[0]);

    }

    Ok(String::from_utf8_lossy(&head).into_owned())

}

fn serve(mut conn: UnixStream, allow: &[String], log: &mpsc::Sender<String>) {

    let _ = conn.set_read_timeout(Some(Duration::from_secs(30)));

    let head = match read_head(&mut conn) {

        Ok(h) => h,

        Err(_) => return,

    };

    let (host, port) = match parse_head(&head) {

        Ask::Connect { host, port } => (host, port),

        Ask::Refused(why) => {

            let _ = log.send(format!("egress: refused: {why}"));

            let _ = refuse(conn, &why);

            return;

        }

    };

    if let Err(why) = decide(allow, &host, port) {

        let _ = log.send(format!("egress: refused {host}:{port}: {why}"));

        let _ = refuse(conn, &format!("{host}:{port}: {why}"));

        return;

    }

    let addrs: Vec<SocketAddr> = match (host.as_str(), port).to_socket_addrs() {

        Ok(a) => a.collect(),

        Err(e) => {

            let _ = log.send(format!("egress: refused {host}:{port}: it could not be resolved ({e})"));

            let _ = refuse(conn, &format!("{host} could not be resolved"));

            return;

        }

    };

    if let Some(bad) = addrs.iter().find(|a| !public(a.ip())) {

        let why = format!("it resolves to {}, which is not a public address", bad.ip());

        let _ = log.send(format!("egress: refused {host}:{port}: {why}"));

        let _ = refuse(conn, &format!("{host}: {why}"));

        return;

    }

    let out = addrs.iter().find_map(|a| TcpStream::connect_timeout(a, CONNECT_TIMEOUT).ok());

    let Some(out) = out else {

        let _ = log.send(format!("egress: {host}:{port} is allowed but did not answer"));

        let _ = conn.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n");

        return;

    };

    let _ = log.send(format!("egress: allowed {host}:{port}"));

    if conn.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n").is_err() {

        return;

    }

    let _ = conn.set_read_timeout(None);

    relay(conn, out);

}

/// Copy both ways until either side closes.

fn relay(guest: UnixStream, out: TcpStream) {

    let (Ok(mut guest_r), Ok(mut out_w)) = (guest.try_clone(), out.try_clone()) else { return };

    let up = std::thread::spawn(move || {

        let _ = std::io::copy(&mut guest_r, &mut out_w);

        let _ = out_w.shutdown(std::net::Shutdown::Write);

    });

    let (mut out_r, mut guest_w) = (out, guest);

    let _ = std::io::copy(&mut out_r, &mut guest_w);

    let _ = guest_w.shutdown(std::net::Shutdown::Write);

    let _ = up.join();

}

#[cfg(test)]

mod tests {

    use super::*;

    fn allow() -> Vec<String> {

        vec!["*.googleapis.com".into(), "registry.opentofu.org".into()]

    }

    #[test]

    fn only_connect_to_an_allowed_name_on_443_is_let_through() {

        assert_eq!(

            parse_head("CONNECT storage.googleapis.com:443 HTTP/1.1\r\nHost: x\r\n\r\n"),

            Ask::Connect { host: "storage.googleapis.com".into(), port: 443 }

        );

        assert!(matches!(parse_head("GET http://example.com/ HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("only https")));

        assert!(matches!(parse_head("CONNECT example.com HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("no port")));

        assert_eq!(decide(&allow(), "storage.googleapis.com", 443), Ok(()));

        assert_eq!(decide(&allow(), "registry.opentofu.org", 443), Ok(()));

        assert!(decide(&allow(), "example.com", 443).unwrap_err().contains("allow-list"));

        assert!(decide(&allow(), "registry.opentofu.org", 22).unwrap_err().contains("only port 443"));

        assert!(decide(&allow(), "142.250.1.1", 443).unwrap_err().contains("names hosts"));

    }

    #[test]

    fn private_addresses_are_not_public() {

        for ip in [

            "127.0.0.1",

            "10.1.2.3",

            "192.168.1.1",

            "172.16.0.1",

            "169.254.169.254",

            "100.64.0.1",

            "0.0.0.0",

            "::1",

            "fd00::1",

            "fe80::1",

            "::ffff:10.0.0.1",

        ] {

            assert!(!public(ip.parse().unwrap()), "{ip}");

        }

        for ip in ["142.250.1.1", "2a00:1450:4009::200e"] {

            assert!(public(ip.parse().unwrap()), "{ip}");

        }

    }

    #[test]

    fn the_proxy_refuses_and_logs_a_host_not_allowed() {

        let dir = tempfile::tempdir().unwrap();

        let sock = dir.path().join("v.sock_3128");

        let (tx, rx) = mpsc::channel();

        let proxy = start(&sock, allow(), tx).unwrap();

        let ask = |head: &str| {

            let mut c = UnixStream::connect(&sock).unwrap();

            c.write_all(head.as_bytes()).unwrap();

            let mut answer = String::new();

            c.read_to_string(&mut answer).unwrap();

            answer

        };

        let a = ask("CONNECT example.com:443 HTTP/1.1\r\n\r\n");

        assert!(a.starts_with("HTTP/1.1 403") && a.contains("not in the sandbox's egress allow-list"), "{a}");

        assert_eq!(

            rx.recv_timeout(Duration::from_secs(5)).unwrap(),

            "egress: refused example.com:443: not in the sandbox's egress allow-list"

        );

        let a = ask("GET http://storage.googleapis.com/ HTTP/1.1\r\n\r\n");

        assert!(a.starts_with("HTTP/1.1 403"), "{a}");

        assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().contains("only https"));

        drop(proxy);

        assert!(!sock.exists(), "the socket goes with the job");

    }

}

    // The hosts egress may reach, sorted; absent without egress.

    if !s.egress_allow.is_empty() {

        fields.push((8, texts(&s.egress_allow)));

    }

    arch_ok, egress_allows, egress_pattern_ok, Identity, Job, Mount, MountSource, Network, Output, OutputKind, Pipeline, Problem, Problems,

    Runtime, Sandbox, Secret, EGRESS_LABEL, PIPELINE_PATH,

    /// The hosts a `network "egress"` sandbox may reach (oxygenflow#55):

    /// names, or `*.` and a domain for any name under it; sorted. Its jobs

    /// reach them through the runner's proxy, and nothing else.

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub egress_allow: Vec<String>,

    let mut egress_allow = Vec::new();

        }

        if key != "network" {

            ctx.no_children(c);

                let kind = ctx.one_string(c);

                if kind.as_deref() != Some("egress") {

                    ctx.no_children(c);

                }

                network = match kind.as_deref() {

                    // Allow-listed egress (oxygenflow#55): still no network

                    // device; a proxy in the guest relays to the runner, which

                    // lets through only the hosts named here.

                        egress_allow = parse_egress(ctx, c);

                        Some(Network::Egress)

    if network == Some(Network::Egress) {

        if runtime == Some(Runtime::Syscall) {

            ctx.err(node.span(), format!("sandbox {name:?}: network \"egress\" needs runtime \"microvm\""));

        }

        // Only a runner that allows egress offers this label.

        if !runs_on.iter().any(|l| l == EGRESS_LABEL) {

            runs_on.push(EGRESS_LABEL.to_string());

            runs_on.sort();

        }

    }

        egress_allow,

}

/// The label a runner offers when it allows egress, and an egress sandbox

/// needs (oxygenflow#55).

pub const EGRESS_LABEL: &str = "egress";

/// The most hosts one sandbox may name.

const MAX_EGRESS_HOSTS: usize = 32;

/// Whether `pattern` is a host name, or `*.` and a domain: lower-case

/// letters, digits, `-` and dots, at least two labels.

pub fn egress_pattern_ok(pattern: &str) -> bool {

    let name = pattern.strip_prefix("*.").unwrap_or(pattern);

    let labels: Vec<&str> = name.split('.').collect();

    name.len() <= 253

        && labels.len() >= 2

        && labels.iter().all(|l| {

            !l.is_empty()

                && l.len() <= 63

                && !l.starts_with('-')

                && !l.ends_with('-')

                && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')

        })

}

/// Whether `host` is one `patterns` allow: a name exactly, or a name under

/// a `*.` domain (not the domain itself). Case and a final dot are ignored.

pub fn egress_allows(patterns: &[String], host: &str) -> bool {

    let host = host.trim_end_matches('.').to_ascii_lowercase();

    patterns.iter().any(|p| match p.strip_prefix("*.") {

        Some(domain) => host.len() > domain.len() + 1 && host.ends_with(domain) && host[..host.len() - domain.len()].ends_with('.'),

        None => host == *p,

    })

}

/// `network "egress" { allow "*.googleapis.com" "registry.opentofu.org" }`.

fn parse_egress(ctx: &mut Ctx, node: &KdlNode) -> Vec<String> {

    let mut out = Vec::new();

    for c in block(ctx, node) {

        ctx.no_children(c);

        if c.name().value() != "allow" {

            ctx.err(c.span(), format!("unknown egress setting `{}`; egress takes `allow \"host\"...`", c.name().value()));

            continue;

        }

        for h in ctx.strings(c) {

            if egress_pattern_ok(&h) {

                out.push(h);

            } else {

                ctx.err(

                    c.span(),

                    format!("allow {h:?}: expected a lower-case host name such as \"registry.opentofu.org\", or \"*.\" and a domain"),

                );

            }

        }

    }

    out.sort();

    out.dedup();

    if out.is_empty() {

        ctx.err(node.span(), "network \"egress\" needs at least one `allow \"host\"`");

    } else if out.len() > MAX_EGRESS_HOSTS {

        ctx.err(node.span(), format!("network \"egress\" may allow at most {MAX_EGRESS_HOSTS} hosts"));

    }

    out

    let src = format!("schema-version 1\nsandbox \"s\" {{\n  runtime \"wasm\"\n  rootfs \"cas://blob/{ROOTFS}\"\n  network \"wifi\"\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n");

fn loopback_and_allow_listed_egress() {

        format!("schema-version 1\nsandbox \"s\" {{\n  rootfs \"cas://blob/{ROOTFS}\"\n  network {net}\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n")

    let p = Pipeline::parse(&src("\"loopback\"")).unwrap();

    assert_ne!(p.hash(), Pipeline::parse(&src("\"isolated\"")).unwrap().hash(), "the network is part of the sandbox");

    // Egress names its hosts (oxygenflow#55), sorted; the sandbox then needs

    // a runner offering the egress label.

    let p =

        Pipeline::parse(&src("\"egress\" { allow \"registry.opentofu.org\" \"*.googleapis.com\"; allow \"*.googleapis.com\"; }")).unwrap();

    let s = &p.sandboxes["s"];

    assert_eq!(s.network, Network::Egress);

    assert_eq!(s.egress_allow, ["*.googleapis.com", "registry.opentofu.org"]);

    assert_eq!(s.runs_on, [crate::EGRESS_LABEL]);

    let other = Pipeline::parse(&src("\"egress\" { allow \"registry.opentofu.org\"; }")).unwrap();

    assert_ne!(p.hash(), other.hash(), "the allow-list is part of the sandbox");

    let has = |net: &str, want: &str| {

        let ps = problems(&src(net));

        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{net}: {ps:?}");

    };

    has("\"egress\"", "needs a block");

    has("\"egress\" { }", "at least one `allow");

    has("\"egress\" { allow \"Example.COM\"; }", "lower-case host name");

    has("\"egress\" { allow \"localhost\"; }", "lower-case host name");

    has("\"egress\" { allow \"example.com:8443\"; }", "lower-case host name");

    has("\"egress\" { deny \"example.com\"; }", "unknown egress setting");

    has("\"loopback\" { allow \"example.com\"; }", "takes no block");

    let syscall = format!(

        "schema-version 1\nsandbox \"s\" {{\n  runtime \"syscall\"\n  rootfs \"cas://blob/{ROOTFS}\"\n  network \"egress\" {{ allow \"example.com\"; }}\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n"

    );

    assert!(problems(&syscall).iter().any(|(_, m)| m.contains("needs runtime \"microvm\"")));

}

#[test]

fn egress_patterns_match_names_and_names_under_a_domain() {

    let allow = vec!["*.googleapis.com".to_string(), "registry.opentofu.org".to_string()];

    for ok in ["storage.googleapis.com", "a.b.googleapis.com", "REGISTRY.opentofu.org.", "registry.opentofu.org"] {

        assert!(crate::egress_allows(&allow, ok), "{ok}");

    }

    for no in ["googleapis.com", "evilgoogleapis.com", "googleapis.com.evil.org", "opentofu.org", "x.registry.opentofu.org", ""] {

        assert!(!crate::egress_allows(&allow, no), "{no}");

    }

mod egress;

    /// `network "egress"` (oxygenflow#55): `lo` comes up and a proxy on

    /// 127.0.0.1:3128 relays over vsock to the runner.

    #[serde(default)]

    egress: bool,

    if job.loopback || job.egress {

    }

    if job.egress {

        egress::start()?;

    /// Let microVM jobs reach the hosts their sandbox allows, through this

    /// runner (oxygenflow#55); off unless set. The runner then offers the

    /// `egress` label, which such jobs need.

    #[serde(default)]

    pub egress: bool,

}

impl RunnerConfig {

    /// The labels offered to the hub: the configured ones, and `egress`

    /// when egress is on.

    pub fn offered_labels(&self) -> Vec<String> {

        let mut l = self.labels.clone();

        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {

            l.push(flow_core::EGRESS_LABEL.to_string());

        }

        l

    }

mod egress;

    // Allow-listed egress (oxygenflow#55): a vsock device and the runner's

    // proxy behind it; the guest relays 127.0.0.1:3128 to it.

    let egress = sandbox.network == flow_core::Network::Egress;

    let (egress_tx, egress_rx) = mpsc::channel::<String>();

    let mut _proxy = None;

    if egress {

        if !home.config.runner.as_ref().is_some_and(|r| r.egress) {

            bail!("this runner does not allow network egress (a runner allows it with `egress = true` under [runner] in flow.toml)");

        }

        _proxy = Some(

            crate::egress::start(&scratch.join(format!("v.sock_{}", crate::egress::VSOCK_PORT)), sandbox.egress_allow.clone(), egress_tx)

                .context("start the egress proxy")?,

        );

        spec["egress"] = serde_json::Value::Bool(true);

        add_proxy_env(&mut spec);

        sink.record(Record {

            t: 0,

            stream: "flow".into(),

            text: format!("egress: through this runner, to {} only", sandbox.egress_allow.join(", ")),

        });

    }

    let mut vm = vm;

    if egress {

        // Relative to the VMM's working directory (the scratch directory),

        // which keeps the socket path short.

        vm["vsock"] = serde_json::json!({ "guest_cid": 3, "uds_path": "v.sock" });

    }

        }

        while let Ok(line) = egress_rx.try_recv() {

            sink.record(Record { t: t(&started), stream: "flow".into(), text: line });

/// The proxy variables a job with egress gets, in both spellings tools

/// read, unless the job sets them itself.

fn add_proxy_env(spec: &mut serde_json::Value) {

    let Some(env) = spec["env"].as_array_mut() else { return };

    let set: Vec<String> = env.iter().filter_map(|kv| kv[0].as_str().map(String::from)).collect();

    for (k, v) in [

        ("HTTPS_PROXY", crate::egress::GUEST_PROXY),

        ("https_proxy", crate::egress::GUEST_PROXY),

        ("HTTP_PROXY", crate::egress::GUEST_PROXY),

        ("http_proxy", crate::egress::GUEST_PROXY),

        ("NO_PROXY", "localhost,127.0.0.1"),

        ("no_proxy", "localhost,127.0.0.1"),

    ] {

        if !set.iter().any(|s| s == k) {

            env.push(serde_json::json!([k, v]));

        }

    }

}

    #[test]

    fn a_job_with_egress_gets_the_proxy_variables_unless_it_sets_them() {

        let mut spec = serde_json::json!({ "env": [["HTTPS_PROXY", "http://mine:1"]] });

        super::add_proxy_env(&mut spec);

        let env = spec["env"].as_array().unwrap();

        let get = |k: &str| env.iter().find(|kv| kv[0] == k).map(|kv| kv[1].as_str().unwrap().to_string());

        assert_eq!(get("HTTPS_PROXY").as_deref(), Some("http://mine:1"), "the job's own is kept");

        assert_eq!(get("https_proxy").as_deref(), Some("http://127.0.0.1:3128"));

        assert_eq!(get("NO_PROXY").as_deref(), Some("localhost,127.0.0.1"));

    }

                    ..RunnerInfo::here(cfg.offered_labels(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)

- **Root:** PID 1 binds the unpacked tree read-only as the new root, the workspace at `/src`, tmpfs at `/tmp`, `/run` and `/run/secrets`, a fresh `/proc`, and a `/dev` of `null`, `zero`, `full`, `random` and `urandom` with `/dev/shm`. It brings `lo` up for `network "loopback"` and `"egress"` (with the vsock relay of spec §4.3.1 for the latter), pivots into the root and detaches the old one, so the host's files, the spec and the scratch directory are out of reach.

    network "isolated"      // Options: "isolated" | "loopback" | "egress" { allow "host"... } (§4.3.1)

#### 4.3.1 Allow-listed egress (oxygenflow#55)

A sandbox reaches the network only by naming its hosts:

```kdl

sandbox "deploy" {

    runtime "microvm"

    network "egress" { allow "*.googleapis.com" "registry.opentofu.org" }

}

```

The VM still has no network device. `flow-init` serves a proxy on the

guest's `127.0.0.1:3128` (the job gets `HTTPS_PROXY`, `HTTP_PROXY` and

their lower-case forms, unless it sets them) and relays each connection,

unread, over Firecracker's vsock to the runner. The runner accepts only

`CONNECT host:443` for a host the list allows (a name exactly, or any

name under a `*.` domain, never the domain itself; never an address),

resolves it itself (the guest does not resolve), refuses any name that

resolves to an address that is not public, and relays the bytes without

opening the TLS. Every decision, allowed or refused, is a `flow` line in

the job's log. Patterns are lower-case host names of two or more labels,

at most 32 per sandbox, and part of the sandbox's hash.

Egress is off unless a runner turns it on (`egress = true` under

`[runner]` in `flow.toml`); such a runner offers the `egress` label, and

an egress sandbox needs it, so the hub hands its jobs only to runners

that allow it, and a runner without it refuses one anyway. The syscall

runtime has no egress yet.