OXYGENHUB

Coverage / crates/flow/src/egress.rs 79.2%

171 of 216 lines run, 15 of 33 functions · run #185 on main · run missed

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.
385×
pub fn parse_head(head: &str) -> Ask {
395×
    let line = head.lines().next().unwrap_or("");
405×
    let mut parts = line.split_whitespace();
415×
    let (method, target, version) = (parts.next().unwrap_or(""), parts.next().unwrap_or(""), parts.next().unwrap_or(""));
425×
    if method != "CONNECT" {
432×
        return Ask::Refused(format!("only https is relayed (CONNECT), not {method:?}"));
443×
    }
453×
    if !version.starts_with("HTTP/1.") {
460×
        return Ask::Refused("not an HTTP/1 request".into());
473×
    }
483×
    let Some((host, port)) = target.rsplit_once(':') else { return Ask::Refused(format!("CONNECT {target:?} names no port")) };
492×
    let Ok(port) = port.parse::<u16>() else { return Ask::Refused(format!("CONNECT {target:?} has no valid port")) };
502×
    let host = host.trim_start_matches('[').trim_end_matches(']').trim_end_matches('.').to_ascii_lowercase();
512×
    if host.is_empty() {
520×
        return Ask::Refused("CONNECT names no host".into());
532×
    }
542×
    Ask::Connect { host, port }
555×
}
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.
596×
pub fn decide(allow: &[String], host: &str, port: u16) -> Result<(), String> {
606×
    if port != PORT {
611×
        return Err(format!("only port {PORT} is relayed"));
625×
    }
635×
    if host.parse::<IpAddr>().is_ok() {
641×
        return Err("an address, not a name: the allow-list names hosts".into());
654×
    }
664×
    if !flow_core::egress_allows(allow, host) {
672×
        return Err("not in the sandbox's egress allow-list".into());
682×
    }
692×
    Ok(())
706×
}
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.
7514×
pub fn public(ip: IpAddr) -> bool {
7614×
    match ip {
779×
        IpAddr::V4(v4) => {
789×
            let o = v4.octets();
799×
            !(v4.is_loopback()
808×
                || v4.is_private()
814×
                || v4.is_link_local()
823×
                || v4.is_multicast()
833×
                || v4.is_broadcast()
843×
                || v4.is_unspecified()
852×
                || v4.is_documentation()
862×
                || o[0] == 0
872×
                || (o[0] == 100 && (o[1] & 0xc0) == 64)
881×
                || (o[0] == 192 && o[1] == 0 && o[2] == 0)
891×
                || o[0] >= 240)
90
        }
915×
        IpAddr::V6(v6) => {
925×
            if let Some(v4) = v6.to_ipv4_mapped() {
931×
                return public(IpAddr::V4(v4));
944×
            }
954×
            let s = v6.segments();
964×
            !(v6.is_loopback()
973×
                || v6.is_unspecified()
983×
                || v6.is_multicast()
993×
                || (s[0] & 0xfe00) == 0xfc00
1002×
                || (s[0] & 0xffc0) == 0xfe80
1011×
                || (s[0] == 0x64 && s[1] == 0xff9b)
1021×
                || (s[0] == 0x2001 && s[1] == 0x0db8))
103
        }
104
    }
10514×
}
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 {
1141×
    fn drop(&mut self) {
1151×
        self.stop.store(true, Ordering::SeqCst);
116
        // Wake the accept loop so it sees the stop.
1171×
        let _ = UnixStream::connect(&self.path);
1181×
        let _ = std::fs::remove_file(&self.path);
1191×
    }
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.
1241×
pub fn start(socket: &Path, allow: Vec<String>, log: mpsc::Sender<String>) -> std::io::Result<Proxy> {
1251×
    let _ = std::fs::remove_file(socket);
1261×
    let listener = UnixListener::bind(socket)?;
1271×
    let stop = Arc::new(AtomicBool::new(false));
1281×
    let open = Arc::new(AtomicUsize::new(0));
1291×
    let allow = Arc::new(allow);
1301×
    let s = stop.clone();
1311×
    std::thread::spawn(move || {
1323×
        for conn in listener.incoming() {
1333×
            if s.load(Ordering::SeqCst) {
1341×
                break;
1352×
            }
1362×
            let Ok(conn) = conn else { continue };
1372×
            if open.load(Ordering::SeqCst) >= MAX_OPEN {
1380×
                let _ = refuse(conn, "too many connections at once");
1390×
                let _ = log.send(format!("egress: refused a connection: {MAX_OPEN} already open"));
1400×
                continue;
1412×
            }
1422×
            open.fetch_add(1, Ordering::SeqCst);
1432×
            let (allow, log, open) = (allow.clone(), log.clone(), open.clone());
1442×
            std::thread::spawn(move || {
1452×
                serve(conn, &allow, &log);
1462×
                open.fetch_sub(1, Ordering::SeqCst);
1472×
            });
148
        }
1491×
    });
1501×
    Ok(Proxy { stop, path: socket.to_path_buf() })
1511×
}
152
1532×
fn refuse(mut conn: UnixStream, why: &str) -> std::io::Result<()> {
1542×
    let body = format!("{why}\n");
1552×
    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())
1562×
}
157
158
/// Read up to the blank line that ends a request head.
1592×
fn read_head(conn: &mut UnixStream) -> std::io::Result<String> {
1602×
    let mut head = Vec::new();
1612×
    let mut byte = [0u8; 1];
16285×
    while !head.ends_with(b"\r\n\r\n") {
16383×
        if head.len() >= MAX_HEAD || conn.read(&mut byte)? == 0 {
1640×
            break;
16583×
        }
16683×
        head.push(byte[0]);
167
    }
1682×
    Ok(String::from_utf8_lossy(&head).into_owned())
1692×
}
170
1712×
fn serve(mut conn: UnixStream, allow: &[String], log: &mpsc::Sender<String>) {
1722×
    let _ = conn.set_read_timeout(Some(Duration::from_secs(30)));
1732×
    let head = match read_head(&mut conn) {
1742×
        Ok(h) => h,
1750×
        Err(_) => return,
176
    };
1772×
    let (host, port) = match parse_head(&head) {
1781×
        Ask::Connect { host, port } => (host, port),
1791×
        Ask::Refused(why) => {
1801×
            let _ = log.send(format!("egress: refused: {why}"));
1811×
            let _ = refuse(conn, &why);
1821×
            return;
183
        }
184
    };
1851×
    if let Err(why) = decide(allow, &host, port) {
1861×
        let _ = log.send(format!("egress: refused {host}:{port}: {why}"));
1871×
        let _ = refuse(conn, &format!("{host}:{port}: {why}"));
1881×
        return;
1890×
    }
1900×
    let addrs: Vec<SocketAddr> = match (host.as_str(), port).to_socket_addrs() {
1910×
        Ok(a) => a.collect(),
1920×
        Err(e) => {
1930×
            let _ = log.send(format!("egress: refused {host}:{port}: it could not be resolved ({e})"));
1940×
            let _ = refuse(conn, &format!("{host} could not be resolved"));
1950×
            return;
196
        }
197
    };
1980×
    if let Some(bad) = addrs.iter().find(|a| !public(a.ip())) {
1990×
        let why = format!("it resolves to {}, which is not a public address", bad.ip());
2000×
        let _ = log.send(format!("egress: refused {host}:{port}: {why}"));
2010×
        let _ = refuse(conn, &format!("{host}: {why}"));
2020×
        return;
2030×
    }
2040×
    let out = addrs.iter().find_map(|a| TcpStream::connect_timeout(a, CONNECT_TIMEOUT).ok());
2050×
    let Some(out) = out else {
2060×
        let _ = log.send(format!("egress: {host}:{port} is allowed but did not answer"));
2070×
        let _ = conn.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n");
2080×
        return;
209
    };
2100×
    let _ = log.send(format!("egress: allowed {host}:{port}"));
2110×
    if conn.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n").is_err() {
2120×
        return;
2130×
    }
2140×
    let _ = conn.set_read_timeout(None);
2150×
    relay(conn, out);
2162×
}
217
218
/// Copy both ways until either side closes.
2190×
fn relay(guest: UnixStream, out: TcpStream) {
2200×
    let (Ok(mut guest_r), Ok(mut out_w)) = (guest.try_clone(), out.try_clone()) else { return };
2210×
    let up = std::thread::spawn(move || {
2220×
        let _ = std::io::copy(&mut guest_r, &mut out_w);
2230×
        let _ = out_w.shutdown(std::net::Shutdown::Write);
2240×
    });
2250×
    let (mut out_r, mut guest_w) = (out, guest);
2260×
    let _ = std::io::copy(&mut out_r, &mut guest_w);
2270×
    let _ = guest_w.shutdown(std::net::Shutdown::Write);
2280×
    let _ = up.join();
2290×
}
230
231
#[cfg(test)]
232
mod tests {
233
    use super::*;
234
2356×
    fn allow() -> Vec<String> {
2366×
        vec!["*.googleapis.com".into(), "registry.opentofu.org".into()]
2376×
    }
238
239
    #[test]
2401×
    fn only_connect_to_an_allowed_name_on_443_is_let_through() {
2411×
        assert_eq!(
2421×
            parse_head("CONNECT storage.googleapis.com:443 HTTP/1.1\r\nHost: x\r\n\r\n"),
2431×
            Ask::Connect { host: "storage.googleapis.com".into(), port: 443 }
244
        );
2451×
        assert!(matches!(parse_head("GET http://example.com/ HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("only https")));
2461×
        assert!(matches!(parse_head("CONNECT example.com HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("no port")));
2471×
        assert_eq!(decide(&allow(), "storage.googleapis.com", 443), Ok(()));
2481×
        assert_eq!(decide(&allow(), "registry.opentofu.org", 443), Ok(()));
2491×
        assert!(decide(&allow(), "example.com", 443).unwrap_err().contains("allow-list"));
2501×
        assert!(decide(&allow(), "registry.opentofu.org", 22).unwrap_err().contains("only port 443"));
2511×
        assert!(decide(&allow(), "142.250.1.1", 443).unwrap_err().contains("names hosts"));
2521×
    }
253
254
    #[test]
2551×
    fn private_addresses_are_not_public() {
25611×
        for ip in [
2571×
            "127.0.0.1",
2581×
            "10.1.2.3",
2591×
            "192.168.1.1",
2601×
            "172.16.0.1",
2611×
            "169.254.169.254",
2621×
            "100.64.0.1",
2631×
            "0.0.0.0",
2641×
            "::1",
2651×
            "fd00::1",
2661×
            "fe80::1",
2671×
            "::ffff:10.0.0.1",
2681×
        ] {
26911×
            assert!(!public(ip.parse().unwrap()), "{ip}");
270
        }
2712×
        for ip in ["142.250.1.1", "2a00:1450:4009::200e"] {
2722×
            assert!(public(ip.parse().unwrap()), "{ip}");
273
        }
2741×
    }
275
276
    #[test]
2771×
    fn the_proxy_refuses_and_logs_a_host_not_allowed() {
2781×
        let dir = tempfile::tempdir().unwrap();
2791×
        let sock = dir.path().join("v.sock_3128");
2801×
        let (tx, rx) = mpsc::channel();
2811×
        let proxy = start(&sock, allow(), tx).unwrap();
2822×
        let ask = |head: &str| {
2832×
            let mut c = UnixStream::connect(&sock).unwrap();
2842×
            c.write_all(head.as_bytes()).unwrap();
2852×
            let mut answer = String::new();
2862×
            c.read_to_string(&mut answer).unwrap();
2872×
            answer
2882×
        };
2891×
        let a = ask("CONNECT example.com:443 HTTP/1.1\r\n\r\n");
2901×
        assert!(a.starts_with("HTTP/1.1 403") && a.contains("not in the sandbox's egress allow-list"), "{a}");
2911×
        assert_eq!(
2921×
            rx.recv_timeout(Duration::from_secs(5)).unwrap(),
293
            "egress: refused example.com:443: not in the sandbox's egress allow-list"
294
        );
2951×
        let a = ask("GET http://storage.googleapis.com/ HTTP/1.1\r\n\r\n");
2961×
        assert!(a.starts_with("HTTP/1.1 403"), "{a}");
2971×
        assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().contains("only https"));
2981×
        drop(proxy);
2991×
        assert!(!sock.exists(), "the socket goes with the job");
3001×
    }
301
}