egress.rs 79.2%

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

//! 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).

#[cfg(unix)]
use std::io::{Read, Write};
use std::net::IpAddr;
#[cfg(unix)]
use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
#[cfg(unix)]
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::Path;
#[cfg(unix)]
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc;
#[cfg(unix)]
use std::sync::Arc;
#[cfg(unix)]
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;
#[cfg(unix)]
/// Connections at once, per job.
const MAX_OPEN: usize = 64;
#[cfg(unix)]
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
#[cfg(unix)]
/// 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))
        }
    }
}

#[cfg(unix)]
/// The proxy for one job: listening until dropped.
pub struct Proxy {
    stop: Arc<AtomicBool>,
    path: std::path::PathBuf,
}

#[cfg(unix)]
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);
    }
}

#[cfg(unix)]
/// 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() })
}

#[cfg(unix)]
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())
}

#[cfg(unix)]
/// 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())
}

#[cfg(unix)]
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);
}

#[cfg(unix)]
/// 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();
}

/// The egress proxy serves a microVM's vsock, which only Linux has.
#[cfg(not(unix))]
pub struct Proxy;

#[cfg(not(unix))]
pub fn start(_socket: &Path, _allow: Vec<String>, _log: mpsc::Sender<String>) -> std::io::Result<Proxy> {
    Err(std::io::Error::new(std::io::ErrorKind::Unsupported, "egress serves microVMs, which run only on Linux"))
}

#[cfg(all(test, unix))]
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");
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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		#[cfg(unix)]
11		use std::io::{Read, Write};
12		use std::net::IpAddr;
13		#[cfg(unix)]
14		use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
15		#[cfg(unix)]
16		use std::os::unix::net::{UnixListener, UnixStream};
17		use std::path::Path;
18		#[cfg(unix)]
19		use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
20		use std::sync::mpsc;
21		#[cfg(unix)]
22		use std::sync::Arc;
23		#[cfg(unix)]
24		use std::time::Duration;
25
26		/// The vsock port the guest's relay connects to on the host (CID 2).
27		pub const VSOCK_PORT: u32 = 3128;
28		/// Where the job's tools find the proxy inside the guest.
29		pub const GUEST_PROXY: &str = "http://127.0.0.1:3128";
30		/// The only port relayed: TLS.
31		const PORT: u16 = 443;
32		#[cfg(unix)]
33		/// Connections at once, per job.
34		const MAX_OPEN: usize = 64;
35		#[cfg(unix)]
36		const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
37		#[cfg(unix)]
38		/// The longest request head read.
39		const MAX_HEAD: usize = 8 * 1024;
40
41		/// What a CONNECT asked for, or why it was refused.
42		#[derive(Debug, PartialEq, Eq)]
43		pub enum Ask {
44		Connect { host: String, port: u16 },
45		Refused(String),
46		}
47
48		/// Read a request head: `CONNECT host:port HTTP/1.x`, and only that.
49	5×	pub fn parse_head(head: &str) -> Ask {
50	5×	let line = head.lines().next().unwrap_or("");
51	5×	let mut parts = line.split_whitespace();
52	5×	let (method, target, version) = (parts.next().unwrap_or(""), parts.next().unwrap_or(""), parts.next().unwrap_or(""));
53	5×	if method != "CONNECT" {
54	2×	return Ask::Refused(format!("only https is relayed (CONNECT), not {method:?}"));
55	3×	}
56	3×	if !version.starts_with("HTTP/1.") {
57	0×	return Ask::Refused("not an HTTP/1 request".into());
58	3×	}
59	3×	let Some((host, port)) = target.rsplit_once(':') else { return Ask::Refused(format!("CONNECT {target:?} names no port")) };
60	2×	let Ok(port) = port.parse::<u16>() else { return Ask::Refused(format!("CONNECT {target:?} has no valid port")) };
61	2×	let host = host.trim_start_matches('[').trim_end_matches(']').trim_end_matches('.').to_ascii_lowercase();
62	2×	if host.is_empty() {
63	0×	return Ask::Refused("CONNECT names no host".into());
64	2×	}
65	2×	Ask::Connect { host, port }
66	5×	}
67
68		/// Whether the job may go to `host:port`: the port is 443, the host is a
69		/// name (an address is refused: the allow-list names hosts) the list allows.
70	6×	pub fn decide(allow: &[String], host: &str, port: u16) -> Result<(), String> {
71	6×	if port != PORT {
72	1×	return Err(format!("only port {PORT} is relayed"));
73	5×	}
74	5×	if host.parse::<IpAddr>().is_ok() {
75	1×	return Err("an address, not a name: the allow-list names hosts".into());
76	4×	}
77	4×	if !flow_core::egress_allows(allow, host) {
78	2×	return Err("not in the sandbox's egress allow-list".into());
79	2×	}
80	2×	Ok(())
81	6×	}
82
83		/// Whether an address is on the public internet: not loopback, private,
84		/// shared (CGNAT), link-local (cloud metadata among them), multicast,
85		/// unspecified or documentation, nor an IPv6 local range or a mapped form.
86	14×	pub fn public(ip: IpAddr) -> bool {
87	14×	match ip {
88	9×	IpAddr::V4(v4) => {
89	9×	let o = v4.octets();
90	9×	!(v4.is_loopback()
91	8×	\|\| v4.is_private()
92	4×	\|\| v4.is_link_local()
93	3×	\|\| v4.is_multicast()
94	3×	\|\| v4.is_broadcast()
95	3×	\|\| v4.is_unspecified()
96	2×	\|\| v4.is_documentation()
97	2×	\|\| o[0] == 0
98	2×	\|\| (o[0] == 100 && (o[1] & 0xc0) == 64)
99	1×	\|\| (o[0] == 192 && o[1] == 0 && o[2] == 0)
100	1×	\|\| o[0] >= 240)
101		}
102	5×	IpAddr::V6(v6) => {
103	5×	if let Some(v4) = v6.to_ipv4_mapped() {
104	1×	return public(IpAddr::V4(v4));
105	4×	}
106	4×	let s = v6.segments();
107	4×	!(v6.is_loopback()
108	3×	\|\| v6.is_unspecified()
109	3×	\|\| v6.is_multicast()
110	3×	\|\| (s[0] & 0xfe00) == 0xfc00
111	2×	\|\| (s[0] & 0xffc0) == 0xfe80
112	1×	\|\| (s[0] == 0x64 && s[1] == 0xff9b)
113	1×	\|\| (s[0] == 0x2001 && s[1] == 0x0db8))
114		}
115		}
116	14×	}
117
118		#[cfg(unix)]
119		/// The proxy for one job: listening until dropped.
120		pub struct Proxy {
121		stop: Arc<AtomicBool>,
122		path: std::path::PathBuf,
123		}
124
125		#[cfg(unix)]
126		impl Drop for Proxy {
127	1×	fn drop(&mut self) {
128	1×	self.stop.store(true, Ordering::SeqCst);
129		// Wake the accept loop so it sees the stop.
130	1×	let _ = UnixStream::connect(&self.path);
131	1×	let _ = std::fs::remove_file(&self.path);
132	1×	}
133		}
134
135		#[cfg(unix)]
136		/// Listen on `socket` (Firecracker's `uds_path` and `_3128`) for the
137		/// guest's connections; each decision is sent to `log` as a line.
138	1×	pub fn start(socket: &Path, allow: Vec<String>, log: mpsc::Sender<String>) -> std::io::Result<Proxy> {
139	1×	let _ = std::fs::remove_file(socket);
140	1×	let listener = UnixListener::bind(socket)?;
141	1×	let stop = Arc::new(AtomicBool::new(false));
142	1×	let open = Arc::new(AtomicUsize::new(0));
143	1×	let allow = Arc::new(allow);
144	1×	let s = stop.clone();
145	1×	std::thread::spawn(move \|\| {
146	3×	for conn in listener.incoming() {
147	3×	if s.load(Ordering::SeqCst) {
148	1×	break;
149	2×	}
150	2×	let Ok(conn) = conn else { continue };
151	2×	if open.load(Ordering::SeqCst) >= MAX_OPEN {
152	0×	let _ = refuse(conn, "too many connections at once");
153	0×	let _ = log.send(format!("egress: refused a connection: {MAX_OPEN} already open"));
154	0×	continue;
155	2×	}
156	2×	open.fetch_add(1, Ordering::SeqCst);
157	2×	let (allow, log, open) = (allow.clone(), log.clone(), open.clone());
158	2×	std::thread::spawn(move \|\| {
159	2×	serve(conn, &allow, &log);
160	2×	open.fetch_sub(1, Ordering::SeqCst);
161	2×	});
162		}
163	1×	});
164	1×	Ok(Proxy { stop, path: socket.to_path_buf() })
165	1×	}
166
167		#[cfg(unix)]
168	2×	fn refuse(mut conn: UnixStream, why: &str) -> std::io::Result<()> {
169	2×	let body = format!("{why}\n");
170	2×	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())
171	2×	}
172
173		#[cfg(unix)]
174		/// Read up to the blank line that ends a request head.
175	2×	fn read_head(conn: &mut UnixStream) -> std::io::Result<String> {
176	2×	let mut head = Vec::new();
177	2×	let mut byte = [0u8; 1];
178	85×	while !head.ends_with(b"\r\n\r\n") {
179	83×	if head.len() >= MAX_HEAD \|\| conn.read(&mut byte)? == 0 {
180	0×	break;
181	83×	}
182	83×	head.push(byte[0]);
183		}
184	2×	Ok(String::from_utf8_lossy(&head).into_owned())
185	2×	}
186
187		#[cfg(unix)]
188	2×	fn serve(mut conn: UnixStream, allow: &[String], log: &mpsc::Sender<String>) {
189	2×	let _ = conn.set_read_timeout(Some(Duration::from_secs(30)));
190	2×	let head = match read_head(&mut conn) {
191	2×	Ok(h) => h,
192	0×	Err(_) => return,
193		};
194	2×	let (host, port) = match parse_head(&head) {
195	1×	Ask::Connect { host, port } => (host, port),
196	1×	Ask::Refused(why) => {
197	1×	let _ = log.send(format!("egress: refused: {why}"));
198	1×	let _ = refuse(conn, &why);
199	1×	return;
200		}
201		};
202	1×	if let Err(why) = decide(allow, &host, port) {
203	1×	let _ = log.send(format!("egress: refused {host}:{port}: {why}"));
204	1×	let _ = refuse(conn, &format!("{host}:{port}: {why}"));
205	1×	return;
206	0×	}
207	0×	let addrs: Vec<SocketAddr> = match (host.as_str(), port).to_socket_addrs() {
208	0×	Ok(a) => a.collect(),
209	0×	Err(e) => {
210	0×	let _ = log.send(format!("egress: refused {host}:{port}: it could not be resolved ({e})"));
211	0×	let _ = refuse(conn, &format!("{host} could not be resolved"));
212	0×	return;
213		}
214		};
215	0×	if let Some(bad) = addrs.iter().find(\|a\| !public(a.ip())) {
216	0×	let why = format!("it resolves to {}, which is not a public address", bad.ip());
217	0×	let _ = log.send(format!("egress: refused {host}:{port}: {why}"));
218	0×	let _ = refuse(conn, &format!("{host}: {why}"));
219	0×	return;
220	0×	}
221	0×	let out = addrs.iter().find_map(\|a\| TcpStream::connect_timeout(a, CONNECT_TIMEOUT).ok());
222	0×	let Some(out) = out else {
223	0×	let _ = log.send(format!("egress: {host}:{port} is allowed but did not answer"));
224	0×	let _ = conn.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n");
225	0×	return;
226		};
227	0×	let _ = log.send(format!("egress: allowed {host}:{port}"));
228	0×	if conn.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n").is_err() {
229	0×	return;
230	0×	}
231	0×	let _ = conn.set_read_timeout(None);
232	0×	relay(conn, out);
233	2×	}
234
235		#[cfg(unix)]
236		/// Copy both ways until either side closes.
237	0×	fn relay(guest: UnixStream, out: TcpStream) {
238	0×	let (Ok(mut guest_r), Ok(mut out_w)) = (guest.try_clone(), out.try_clone()) else { return };
239	0×	let up = std::thread::spawn(move \|\| {
240	0×	let _ = std::io::copy(&mut guest_r, &mut out_w);
241	0×	let _ = out_w.shutdown(std::net::Shutdown::Write);
242	0×	});
243	0×	let (mut out_r, mut guest_w) = (out, guest);
244	0×	let _ = std::io::copy(&mut out_r, &mut guest_w);
245	0×	let _ = guest_w.shutdown(std::net::Shutdown::Write);
246	0×	let _ = up.join();
247	0×	}
248
249		/// The egress proxy serves a microVM's vsock, which only Linux has.
250		#[cfg(not(unix))]
251		pub struct Proxy;
252
253		#[cfg(not(unix))]
254		pub fn start(_socket: &Path, _allow: Vec<String>, _log: mpsc::Sender<String>) -> std::io::Result<Proxy> {
255		Err(std::io::Error::new(std::io::ErrorKind::Unsupported, "egress serves microVMs, which run only on Linux"))
256		}
257
258		#[cfg(all(test, unix))]
259		mod tests {
260		use super::*;
261
262	6×	fn allow() -> Vec<String> {
263	6×	vec!["*.googleapis.com".into(), "registry.opentofu.org".into()]
264	6×	}
265
266		#[test]
267	1×	fn only_connect_to_an_allowed_name_on_443_is_let_through() {
268	1×	assert_eq!(
269	1×	parse_head("CONNECT storage.googleapis.com:443 HTTP/1.1\r\nHost: x\r\n\r\n"),
270	1×	Ask::Connect { host: "storage.googleapis.com".into(), port: 443 }
271		);
272	1×	assert!(matches!(parse_head("GET http://example.com/ HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("only https")));
273	1×	assert!(matches!(parse_head("CONNECT example.com HTTP/1.1\r\n\r\n"), Ask::Refused(w) if w.contains("no port")));
274	1×	assert_eq!(decide(&allow(), "storage.googleapis.com", 443), Ok(()));
275	1×	assert_eq!(decide(&allow(), "registry.opentofu.org", 443), Ok(()));
276	1×	assert!(decide(&allow(), "example.com", 443).unwrap_err().contains("allow-list"));
277	1×	assert!(decide(&allow(), "registry.opentofu.org", 22).unwrap_err().contains("only port 443"));
278	1×	assert!(decide(&allow(), "142.250.1.1", 443).unwrap_err().contains("names hosts"));
279	1×	}
280
281		#[test]
282	1×	fn private_addresses_are_not_public() {
283	11×	for ip in [
284	1×	"127.0.0.1",
285	1×	"10.1.2.3",
286	1×	"192.168.1.1",
287	1×	"172.16.0.1",
288	1×	"169.254.169.254",
289	1×	"100.64.0.1",
290	1×	"0.0.0.0",
291	1×	"::1",
292	1×	"fd00::1",
293	1×	"fe80::1",
294	1×	"::ffff:10.0.0.1",
295	1×	] {
296	11×	assert!(!public(ip.parse().unwrap()), "{ip}");
297		}
298	2×	for ip in ["142.250.1.1", "2a00:1450:4009::200e"] {
299	2×	assert!(public(ip.parse().unwrap()), "{ip}");
300		}
301	1×	}
302
303		#[test]
304	1×	fn the_proxy_refuses_and_logs_a_host_not_allowed() {
305	1×	let dir = tempfile::tempdir().unwrap();
306	1×	let sock = dir.path().join("v.sock_3128");
307	1×	let (tx, rx) = mpsc::channel();
308	1×	let proxy = start(&sock, allow(), tx).unwrap();
309	2×	let ask = \|head: &str\| {
310	2×	let mut c = UnixStream::connect(&sock).unwrap();
311	2×	c.write_all(head.as_bytes()).unwrap();
312	2×	let mut answer = String::new();
313	2×	c.read_to_string(&mut answer).unwrap();
314	2×	answer
315	2×	};
316	1×	let a = ask("CONNECT example.com:443 HTTP/1.1\r\n\r\n");
317	1×	assert!(a.starts_with("HTTP/1.1 403") && a.contains("not in the sandbox's egress allow-list"), "{a}");
318	1×	assert_eq!(
319	1×	rx.recv_timeout(Duration::from_secs(5)).unwrap(),
320		"egress: refused example.com:443: not in the sandbox's egress allow-list"
321		);
322	1×	let a = ask("GET http://storage.googleapis.com/ HTTP/1.1\r\n\r\n");
323	1×	assert!(a.starts_with("HTTP/1.1 403"), "{a}");
324	1×	assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().contains("only https"));
325	1×	drop(proxy);
326	1×	assert!(!sock.exists(), "the socket goes with the job");
327	1×	}
328		}

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

Keyboard