OXYGENHUB

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

flow-core parses the allow-list (lower-case names or *. domains, at most 32,
sorted, hashed with the sandbox; microvm only) and gives an egress sandbox
the egress label. The VM keeps no network device: flow-init relays
127.0.0.1:3128 over vsock to the runner, whose proxy takes only CONNECT
host:443 for an allowed name, resolves it, refuses non-public addresses,
logs every decision into the job log and relays the bytes untouched. The
job gets HTTPS_PROXY and friends. Off unless a runner sets egress = true
under [runner] (it then offers the egress label); a runner without it
refuses such a job. Spec 4.3.1. Tests: parsing and patterns (flow-core),
CONNECT handling, allow/deny, private addresses, a refusal logged (runner).

Refs #55
Author
Andy McCall @andymccall <[email protected]>
via
agent Claude Code <[email protected]> · delegate for changes, until 2026-11-13 19:00Z · key ea5f78160088
When
· 2026-10-16 02:42 UTC
Signature
✔ signed by Claude Code
Hash
ea81796bb757d87bf961a6e92044f84febf655d90b08c665c815d665fbaab873
Depends on
02b127d2c721 1abadd4a2fa6 3007d8f011fa 62829d15ef09 6d53fb7d00ec 8f211e20bea5 a083cc802892 a72abeeedd78 b14aeaae683a bf7bb1b80fa3 d60aac6c3f8a e0918b887850 e2c125610f43 e576e08617dc ea56c56eaf90 eb5c4db673e5 fc10d9e80383
Depended on by
3554afefb6aa 85275e1ef3b7 5cf1e8aefbb6 4be5e4cd0af6 077a65242359 35f7a849bcc5 46fe2c815cc3
In bookmarks
main
In tags
v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits
40 atoms, 2 tree operations, 13 files

Browse the tree at this change

added crates/flow-init/src/egress.rs +80 −0
1 +
//! The guest's half of allow-listed egress (oxygenflow#55): a proxy on
2 +
//! 127.0.0.1:3128 that relays each connection, unread, over vsock to the
3 +
//! runner (host CID 2), which decides what may pass. The VM has no network
4 +
//! device; this is its only way out.
5 +
6 +
use std::fs::File;
7 +
use std::io::{Read, Write};
8 +
use std::net::{TcpListener, TcpStream};
9 +
use std::os::fd::{AsRawFd, FromRawFd};
10 +
11 +
/// Where the job's tools reach the proxy, and the vsock port behind it.
12 +
const LISTEN: &str = "127.0.0.1:3128";
13 +
const VSOCK_PORT: u32 = 3128;
14 +
15 +
/// Start relaying in the background (before the job starts, so its first
16 +
/// connection finds the proxy listening).
17 +
pub fn start() -> Result<(), String> {
18 +
    let listener = TcpListener::bind(LISTEN).map_err(|e| format!("egress proxy on {LISTEN}: {e}"))?;
19 +
    std::thread::spawn(move || {
20 +
        for conn in listener.incoming().flatten() {
21 +
            std::thread::spawn(move || relay(conn));
22 +
        }
23 +
    });
24 +
    Ok(())
25 +
}
26 +
27 +
fn vsock_to_host(port: u32) -> std::io::Result<File> {
28 +
    let fd = unsafe { libc::socket(libc::AF_VSOCK, libc::SOCK_STREAM | libc::SOCK_CLOEXEC, 0) };
29 +
    if fd < 0 {
30 +
        return Err(std::io::Error::last_os_error());
31 +
    }
32 +
    // Owned from here, so it is closed on every path.
33 +
    let file = unsafe { File::from_raw_fd(fd) };
34 +
    let mut addr: libc::sockaddr_vm = unsafe { std::mem::zeroed() };
35 +
    addr.svm_family = libc::AF_VSOCK as libc::sa_family_t;
36 +
    addr.svm_cid = libc::VMADDR_CID_HOST;
37 +
    addr.svm_port = port;
38 +
    let rc = unsafe {
39 +
        libc::connect(
40 +
            fd,
41 +
            &addr as *const libc::sockaddr_vm as *const libc::sockaddr,
42 +
            std::mem::size_of::<libc::sockaddr_vm>() as libc::socklen_t,
43 +
        )
44 +
    };
45 +
    if rc != 0 {
46 +
        return Err(std::io::Error::last_os_error());
47 +
    }
48 +
    Ok(file)
49 +
}
50 +
51 +
fn relay(tcp: TcpStream) {
52 +
    let Ok(host) = vsock_to_host(VSOCK_PORT) else {
53 +
        let mut tcp = tcp;
54 +
        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");
55 +
        return;
56 +
    };
57 +
    let (Ok(mut tcp_r), Ok(mut host_w)) = (tcp.try_clone(), host.try_clone()) else { return };
58 +
    let up = std::thread::spawn(move || {
59 +
        copy(&mut tcp_r, &mut host_w);
60 +
        unsafe { libc::shutdown(host_w.as_raw_fd(), libc::SHUT_WR) };
61 +
    });
62 +
    let (mut host_r, mut tcp_w) = (host, tcp);
63 +
    copy(&mut host_r, &mut tcp_w);
64 +
    let _ = tcp_w.shutdown(std::net::Shutdown::Write);
65 +
    let _ = up.join();
66 +
}
67 +
68 +
fn copy(from: &mut impl Read, to: &mut impl Write) {
69 +
    let mut buf = [0u8; 16 * 1024];
70 +
    loop {
71 +
        match from.read(&mut buf) {
72 +
            Ok(0) | Err(_) => return,
73 +
            Ok(n) => {
74 +
                if to.write_all(&buf[..n]).is_err() {
75 +
                    return;
76 +
                }
77 +
            }
78 +
        }
79 +
    }
80 +
}
added crates/flow/src/egress.rs +301 −0
1 +
//! Allow-listed egress for a microVM job (oxygenflow#55). The VM has no
2 +
//! network device: `flow-init` serves a proxy on the guest's 127.0.0.1
3 +
//! and relays each connection over vsock to this end, a Unix socket that
4 +
//! Firecracker connects for the guest. Here each `CONNECT host:443` is
5 +
//! checked against the sandbox's allow-list, the host is resolved (the
6 +
//! guest never resolves), any address that is not public is refused, the
7 +
//! decision goes into the job log, and the bytes are relayed untouched (no
8 +
//! TLS is opened here).
9 +
10 +
use std::io::{Read, Write};
11 +
use std::net::{IpAddr, SocketAddr, TcpStream, ToSocketAddrs};
12 +
use std::os::unix::net::{UnixListener, UnixStream};
13 +
use std::path::Path;
14 +
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
15 +
use std::sync::{mpsc, Arc};
16 +
use std::time::Duration;
17 +
18 +
/// The vsock port the guest's relay connects to on the host (CID 2).
19 +
pub const VSOCK_PORT: u32 = 3128;
20 +
/// Where the job's tools find the proxy inside the guest.
21 +
pub const GUEST_PROXY: &str = "http://127.0.0.1:3128";
22 +
/// The only port relayed: TLS.
23 +
const PORT: u16 = 443;
24 +
/// Connections at once, per job.
25 +
const MAX_OPEN: usize = 64;
26 +
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
27 +
/// The longest request head read.
28 +
const MAX_HEAD: usize = 8 * 1024;
29 +
30 +
/// What a CONNECT asked for, or why it was refused.
31 +
#[derive(Debug, PartialEq, Eq)]
32 +
pub enum Ask {
33 +
    Connect { host: String, port: u16 },
34 +
    Refused(String),
35 +
}
36 +
37 +
/// Read a request head: `CONNECT host:port HTTP/1.x`, and only that.
38 +
pub fn parse_head(head: &str) -> Ask {
39 +
    let line = head.lines().next().unwrap_or("");
40 +
    let mut parts = line.split_whitespace();
41 +
    let (method, target, version) = (parts.next().unwrap_or(""), parts.next().unwrap_or(""), parts.next().unwrap_or(""));
42 +
    if method != "CONNECT" {
43 +
        return Ask::Refused(format!("only https is relayed (CONNECT), not {method:?}"));
44 +
    }
45 +
    if !version.starts_with("HTTP/1.") {
46 +
        return Ask::Refused("not an HTTP/1 request".into());
47 +
    }
48 +
    let Some((host, port)) = target.rsplit_once(':') else { return Ask::Refused(format!("CONNECT {target:?} names no port")) };
49 +
    let Ok(port) = port.parse::<u16>() else { return Ask::Refused(format!("CONNECT {target:?} has no valid port")) };
50 +
    let host = host.trim_start_matches('[').trim_end_matches(']').trim_end_matches('.').to_ascii_lowercase();
51 +
    if host.is_empty() {
52 +
        return Ask::Refused("CONNECT names no host".into());
53 +
    }
54 +
    Ask::Connect { host, port }
55 +
}
56 +
57 +
/// Whether the job may go to `host:port`: the port is 443, the host is a
58 +
/// name (an address is refused: the allow-list names hosts) the list allows.
59 +
pub fn decide(allow: &[String], host: &str, port: u16) -> Result<(), String> {
60 +
    if port != PORT {
61 +
        return Err(format!("only port {PORT} is relayed"));
62 +
    }
63 +
    if host.parse::<IpAddr>().is_ok() {
64 +
        return Err("an address, not a name: the allow-list names hosts".into());
65 +
    }
66 +
    if !flow_core::egress_allows(allow, host) {
67 +
        return Err("not in the sandbox's egress allow-list".into());
68 +
    }
69 +
    Ok(())
70 +
}
71 +
72 +
/// Whether an address is on the public internet: not loopback, private,
73 +
/// shared (CGNAT), link-local (cloud metadata among them), multicast,
74 +
/// unspecified or documentation, nor an IPv6 local range or a mapped form.
75 +
pub fn public(ip: IpAddr) -> bool {
76 +
    match ip {
77 +
        IpAddr::V4(v4) => {
78 +
            let o = v4.octets();
79 +
            !(v4.is_loopback()
80 +
                || v4.is_private()
81 +
                || v4.is_link_local()
82 +
                || v4.is_multicast()
83 +
                || v4.is_broadcast()
84 +
                || v4.is_unspecified()
85 +
                || v4.is_documentation()
86 +
                || o[0] == 0
87 +
                || (o[0] == 100 && (o[1] & 0xc0) == 64)
88 +
                || (o[0] == 192 && o[1] == 0 && o[2] == 0)
89 +
                || o[0] >= 240)
90 +
        }
91 +
        IpAddr::V6(v6) => {
92 +
            if let Some(v4) = v6.to_ipv4_mapped() {
93 +
                return public(IpAddr::V4(v4));
94 +
            }
95 +
            let s = v6.segments();
96 +
            !(v6.is_loopback()
97 +
                || v6.is_unspecified()
98 +
                || v6.is_multicast()
99 +
                || (s[0] & 0xfe00) == 0xfc00
100 +
                || (s[0] & 0xffc0) == 0xfe80
101 +
                || (s[0] == 0x64 && s[1] == 0xff9b)
102 +
                || (s[0] == 0x2001 && s[1] == 0x0db8))
103 +
        }
104 +
    }
105 +
}
106 +
107 +
/// The proxy for one job: listening until dropped.
108 +
pub struct Proxy {
109 +
    stop: Arc<AtomicBool>,
110 +
    path: std::path::PathBuf,
111 +
}
112 +
113 +
impl Drop for Proxy {
114 +
    fn drop(&mut self) {
115 +
        self.stop.store(true, Ordering::SeqCst);
116 +
        // Wake the accept loop so it sees the stop.
117 +
        let _ = UnixStream::connect(&self.path);
118 +
        let _ = std::fs::remove_file(&self.path);
119 +
    }
120 +
}
121 +
122 +
/// Listen on `socket` (Firecracker's `uds_path` and `_3128`) for the
123 +
/// guest's connections; each decision is sent to `log` as a line.
124 +
pub fn start(socket: &Path, allow: Vec<String>, log: mpsc::Sender<String>) -> std::io::Result<Proxy> {
125 +
    let _ = std::fs::remove_file(socket);
126 +
    let listener = UnixListener::bind(socket)?;
127 +
    let stop = Arc::new(AtomicBool::new(false));
128 +
    let open = Arc::new(AtomicUsize::new(0));
129 +
    let allow = Arc::new(allow);
130 +
    let s = stop.clone();
131 +
    std::thread::spawn(move || {
132 +
        for conn in listener.incoming() {
133 +
            if s.load(Ordering::SeqCst) {
134 +
                break;
135 +
            }
136 +
            let Ok(conn) = conn else { continue };
137 +
            if open.load(Ordering::SeqCst) >= MAX_OPEN {
138 +
                let _ = refuse(conn, "too many connections at once");
139 +
                let _ = log.send(format!("egress: refused a connection: {MAX_OPEN} already open"));
140 +
                continue;
141 +
            }
142 +
            open.fetch_add(1, Ordering::SeqCst);
143 +
            let (allow, log, open) = (allow.clone(), log.clone(), open.clone());
144 +
            std::thread::spawn(move || {
145 +
                serve(conn, &allow, &log);
146 +
                open.fetch_sub(1, Ordering::SeqCst);
147 +
            });
148 +
        }
149 +
    });
150 +
    Ok(Proxy { stop, path: socket.to_path_buf() })
151 +
}
152 +
153 +
fn refuse(mut conn: UnixStream, why: &str) -> std::io::Result<()> {
154 +
    let body = format!("{why}\n");
155 +
    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())
156 +
}
157 +
158 +
/// Read up to the blank line that ends a request head.
159 +
fn read_head(conn: &mut UnixStream) -> std::io::Result<String> {
160 +
    let mut head = Vec::new();
161 +
    let mut byte = [0u8; 1];
162 +
    while !head.ends_with(b"\r\n\r\n") {
163 +
        if head.len() >= MAX_HEAD || conn.read(&mut byte)? == 0 {
164 +
            break;
165 +
        }
166 +
        head.push(byte[0]);
167 +
    }
168 +
    Ok(String::from_utf8_lossy(&head).into_owned())
169 +
}
170 +
171 +
fn serve(mut conn: UnixStream, allow: &[String], log: &mpsc::Sender<String>) {
172 +
    let _ = conn.set_read_timeout(Some(Duration::from_secs(30)));
173 +
    let head = match read_head(&mut conn) {
174 +
        Ok(h) => h,
175 +
        Err(_) => return,
176 +
    };
177 +
    let (host, port) = match parse_head(&head) {
178 +
        Ask::Connect { host, port } => (host, port),
179 +
        Ask::Refused(why) => {
180 +
            let _ = log.send(format!("egress: refused: {why}"));
181 +
            let _ = refuse(conn, &why);
182 +
            return;
183 +
        }
184 +
    };
185 +
    if let Err(why) = decide(allow, &host, port) {
186 +
        let _ = log.send(format!("egress: refused {host}:{port}: {why}"));
187 +
        let _ = refuse(conn, &format!("{host}:{port}: {why}"));
188 +
        return;
189 +
    }
190 +
    let addrs: Vec<SocketAddr> = match (host.as_str(), port).to_socket_addrs() {
191 +
        Ok(a) => a.collect(),
192 +
        Err(e) => {
193 +
            let _ = log.send(format!("egress: refused {host}:{port}: it could not be resolved ({e})"));
194 +
            let _ = refuse(conn, &format!("{host} could not be resolved"));
195 +
            return;
196 +
        }
197 +
    };
198 +
    if let Some(bad) = addrs.iter().find(|a| !public(a.ip())) {
199 +
        let why = format!("it resolves to {}, which is not a public address", bad.ip());
200 +
        let _ = log.send(format!("egress: refused {host}:{port}: {why}"));
201 +
        let _ = refuse(conn, &format!("{host}: {why}"));
202 +
        return;
203 +
    }
204 +
    let out = addrs.iter().find_map(|a| TcpStream::connect_timeout(a, CONNECT_TIMEOUT).ok());
205 +
    let Some(out) = out else {
206 +
        let _ = log.send(format!("egress: {host}:{port} is allowed but did not answer"));
207 +
        let _ = conn.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n");
208 +
        return;
209 +
    };
210 +
    let _ = log.send(format!("egress: allowed {host}:{port}"));
211 +
    if conn.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n").is_err() {
212 +
        return;
213 +
    }
214 +
    let _ = conn.set_read_timeout(None);
215 +
    relay(conn, out);
216 +
}
217 +
218 +
/// Copy both ways until either side closes.
219 +
fn relay(guest: UnixStream, out: TcpStream) {
220 +
    let (Ok(mut guest_r), Ok(mut out_w)) = (guest.try_clone(), out.try_clone()) else { return };
221 +
    let up = std::thread::spawn(move || {
222 +
        let _ = std::io::copy(&mut guest_r, &mut out_w);
223 +
        let _ = out_w.shutdown(std::net::Shutdown::Write);
224 +
    });
225 +
    let (mut out_r, mut guest_w) = (out, guest);
226 +
    let _ = std::io::copy(&mut out_r, &mut guest_w);
227 +
    let _ = guest_w.shutdown(std::net::Shutdown::Write);
228 +
    let _ = up.join();
229 +
}
230 +
231 +
#[cfg(test)]
232 +
mod tests {
233 +
    use super::*;
234 +
235 +
    fn allow() -> Vec<String> {
236 +
        vec!["*.googleapis.com".into(), "registry.opentofu.org".into()]
237 +
    }
238 +
239 +
    #[test]
240 +
    fn only_connect_to_an_allowed_name_on_443_is_let_through() {
241 +
        assert_eq!(
242 +
            parse_head("CONNECT storage.googleapis.com:443 HTTP/1.1\r\nHost: x\r\n\r\n"),
243 +
            Ask::Connect { host: "storage.googleapis.com".into(), port: 443 }
244 +
        );
245 +
        assert!(matches!(parse_head("GET http://example.com/ HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("only https")));
246 +
        assert!(matches!(parse_head("CONNECT example.com HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("no port")));
247 +
        assert_eq!(decide(&allow(), "storage.googleapis.com", 443), Ok(()));
248 +
        assert_eq!(decide(&allow(), "registry.opentofu.org", 443), Ok(()));
249 +
        assert!(decide(&allow(), "example.com", 443).unwrap_err().contains("allow-list"));
250 +
        assert!(decide(&allow(), "registry.opentofu.org", 22).unwrap_err().contains("only port 443"));
251 +
        assert!(decide(&allow(), "142.250.1.1", 443).unwrap_err().contains("names hosts"));
252 +
    }
253 +
254 +
    #[test]
255 +
    fn private_addresses_are_not_public() {
256 +
        for ip in [
257 +
            "127.0.0.1",
258 +
            "10.1.2.3",
259 +
            "192.168.1.1",
260 +
            "172.16.0.1",
261 +
            "169.254.169.254",
262 +
            "100.64.0.1",
263 +
            "0.0.0.0",
264 +
            "::1",
265 +
            "fd00::1",
266 +
            "fe80::1",
267 +
            "::ffff:10.0.0.1",
268 +
        ] {
269 +
            assert!(!public(ip.parse().unwrap()), "{ip}");
270 +
        }
271 +
        for ip in ["142.250.1.1", "2a00:1450:4009::200e"] {
272 +
            assert!(public(ip.parse().unwrap()), "{ip}");
273 +
        }
274 +
    }
275 +
276 +
    #[test]
277 +
    fn the_proxy_refuses_and_logs_a_host_not_allowed() {
278 +
        let dir = tempfile::tempdir().unwrap();
279 +
        let sock = dir.path().join("v.sock_3128");
280 +
        let (tx, rx) = mpsc::channel();
281 +
        let proxy = start(&sock, allow(), tx).unwrap();
282 +
        let ask = |head: &str| {
283 +
            let mut c = UnixStream::connect(&sock).unwrap();
284 +
            c.write_all(head.as_bytes()).unwrap();
285 +
            let mut answer = String::new();
286 +
            c.read_to_string(&mut answer).unwrap();
287 +
            answer
288 +
        };
289 +
        let a = ask("CONNECT example.com:443 HTTP/1.1\r\n\r\n");
290 +
        assert!(a.starts_with("HTTP/1.1 403") && a.contains("not in the sandbox's egress allow-list"), "{a}");
291 +
        assert_eq!(
292 +
            rx.recv_timeout(Duration::from_secs(5)).unwrap(),
293 +
            "egress: refused example.com:443: not in the sandbox's egress allow-list"
294 +
        );
295 +
        let a = ask("GET http://storage.googleapis.com/ HTTP/1.1\r\n\r\n");
296 +
        assert!(a.starts_with("HTTP/1.1 403"), "{a}");
297 +
        assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().contains("only https"));
298 +
        drop(proxy);
299 +
        assert!(!sock.exists(), "the socket goes with the job");
300 +
    }
301 +
}
modified crates/flow-core/src/canonical.rs +4 −0
…
46 unchanged lines
47 47
    if let Some(a) = &s.arch {
48 48
        fields.push((7, text(a)));
49 49
    }
50 +
    // The hosts egress may reach, sorted; absent without egress.
51 +
    if !s.egress_allow.is_empty() {
52 +
        fields.push((8, texts(&s.egress_allow)));
53 +
    }
50 54
    Value::map(fields)
51 55
}
52 56
…
96 unchanged lines
modified crates/flow-core/src/lib.rs +2 −2
…
11 unchanged lines
12 12
13 13
pub use canonical::{object_hash, scope_tree_hash, Hash};
14 14
pub use pipeline::{
15 -
    arch_ok, Identity, Job, Mount, MountSource, Network, Output, OutputKind, Pipeline, Problem, Problems, Runtime, Sandbox, Secret,
16 -
    PIPELINE_PATH,
15 +
    arch_ok, egress_allows, egress_pattern_ok, Identity, Job, Mount, MountSource, Network, Output, OutputKind, Pipeline, Problem, Problems,
16 +
    Runtime, Sandbox, Secret, EGRESS_LABEL, PIPELINE_PATH,
17 17
};
18 18
pub use receipt::{Environment, Receipt};
19 19
…
2 unchanged lines
modified crates/flow-core/src/pipeline.rs +93 −4
…
134 unchanged lines
135 135
    /// Labels a runner must have to run jobs here (`runs-on`), sorted.
136 136
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
137 137
    pub runs_on: Vec<String>,
138 +
    /// The hosts a `network "egress"` sandbox may reach (oxygenflow#55):
139 +
    /// names, or `*.` and a domain for any name under it; sorted. Its jobs
140 +
    /// reach them through the runner's proxy, and nothing else.
141 +
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
142 +
    pub egress_allow: Vec<String>,
138 143
    /// How long each job here may run, in seconds (`timeout "2h"`), when
139 144
    /// the job does not say; the hub caps it at the site's maximum.
140 145
    #[serde(default, skip_serializing_if = "Option::is_none")]
…
546 unchanged lines
687 692
    let mut runs_on = Vec::new();
688 693
    let mut timeout_s = None;
689 694
    let mut arch: Option<String> = None;
695 +
    let mut egress_allow = Vec::new();
690 696
    for c in block(ctx, node) {
691 697
        let key = c.name().value();
692 698
        if key != "runs-on" && !seen.insert(key.to_string()) {
…
7 unchanged lines
700 706
            ctx.no_children(c);
701 707
            parse_labels(ctx, c, &mut runs_on);
702 708
            continue;
709 +
        }
710 +
        if key != "network" {
711 +
            ctx.no_children(c);
703 712
        }
704 -
        ctx.no_children(c);
705 713
        match key {
706 714
            "runtime" => {
707 715
                runtime = match ctx.one_string(c).as_deref() {
…
61 unchanged lines
769 777
                }
770 778
            }
771 779
            "network" => {
772 -
                network = match ctx.one_string(c).as_deref() {
780 +
                let kind = ctx.one_string(c);
781 +
                if kind.as_deref() != Some("egress") {
782 +
                    ctx.no_children(c);
783 +
                }
784 +
                network = match kind.as_deref() {
773 785
                    Some("isolated") => Some(Network::Isolated),
774 786
                    // Loopback only: the guest's `lo` comes up, and it still has
775 787
                    // no device that reaches anything outside it.
776 788
                    Some("loopback") => Some(Network::Loopback),
789 +
                    // Allow-listed egress (oxygenflow#55): still no network
790 +
                    // device; a proxy in the guest relays to the runner, which
791 +
                    // lets through only the hosts named here.
777 792
                    Some("egress") => {
778 -
                        ctx.err(c.span(), "network \"egress\" is not in M0; M0 has \"isolated\" and \"loopback\"");
779 -
                        None
793 +
                        egress_allow = parse_egress(ctx, c);
794 +
                        Some(Network::Egress)
780 795
                    }
781 796
                    Some(other) => {
782 797
                        ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
…
9 unchanged lines
792 807
    if rootfs.is_none() && !seen.contains("rootfs") {
793 808
        ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
794 809
    }
810 +
    if network == Some(Network::Egress) {
811 +
        if runtime == Some(Runtime::Syscall) {
812 +
            ctx.err(node.span(), format!("sandbox {name:?}: network \"egress\" needs runtime \"microvm\""));
813 +
        }
814 +
        // Only a runner that allows egress offers this label.
815 +
        if !runs_on.iter().any(|l| l == EGRESS_LABEL) {
816 +
            runs_on.push(EGRESS_LABEL.to_string());
817 +
            runs_on.sort();
818 +
        }
819 +
    }
795 820
    if !named {
796 821
        return None;
797 822
    }
…
6 unchanged lines
804 829
        network: network?,
805 830
        env,
806 831
        runs_on,
832 +
        egress_allow,
807 833
        timeout_s,
808 834
        arch,
809 835
    })
836 +
}
837 +
838 +
/// The label a runner offers when it allows egress, and an egress sandbox
839 +
/// needs (oxygenflow#55).
840 +
pub const EGRESS_LABEL: &str = "egress";
841 +
842 +
/// The most hosts one sandbox may name.
843 +
const MAX_EGRESS_HOSTS: usize = 32;
844 +
845 +
/// Whether `pattern` is a host name, or `*.` and a domain: lower-case
846 +
/// letters, digits, `-` and dots, at least two labels.
847 +
pub fn egress_pattern_ok(pattern: &str) -> bool {
848 +
    let name = pattern.strip_prefix("*.").unwrap_or(pattern);
849 +
    let labels: Vec<&str> = name.split('.').collect();
850 +
    name.len() <= 253
851 +
        && labels.len() >= 2
852 +
        && labels.iter().all(|l| {
853 +
            !l.is_empty()
854 +
                && l.len() <= 63
855 +
                && !l.starts_with('-')
856 +
                && !l.ends_with('-')
857 +
                && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
858 +
        })
859 +
}
860 +
861 +
/// Whether `host` is one `patterns` allow: a name exactly, or a name under
862 +
/// a `*.` domain (not the domain itself). Case and a final dot are ignored.
863 +
pub fn egress_allows(patterns: &[String], host: &str) -> bool {
864 +
    let host = host.trim_end_matches('.').to_ascii_lowercase();
865 +
    patterns.iter().any(|p| match p.strip_prefix("*.") {
866 +
        Some(domain) => host.len() > domain.len() + 1 && host.ends_with(domain) && host[..host.len() - domain.len()].ends_with('.'),
867 +
        None => host == *p,
868 +
    })
869 +
}
870 +
871 +
/// `network "egress" { allow "*.googleapis.com" "registry.opentofu.org" }`.
872 +
fn parse_egress(ctx: &mut Ctx, node: &KdlNode) -> Vec<String> {
873 +
    let mut out = Vec::new();
874 +
    for c in block(ctx, node) {
875 +
        ctx.no_children(c);
876 +
        if c.name().value() != "allow" {
877 +
            ctx.err(c.span(), format!("unknown egress setting `{}`; egress takes `allow \"host\"...`", c.name().value()));
878 +
            continue;
879 +
        }
880 +
        for h in ctx.strings(c) {
881 +
            if egress_pattern_ok(&h) {
882 +
                out.push(h);
883 +
            } else {
884 +
                ctx.err(
885 +
                    c.span(),
886 +
                    format!("allow {h:?}: expected a lower-case host name such as \"registry.opentofu.org\", or \"*.\" and a domain"),
887 +
                );
888 +
            }
889 +
        }
890 +
    }
891 +
    out.sort();
892 +
    out.dedup();
893 +
    if out.is_empty() {
894 +
        ctx.err(node.span(), "network \"egress\" needs at least one `allow \"host\"`");
895 +
    } else if out.len() > MAX_EGRESS_HOSTS {
896 +
        ctx.err(node.span(), format!("network \"egress\" may allow at most {MAX_EGRESS_HOSTS} hosts"));
897 +
    }
898 +
    out
810 899
}
811 900
812 901
/// An `on` block: `push` and `tag` with globs (none: any), `manual`,
…
496 unchanged lines
modified crates/flow-core/src/tests.rs +43 −6
…
94 unchanged lines
95 95
96 96
#[test]
97 97
fn m0_refuses_what_it_cannot_honour() {
98 -
    let src = format!("schema-version 1\nsandbox \"s\" {{\n  runtime \"wasm\"\n  rootfs \"cas://blob/{ROOTFS}\"\n  network \"egress\"\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n");
98 +
    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");
99 99
    let ps = problems(&src);
100 100
    assert!(ps.iter().any(|(l, m)| *l == 3 && m.contains("wasm")), "{ps:?}");
101 101
    assert!(ps.iter().any(|(l, m)| *l == 5 && m.contains("isolated")), "{ps:?}");
…
173 unchanged lines
275 275
}
276 276
277 277
#[test]
278 -
fn loopback_is_in_m0_and_egress_is_not() {
278 +
fn loopback_and_allow_listed_egress() {
279 279
    let src = |net: &str| {
280 -
        format!("schema-version 1\nsandbox \"s\" {{\n  rootfs \"cas://blob/{ROOTFS}\"\n  network \"{net}\"\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n")
280 +
        format!("schema-version 1\nsandbox \"s\" {{\n  rootfs \"cas://blob/{ROOTFS}\"\n  network {net}\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n")
281 281
    };
282 -
    let p = Pipeline::parse(&src("loopback")).unwrap();
282 +
    let p = Pipeline::parse(&src("\"loopback\"")).unwrap();
283 283
    assert_eq!(p.sandboxes["s"].network, Network::Loopback);
284 -
    assert_ne!(p.hash(), Pipeline::parse(&src("isolated")).unwrap().hash(), "the network is part of the sandbox");
285 -
    assert!(problems(&src("egress")).iter().any(|(l, m)| *l == 4 && m.contains("not in M0")));
284 +
    assert_ne!(p.hash(), Pipeline::parse(&src("\"isolated\"")).unwrap().hash(), "the network is part of the sandbox");
285 +
286 +
    // Egress names its hosts (oxygenflow#55), sorted; the sandbox then needs
287 +
    // a runner offering the egress label.
288 +
    let p =
289 +
        Pipeline::parse(&src("\"egress\" { allow \"registry.opentofu.org\" \"*.googleapis.com\"; allow \"*.googleapis.com\"; }")).unwrap();
290 +
    let s = &p.sandboxes["s"];
291 +
    assert_eq!(s.network, Network::Egress);
292 +
    assert_eq!(s.egress_allow, ["*.googleapis.com", "registry.opentofu.org"]);
293 +
    assert_eq!(s.runs_on, [crate::EGRESS_LABEL]);
294 +
    let other = Pipeline::parse(&src("\"egress\" { allow \"registry.opentofu.org\"; }")).unwrap();
295 +
    assert_ne!(p.hash(), other.hash(), "the allow-list is part of the sandbox");
296 +
297 +
    let has = |net: &str, want: &str| {
298 +
        let ps = problems(&src(net));
299 +
        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{net}: {ps:?}");
300 +
    };
301 +
    has("\"egress\"", "needs a block");
302 +
    has("\"egress\" { }", "at least one `allow");
303 +
    has("\"egress\" { allow \"Example.COM\"; }", "lower-case host name");
304 +
    has("\"egress\" { allow \"localhost\"; }", "lower-case host name");
305 +
    has("\"egress\" { allow \"example.com:8443\"; }", "lower-case host name");
306 +
    has("\"egress\" { deny \"example.com\"; }", "unknown egress setting");
307 +
    has("\"loopback\" { allow \"example.com\"; }", "takes no block");
308 +
    let syscall = format!(
309 +
        "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"
310 +
    );
311 +
    assert!(problems(&syscall).iter().any(|(_, m)| m.contains("needs runtime \"microvm\"")));
312 +
}
313 +
314 +
#[test]
315 +
fn egress_patterns_match_names_and_names_under_a_domain() {
316 +
    let allow = vec!["*.googleapis.com".to_string(), "registry.opentofu.org".to_string()];
317 +
    for ok in ["storage.googleapis.com", "a.b.googleapis.com", "REGISTRY.opentofu.org.", "registry.opentofu.org"] {
318 +
        assert!(crate::egress_allows(&allow, ok), "{ok}");
319 +
    }
320 +
    for no in ["googleapis.com", "evilgoogleapis.com", "googleapis.com.evil.org", "opentofu.org", "x.registry.opentofu.org", ""] {
321 +
        assert!(!crate::egress_allows(&allow, no), "{no}");
322 +
    }
286 323
}
287 324
288 325
#[test]
…
316 unchanged lines
modified crates/flow-init/src/main.rs +9 −1
…
14 unchanged lines
15 15
use std::os::fd::FromRawFd;
16 16
use std::sync::{Arc, Mutex};
17 17
18 +
mod egress;
18 19
mod job;
19 20
use job::{dev_links, job_env, loopback_up, resolve_program, PATH};
20 21
…
6 unchanged lines
27 28
    /// `network "loopback"`: bring up `lo` (the guest has no other device).
28 29
    #[serde(default)]
29 30
    loopback: bool,
31 +
    /// `network "egress"` (oxygenflow#55): `lo` comes up and a proxy on
32 +
    /// 127.0.0.1:3128 relays over vsock to the runner.
33 +
    #[serde(default)]
34 +
    egress: bool,
30 35
    /// Secrets to write before the job starts (spec §9): each at `path`
31 36
    /// under `/run/secrets`, 0400 and the job's user's, on a tmpfs.
32 37
    #[serde(default)]
…
152 unchanged lines
185 190
                .map_err(|e| format!("chmod {}: {e}", s.path))?;
186 191
        }
187 192
    }
188 -
    if job.loopback {
193 +
    if job.loopback || job.egress {
189 194
        loopback_up()?;
195 +
    }
196 +
    if job.egress {
197 +
        egress::start()?;
190 198
    }
191 199
    let root = cstr("/newroot")?;
192 200
    if unsafe { libc::chroot(root.as_ptr()) } != 0 || std::env::set_current_dir("/src").is_err() {
…
89 unchanged lines
modified crates/flow/src/home.rs +17 −0
…
63 unchanged lines
64 64
    /// The build cache's size limit in GiB (#28; default 64).
65 65
    #[serde(default)]
66 66
    pub cache_gb: Option<u64>,
67 +
    /// Let microVM jobs reach the hosts their sandbox allows, through this
68 +
    /// runner (oxygenflow#55); off unless set. The runner then offers the
69 +
    /// `egress` label, which such jobs need.
70 +
    #[serde(default)]
71 +
    pub egress: bool,
72 +
}
73 +
74 +
impl RunnerConfig {
75 +
    /// The labels offered to the hub: the configured ones, and `egress`
76 +
    /// when egress is on.
77 +
    pub fn offered_labels(&self) -> Vec<String> {
78 +
        let mut l = self.labels.clone();
79 +
        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {
80 +
            l.push(flow_core::EGRESS_LABEL.to_string());
81 +
        }
82 +
        l
83 +
    }
67 84
}
68 85
69 86
/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.
…
184 unchanged lines
modified crates/flow/src/main.rs +1 −0
…
2 unchanged lines
3 3
4 4
mod cache;
5 5
mod cpio;
6 +
mod egress;
6 7
mod facts;
7 8
/// The job-starting code `flow-init` uses in a microVM, shared with the
8 9
/// syscall tier.
…
559 unchanged lines
modified crates/flow/src/microvm.rs +60 −0
…
84 unchanged lines
85 85
86 86
    let mut spec = guest_spec(job);
87 87
    spec["loopback"] = serde_json::Value::Bool(sandbox.network == flow_core::Network::Loopback);
88 +
    // Allow-listed egress (oxygenflow#55): a vsock device and the runner's
89 +
    // proxy behind it; the guest relays 127.0.0.1:3128 to it.
90 +
    let egress = sandbox.network == flow_core::Network::Egress;
91 +
    let (egress_tx, egress_rx) = mpsc::channel::<String>();
92 +
    let mut _proxy = None;
93 +
    if egress {
94 +
        if !home.config.runner.as_ref().is_some_and(|r| r.egress) {
95 +
            bail!("this runner does not allow network egress (a runner allows it with `egress = true` under [runner] in flow.toml)");
96 +
        }
97 +
        _proxy = Some(
98 +
            crate::egress::start(&scratch.join(format!("v.sock_{}", crate::egress::VSOCK_PORT)), sandbox.egress_allow.clone(), egress_tx)
99 +
                .context("start the egress proxy")?,
100 +
        );
101 +
        spec["egress"] = serde_json::Value::Bool(true);
102 +
        add_proxy_env(&mut spec);
103 +
        sink.record(Record {
104 +
            t: 0,
105 +
            stream: "flow".into(),
106 +
            text: format!("egress: through this runner, to {} only", sandbox.egress_allow.join(", ")),
107 +
        });
108 +
    }
88 109
    // Secrets ride in the guest's job file, which only flow-init reads,
89 110
    // outside the root the job is confined to; it writes each as a file.
90 111
    if !secrets.is_empty() {
…
37 unchanged lines
128 149
        "drives": drives,
129 150
        "machine-config": { "vcpu_count": sandbox.cpus, "mem_size_mib": sandbox.memory_mib },
130 151
    });
152 +
    let mut vm = vm;
153 +
    if egress {
154 +
        // Relative to the VMM's working directory (the scratch directory),
155 +
        // which keeps the socket path short.
156 +
        vm["vsock"] = serde_json::json!({ "guest_cid": 3, "uds_path": "v.sock" });
157 +
    }
131 158
    let vm_path = scratch.join("vm.json");
132 159
    std::fs::write(&vm_path, serde_json::to_vec_pretty(&vm)?)?;
133 160
…
25 unchanged lines
159 186
    loop {
160 187
        if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
161 188
            peak_memory_mib = Some(m);
189 +
        }
190 +
        while let Ok(line) = egress_rx.try_recv() {
191 +
            sink.record(Record { t: t(&started), stream: "flow".into(), text: line });
162 192
        }
163 193
        match rx.recv_timeout(Duration::from_millis(200)) {
164 194
            Ok((from_console, line)) => {
…
54 unchanged lines
219 249
    })
220 250
}
221 251
252 +
/// The proxy variables a job with egress gets, in both spellings tools
253 +
/// read, unless the job sets them itself.
254 +
fn add_proxy_env(spec: &mut serde_json::Value) {
255 +
    let Some(env) = spec["env"].as_array_mut() else { return };
256 +
    let set: Vec<String> = env.iter().filter_map(|kv| kv[0].as_str().map(String::from)).collect();
257 +
    for (k, v) in [
258 +
        ("HTTPS_PROXY", crate::egress::GUEST_PROXY),
259 +
        ("https_proxy", crate::egress::GUEST_PROXY),
260 +
        ("HTTP_PROXY", crate::egress::GUEST_PROXY),
261 +
        ("http_proxy", crate::egress::GUEST_PROXY),
262 +
        ("NO_PROXY", "localhost,127.0.0.1"),
263 +
        ("no_proxy", "localhost,127.0.0.1"),
264 +
    ] {
265 +
        if !set.iter().any(|s| s == k) {
266 +
            env.push(serde_json::json!([k, v]));
267 +
        }
268 +
    }
269 +
}
270 +
222 271
/// What `flow-init` reads from `/job.json`: argv, env, and the mounts
223 272
/// with the device each disk mount is attached as (vdc onwards).
224 273
pub fn guest_spec(job: &Job) -> serde_json::Value {
…
36 unchanged lines
261 310
mod tests {
262 311
    use super::*;
263 312
    use flow_core::Mount;
313 +
314 +
    #[test]
315 +
    fn a_job_with_egress_gets_the_proxy_variables_unless_it_sets_them() {
316 +
        let mut spec = serde_json::json!({ "env": [["HTTPS_PROXY", "http://mine:1"]] });
317 +
        super::add_proxy_env(&mut spec);
318 +
        let env = spec["env"].as_array().unwrap();
319 +
        let get = |k: &str| env.iter().find(|kv| kv[0] == k).map(|kv| kv[1].as_str().unwrap().to_string());
320 +
        assert_eq!(get("HTTPS_PROXY").as_deref(), Some("http://mine:1"), "the job's own is kept");
321 +
        assert_eq!(get("https_proxy").as_deref(), Some("http://127.0.0.1:3128"));
322 +
        assert_eq!(get("NO_PROXY").as_deref(), Some("localhost,127.0.0.1"));
323 +
    }
264 324
265 325
    #[test]
266 326
    fn guest_spec_numbers_disks_after_the_workspace() {
…
17 unchanged lines
modified crates/flow/src/runner.rs +1 −1
…
519 unchanged lines
520 520
                free_cpus,
521 521
                info: Some(RunnerInfo {
522 522
                    version: env!("OXY_VERSION_LONG").into(),
523 -
                    ..RunnerInfo::here(cfg.labels.clone(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)
523 +
                    ..RunnerInfo::here(cfg.offered_labels(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)
524 524
                }),
525 525
                // Read each time: an image imported while the runner runs counts.
526 526
                images: Some(home.images()),
…
712 unchanged lines
modified docs/m0.md +1 −1
…
171 unchanged lines
172 172
173 173
**How it runs.**
174 174
- **Helper:** for each job the runner starts `flow __sandbox <spec>` (its own binary, through `/proc/self/exe`). The helper joins the job's cgroup, makes the namespaces with the job's user (uid and gid 1000) mapped to the runner's, and forks the sandbox's PID 1.
175 -
- **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"`, pivots into the root and detaches the old one, so the host's files, the spec and the scratch directory are out of reach.
175 +
- **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.
176 176
- **Job:** the job gets no new privileges, then Landlock (read and execute everywhere, write beneath `/src`, `/tmp`, `/dev` and its mounts), then the seccomp filter, then `execve` with exactly the job's environment, found the way `flow-init` finds it (the two share that code). The filter also refuses io_uring (which could open sockets past it), the new mount calls, kexec, swap and reboot.
177 177
- **Unpacking:** `debugfs rdump` unpacks the image into `trees/<hash>` in the home once, without mount rights. Set-id bits and device nodes are dropped.
178 178
- **Ephemeral mounts:** a `tmpfs` mount is a tmpfs of its size. An `ephemeral://disk` mount is a scratch directory, so its size is not enforced in this tier.
…
75 unchanged lines
modified docs/spec.md +30 −1
…
90 unchanged lines
91 91
    rootfs "cas://blob/9b5c2a1e8f3d4e7a6b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a"
92 92
    memory-limit "4GiB"
93 93
    cpu-limit 4
94 -
    network "isolated"      // Options: "isolated" | "loopback" | "egress"
94 +
    network "isolated"      // Options: "isolated" | "loopback" | "egress" { allow "host"... } (§4.3.1)
95 95
}
96 96
97 97
// Job evaluated based on Oxygen change closure
…
112 unchanged lines
210 210
  directly to FastCDC chunks stored in Oxygen's CAS.
211 211
- Guest VM memory is cleared upon termination.
212 212
213 +
#### 4.3.1 Allow-listed egress (oxygenflow#55)
214 +
215 +
A sandbox reaches the network only by naming its hosts:
216 +
217 +
```kdl
218 +
sandbox "deploy" {
219 +
    runtime "microvm"
220 +
    network "egress" { allow "*.googleapis.com" "registry.opentofu.org" }
221 +
}
222 +
```
223 +
224 +
The VM still has no network device. `flow-init` serves a proxy on the
225 +
guest's `127.0.0.1:3128` (the job gets `HTTPS_PROXY`, `HTTP_PROXY` and
226 +
their lower-case forms, unless it sets them) and relays each connection,
227 +
unread, over Firecracker's vsock to the runner. The runner accepts only
228 +
`CONNECT host:443` for a host the list allows (a name exactly, or any
229 +
name under a `*.` domain, never the domain itself; never an address),
230 +
resolves it itself (the guest does not resolve), refuses any name that
231 +
resolves to an address that is not public, and relays the bytes without
232 +
opening the TLS. Every decision, allowed or refused, is a `flow` line in
233 +
the job's log. Patterns are lower-case host names of two or more labels,
234 +
at most 32 per sandbox, and part of the sandbox's hash.
235 +
236 +
Egress is off unless a runner turns it on (`egress = true` under
237 +
`[runner]` in `flow.toml`); such a runner offers the `egress` label, and
238 +
an egress sandbox needs it, so the hub hands its jobs only to runners
239 +
that allow it, and a runner without it refuses one anyway. The syscall
240 +
runtime has no egress yet.
241 +
213 242
## 5. Content-Addressable Storage (CAS) & Zero-Copy Mounting
214 243
215 244
OxygenFlow runners bypass traditional clone operations. Repositories are accessed
…
379 unchanged lines