diff --git a/crates/toolkit/src/egress.rs b/crates/toolkit/src/egress.rs new file mode 100644 index 0000000..9b7e3f6 --- /dev/null +++ b/crates/toolkit/src/egress.rs @@ -0,0 +1,279 @@ +//! The egress proxy: the only way `http_fetch`'s container reaches the network. It speaks the part +//! of SOCKS5 (RFC 1928) that `curl --proxy socks5h://` uses, allows only the call's hosts, only +//! port 443, only public addresses, and after the handshake copies bytes both ways without looking +//! at them. Everything read during the handshake comes from the untrusted side: read exactly what +//! the protocol says, never index or allocate by a number that has not been checked. Spec section 5. + +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); +pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +pub const MAX_CONNECTIONS: usize = 8; +// SOCKS5 reply codes the proxy sends: +pub const NOT_ALLOWED: u8 = 2; +pub const HOST_UNREACHABLE: u8 = 4; +pub const CONNECTION_REFUSED: u8 = 5; +pub const COMMAND_NOT_SUPPORTED: u8 = 7; +pub const ADDRESS_TYPE_NOT_SUPPORTED: u8 = 8; + +/// The call's hosts: one or more patterns separated by ','. An empty piece is an error. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Allow { + patterns: Vec, +} + +impl Allow { + /// One or more host patterns separated by ','. Every piece must pass + /// `proto::hosts::valid_host_pattern`; an empty piece (",", "a.com,", "") is an error. + pub fn parse(list: &str) -> Result { + let mut patterns = Vec::new(); + for piece in list.split(',') { + if !proto::hosts::valid_host_pattern(piece) { + return Err(format!("'{piece}' is not a host or a *.host")); + } + patterns.push(piece.to_string()); + } + Ok(Allow { patterns }) + } + + /// `valid_host(host)` and some pattern `host_matches` it. + pub fn permits(&self, host: &str) -> bool { + proto::hosts::valid_host(host) + && self + .patterns + .iter() + .any(|p| proto::hosts::host_matches(p, host)) + } +} + +/// Name resolution and connecting, so tests need no network. +pub trait Dial: Send + Sync { + fn resolve(&self, host: &str, port: u16) -> std::io::Result>; + fn connect(&self, addr: SocketAddr, timeout: Duration) -> std::io::Result; +} +/// The system's: `(host, port).to_socket_addrs()` and `TcpStream::connect_timeout`. +pub struct SystemDial; + +impl Dial for SystemDial { + fn resolve(&self, host: &str, port: u16) -> std::io::Result> { + let addrs = format!("{}:{}", host, port).to_socket_addrs()?; + Ok(addrs.collect()) + } + fn connect(&self, addr: SocketAddr, timeout: Duration) -> std::io::Result { + TcpStream::connect_timeout(&addr, timeout) + } +} + +pub struct Proxy { + allow: Allow, + dial: Arc, + handshake: Duration, +} +impl Proxy { + pub fn new(allow: Allow, dial: Arc) -> Proxy { + Proxy { + allow, + dial, + handshake: HANDSHAKE_TIMEOUT, + } + } + pub fn with_handshake_timeout(self, handshake: Duration) -> Proxy { + Proxy { handshake, ..self } + } + /// Accept for ever; at most MAX_CONNECTIONS handled at once, more are closed at once. + pub fn serve(self: Arc, listener: UnixListener) -> std::io::Result<()> { + let count = Arc::new(AtomicUsize::new(0)); + for stream in listener.incoming() { + let stream = stream?; + if count.load(Ordering::SeqCst) >= MAX_CONNECTIONS { + drop(stream); + continue; + } + count.fetch_add(1, Ordering::SeqCst); + let proxy = Arc::clone(&self); + let closer = Arc::clone(&count); + let decrement = Arc::clone(&count); + let worker = match stream.try_clone() { + Ok(w) => w, + Err(_) => { + decrement.fetch_sub(1, Ordering::SeqCst); + drop(stream); + continue; + } + }; + match start(move || { + proxy.handle(worker); + closer.fetch_sub(1, Ordering::SeqCst); + }) { + Ok(_) => {} + Err(_) => { + decrement.fetch_sub(1, Ordering::SeqCst); + drop(stream); + } + } + } + Ok(()) + } + /// One connection, from greeting to the end of the copy. + pub fn handle(&self, client: UnixStream) { + let mut client = client; + let deadline = Instant::now() + self.handshake; + + // 1. version, count. Not SOCKS5 -> stop, no reply. + let greeting = match read_n(&mut client, 2, deadline) { + Some(b) if b[0] == 5 => b, + _ => return, + }; + + // 2. methods. No auth method 0 (unauthenticated) -> refuse, else accept. + let methods = match read_n(&mut client, greeting[1] as usize, deadline) { + Some(b) => b, + None => return, + }; + if methods.contains(&0) { + let _ = client.write_all(&[5, 0]); + } else { + let _ = client.write_all(&[5, 0xff]); + return; + } + + // 3. version, command, reserved, kind. + let req = match read_n(&mut client, 4, deadline) { + Some(b) => b, + None => return, + }; + if req[0] != 5 || req[2] != 0 { + return; + } + + // 4. only connect. + if req[1] != 1 { + return refuse(&mut client, COMMAND_NOT_SUPPORTED); + } + // 5. only domain names; do not read the address. + if req[3] != 3 { + return refuse(&mut client, ADDRESS_TYPE_NOT_SUPPORTED); + } + // 6. name length; zero is not allowed. + let length = match read_n(&mut client, 1, deadline) { + Some(b) => b[0], + None => return, + } as usize; + if length == 0 { + return refuse(&mut client, NOT_ALLOWED); + } + // 7. name, then port. + let name = match read_n(&mut client, length, deadline) { + Some(b) => b, + None => return, + }; + let port = match read_n(&mut client, 2, deadline) { + Some(b) => b, + None => return, + }; + let port = u16::from_be_bytes([port[0], port[1]]); + // 8. UTF-8, port 443, host allowed. + let name = match std::str::from_utf8(&name) { + Ok(s) => s, + Err(_) => return refuse(&mut client, NOT_ALLOWED), + }; + if port != 443 || !self.allow.permits(name) { + return refuse(&mut client, NOT_ALLOWED); + } + // 9. resolve; first public address, others skipped never tried. + let addrs = match self.dial.resolve(name, port) { + Ok(a) => a, + Err(_) => return refuse(&mut client, HOST_UNREACHABLE), + }; + let addr = match addrs + .iter() + .find(|a| crate::addr::is_public(a.ip())) + .copied() + { + Some(a) => a, + None => return refuse(&mut client, HOST_UNREACHABLE), + }; + // 10. connect. + let server = match self.dial.connect(addr, CONNECT_TIMEOUT) { + Ok(s) => s, + Err(_) => return refuse(&mut client, CONNECTION_REFUSED), + }; + // 11. success reply. + if client.write_all(&[5, 0, 0, 1, 0, 0, 0, 0, 0, 0]).is_err() { + return; + } + + copy_both_ways(client, server); + } +} + +/// Start a thread that never panics on input: `std::thread::Builder`, which returns an error +/// instead of unwinding, unlike `spawn`. +fn start(f: F) -> std::io::Result> +where + F: FnOnce() + Send + 'static, +{ + std::thread::Builder::new() + .name("egress-handle".into()) + .spawn(f) +} + +/// Refuse: write the SOCKS5 reply and return, which closes the connection. +fn refuse(client: &mut UnixStream, code: u8) { + let _ = client.write_all(&[5, code, 0, 1, 0, 0, 0, 0, 0, 0]); +} + +/// Read exactly `n` bytes before `deadline`, never more. The whole handshake shares one deadline: +/// before every read, set the read timeout to the time left; if none is left, or a read times out, +/// fails, or reads 0 bytes (the peer closed), stop: return `None`. +fn read_n(stream: &mut UnixStream, n: usize, deadline: Instant) -> Option> { + let mut out = Vec::with_capacity(n.min(255)); + while out.len() < n { + let remaining = deadline.checked_duration_since(Instant::now())?; + if stream.set_read_timeout(Some(remaining)).is_err() { + return None; + } + let mut byte = [0u8; 1]; + match stream.read(&mut byte) { + Ok(0) | Err(_) => return None, + Ok(_) => out.push(byte[0]), + } + } + Some(out) +} + +/// After the handshake: copy bytes both ways, passing a half-close on. The client's read timeout +/// is cleared so the copy can run to its natural end. +fn copy_both_ways(mut client: UnixStream, mut server: TcpStream) { + let _ = client.set_read_timeout(None); + let mut server_write = match server.try_clone() { + Ok(s) => s, + Err(_) => return, + }; + let mut client_read = match client.try_clone() { + Ok(s) => s, + Err(_) => return, + }; + + // Second thread: client -> server, then half-close the server's write side. + let upstream = match std::thread::Builder::new() + .name("egress-upstream".into()) + .spawn(move || { + let _ = std::io::copy(&mut client_read, &mut server_write); + let _ = server_write.shutdown(std::net::Shutdown::Write); + }) { + Ok(handle) => handle, + Err(_) => return, + }; + + // Own thread: server -> client, then half-close the client's write side, then join. + let _ = std::io::copy(&mut server, &mut client); + let _ = client.shutdown(std::net::Shutdown::Write); + let _ = upstream.join(); +} diff --git a/crates/toolkit/src/lib.rs b/crates/toolkit/src/lib.rs index 26447b7..c94b3a6 100644 --- a/crates/toolkit/src/lib.rs +++ b/crates/toolkit/src/lib.rs @@ -3,6 +3,7 @@ use proto::tools::{HttpFetchArgs, ReadFileArgs, ShellArgs, WriteFileArgs}; pub mod addr; +pub mod egress; pub mod fetch; pub mod files; pub mod input; diff --git a/crates/toolkit/src/main.rs b/crates/toolkit/src/main.rs index 56059c1..acd35fd 100644 --- a/crates/toolkit/src/main.rs +++ b/crates/toolkit/src/main.rs @@ -1,8 +1,72 @@ +use std::ffi::OsString; use std::io::Write; +use std::os::unix::ffi::OsStrExt; +use std::os::unix::net::UnixListener; use std::process::ExitCode; +use std::sync::Arc; + +use toolkit::egress::{Allow, Proxy, SystemDial}; fn main() -> ExitCode { - let mut rest = std::env::args_os().skip(1); + let args: Vec = std::env::args_os().skip(1).collect(); + match parse_egress_proxy(&args) { + Some(code) => code, + None => run_tool(&args), + } +} + +/// The `egress-proxy --socket --allow ` form. Anything else, including a program name +/// that is not `egress-proxy`, returns `None` so the tool form handles it. +fn parse_egress_proxy(args: &[OsString]) -> Option { + if args.first()?.as_bytes() != b"egress-proxy" { + return None; + } + // The form is `egress-proxy --socket --allow `, in that order. + let socket = match (args.get(1), args.get(2)) { + (Some(flag), Some(path)) if flag.as_bytes() == b"--socket" => path, + _ => return None, + }; + let list = match (args.get(3), args.get(4)) { + (Some(flag), Some(value)) if flag.as_bytes() == b"--allow" => value, + _ => return None, + }; + + let list = match list.to_str() { + Some(s) => s, + None => { + eprint!("toolkit: egress-proxy: --allow: not valid UTF-8"); + return Some(ExitCode::from(2)); + } + }; + let allow = match Allow::parse(list) { + Ok(a) => a, + Err(e) => { + eprint!("toolkit: egress-proxy: --allow: {e}"); + return Some(ExitCode::from(2)); + } + }; + // The directory is fresh for each call, so a file already at the path is a mistake; do not + // remove it first. + let listener = match UnixListener::bind(socket) { + Ok(l) => l, + Err(e) => { + eprint!( + "toolkit: egress-proxy: cannot listen on {path}: {e}", + path = display(socket) + ); + return Some(ExitCode::from(2)); + } + }; + let proxy = Arc::new(Proxy::new(allow, Arc::new(SystemDial))); + if let Err(e) = proxy.serve(listener) { + eprint!("toolkit: egress-proxy: {e}"); + return Some(ExitCode::from(2)); + } + Some(ExitCode::SUCCESS) +} + +fn run_tool(args: &[OsString]) -> ExitCode { + let mut rest = args.iter(); let name = match (rest.next(), rest.next()) { (Some(arg), None) => arg.to_str().unwrap_or("").to_string(), _ => String::new(), @@ -13,3 +77,7 @@ fn main() -> ExitCode { let _ = std::io::stderr().write_all(outcome.stderr.as_bytes()); ExitCode::from(outcome.exit.code()) } + +fn display(o: &OsString) -> String { + String::from_utf8_lossy(o.as_bytes()).into_owned() +} diff --git a/crates/toolkit/tests/egress.rs b/crates/toolkit/tests/egress.rs new file mode 100644 index 0000000..68ea61c --- /dev/null +++ b/crates/toolkit/tests/egress.rs @@ -0,0 +1,451 @@ +//! The egress proxy against a fake resolver and a local echo server: every reply code, the host, +//! port and address checks, the byte copy both ways with half-close, the handshake deadline, the +//! connection limit, and `toolkit egress-proxy` as a program. No test needs a network. Do not edit. + +mod support; + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use support::TempDir; +use toolkit::egress::{ + ADDRESS_TYPE_NOT_SUPPORTED, Allow, COMMAND_NOT_SUPPORTED, CONNECTION_REFUSED, Dial, + HOST_UNREACHABLE, MAX_CONNECTIONS, NOT_ALLOWED, Proxy, +}; + +/// Names resolve from a table; every connection goes to one local echo server, except to +/// addresses listed as refusing. Records every address it was asked to connect to. +struct FakeDial { + names: HashMap>, + refusing: Vec, + echo: SocketAddr, + connected: Mutex>, +} + +impl Dial for FakeDial { + fn resolve(&self, host: &str, _port: u16) -> std::io::Result> { + self.names + .get(host) + .cloned() + .ok_or_else(|| std::io::Error::other("no such name")) + } + fn connect(&self, addr: SocketAddr, _timeout: Duration) -> std::io::Result { + self.connected.lock().unwrap().push(addr); + if self.refusing.contains(&addr) { + return Err(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)); + } + TcpStream::connect(self.echo) + } +} + +fn sa(s: &str) -> SocketAddr { + s.parse().unwrap() +} + +/// An echo server: copies back what it reads, and closes its side after reading the end. +fn echo_server() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + std::thread::spawn(move || { + let mut back = stream.try_clone().unwrap(); + let _ = std::io::copy(&mut stream, &mut back); + let _ = back.shutdown(Shutdown::Write); + }); + } + }); + addr +} + +fn dial() -> Arc { + let mut names = HashMap::new(); + names.insert("example.com".to_string(), vec![sa("93.184.216.34:443")]); + names.insert( + "www.example.org".to_string(), + vec![sa("[2606:2800::1]:443")], + ); + names.insert( + "mixed.example.com".to_string(), + vec![ + sa("10.0.0.1:443"), + sa("100.100.100.100:443"), + sa("1.1.1.1:443"), + ], + ); + names.insert( + "inside.example.com".to_string(), + vec![ + sa("100.101.102.103:443"), + sa("127.0.0.1:443"), + sa("[::1]:443"), + ], + ); + names.insert("refusing.example.com".to_string(), vec![sa("8.8.8.8:443")]); + Arc::new(FakeDial { + names, + refusing: vec![sa("8.8.8.8:443")], + echo: echo_server(), + connected: Mutex::new(Vec::new()), + }) +} + +fn allow() -> Allow { + Allow::parse( + "example.com,*.example.org,mixed.example.com,inside.example.com,refusing.example.com,nowhere.example.com", + ) + .unwrap() +} + +/// A proxy on one end of a socket pair, handling that one connection on its own thread. +fn start(dial: Arc) -> (UnixStream, std::thread::JoinHandle<()>) { + let (client, server) = UnixStream::pair().unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let proxy = Proxy::new(allow(), dial).with_handshake_timeout(Duration::from_millis(300)); + let handle = std::thread::spawn(move || proxy.handle(server)); + (client, handle) +} + +fn request(host: &str, port: u16) -> Vec { + let mut bytes = vec![5, 1, 0, 3, u8::try_from(host.len()).unwrap()]; + bytes.extend_from_slice(host.as_bytes()); + bytes.extend_from_slice(&port.to_be_bytes()); + bytes +} + +fn read_exactly(client: &mut UnixStream, n: usize) -> Vec { + let mut buf = vec![0u8; n]; + client.read_exact(&mut buf).unwrap(); + buf +} + +/// Read until the proxy closes; the bytes read. A close that leaves some of our bytes unread by +/// the proxy arrives as "connection reset" rather than as the end: both mean closed. A timeout +/// does not: the proxy did not close. +fn read_to_close(client: &mut UnixStream) -> Vec { + let mut rest = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + match client.read(&mut buf) { + Ok(0) => return rest, + Ok(n) => rest.extend_from_slice(&buf[..n]), + Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => return rest, + Err(e) => panic!("the proxy did not close: {e}"), + } + } +} + +/// Greeting, then `req`; the reply's code, and that the proxy then closed (for failures). +fn reply_code(req: &[u8]) -> u8 { + let (mut client, handle) = start(dial()); + client.write_all(&[5, 1, 0]).unwrap(); + assert_eq!(read_exactly(&mut client, 2), [5, 0]); + client.write_all(req).unwrap(); + let reply = read_exactly(&mut client, 10); + assert_eq!(reply[0], 5); + assert_eq!(&reply[2..], &[0, 1, 0, 0, 0, 0, 0, 0]); + if reply[1] != 0 { + assert_eq!( + read_to_close(&mut client), + b"", + "a refusal is followed by the close" + ); + handle.join().unwrap(); + } + reply[1] +} + +#[test] +fn an_allowed_host_is_connected_and_bytes_flow_both_ways_until_both_ends_close() { + let dial = dial(); + let (mut client, handle) = start(Arc::clone(&dial)); + client.write_all(&[5, 2, 2, 0]).unwrap(); + assert_eq!( + read_exactly(&mut client, 2), + [5, 0], + "method 0 among others" + ); + client.write_all(&request("example.com", 443)).unwrap(); + assert_eq!( + read_exactly(&mut client, 10), + [5, 0, 0, 1, 0, 0, 0, 0, 0, 0] + ); + client.write_all(b"hello through the tunnel").unwrap(); + assert_eq!(read_exactly(&mut client, 24), b"hello through the tunnel"); + client.shutdown(Shutdown::Write).unwrap(); + assert_eq!( + read_to_close(&mut client), + b"", + "the half-close reached the server and back" + ); + handle.join().unwrap(); + assert_eq!( + *dial.connected.lock().unwrap(), + vec![sa("93.184.216.34:443")] + ); +} + +#[test] +fn a_wildcard_pattern_allows_a_name_under_it() { + assert_eq!(reply_code(&request("www.example.org", 443)), 0); +} + +#[test] +fn a_greeting_without_method_zero_is_answered_ff_and_closed() { + let (mut client, handle) = start(dial()); + client.write_all(&[5, 2, 1, 2]).unwrap(); + assert_eq!(read_exactly(&mut client, 2), [5, 0xff]); + assert_eq!(read_to_close(&mut client), b""); + handle.join().unwrap(); +} + +#[test] +fn another_version_is_closed_without_a_word() { + let (mut client, handle) = start(dial()); + client.write_all(&[4, 1, 0]).unwrap(); + assert_eq!(read_to_close(&mut client), b""); + handle.join().unwrap(); +} + +#[test] +fn only_connect_by_name_to_an_allowed_host_on_443() { + let mut bind = request("example.com", 443); + bind[1] = 2; + assert_eq!(reply_code(&bind), COMMAND_NOT_SUPPORTED); + let mut udp = request("example.com", 443); + udp[1] = 3; + assert_eq!(reply_code(&udp), COMMAND_NOT_SUPPORTED); + assert_eq!( + reply_code(&[5, 1, 0, 1, 93, 184, 216, 34, 1, 187]), + ADDRESS_TYPE_NOT_SUPPORTED + ); + let mut v6 = vec![5, 1, 0, 4]; + v6.extend_from_slice(&[0; 16]); + v6.extend_from_slice(&443u16.to_be_bytes()); + assert_eq!(reply_code(&v6), ADDRESS_TYPE_NOT_SUPPORTED); + for (host, port) in [ + ("example.com", 80), + ("example.com", 8443), + ("evil.test", 443), + ("example.org", 443), // *.example.org does not cover example.org + ("Example.com", 443), // not a valid host name + ("example.com.", 443), + ("127.0.0.1", 443), // an IP literal sent as a name + ("wwwexample.com", 443), + ] { + assert_eq!( + reply_code(&request(host, port)), + NOT_ALLOWED, + "{host}:{port}" + ); + } +} + +#[test] +fn an_empty_or_non_utf8_name_is_not_allowed() { + assert_eq!(reply_code(&[5, 1, 0, 3, 0, 1, 187]), NOT_ALLOWED); + assert_eq!( + reply_code(&[5, 1, 0, 3, 2, 0xff, 0xfe, 1, 187]), + NOT_ALLOWED + ); +} + +#[test] +fn only_public_addresses_are_used() { + let dial = dial(); + let (mut client, handle) = start(Arc::clone(&dial)); + client.write_all(&[5, 1, 0]).unwrap(); + read_exactly(&mut client, 2); + client + .write_all(&request("mixed.example.com", 443)) + .unwrap(); + assert_eq!(read_exactly(&mut client, 10)[1], 0); + drop(client); + handle.join().unwrap(); + assert_eq!( + *dial.connected.lock().unwrap(), + vec![sa("1.1.1.1:443")], + "the private and tailnet addresses were skipped, not tried" + ); +} + +#[test] +fn a_name_with_no_public_address_or_no_address_is_unreachable() { + assert_eq!( + reply_code(&request("inside.example.com", 443)), + HOST_UNREACHABLE + ); + assert_eq!( + reply_code(&request("nowhere.example.com", 443)), + HOST_UNREACHABLE + ); +} + +#[test] +fn a_refused_connection_is_reply_5() { + assert_eq!( + reply_code(&request("refusing.example.com", 443)), + CONNECTION_REFUSED + ); +} + +#[test] +fn a_stalled_handshake_is_closed_at_the_deadline() { + let (mut client, handle) = start(dial()); + client.write_all(&[5, 1, 0]).unwrap(); + assert_eq!(read_exactly(&mut client, 2), [5, 0]); + client.write_all(&[5, 1]).unwrap(); // half a request, then nothing + let started = Instant::now(); + assert_eq!(read_to_close(&mut client), b""); + let took = started.elapsed(); + handle.join().unwrap(); + assert!( + took < Duration::from_secs(2), + "one deadline for the handshake: {took:?}" + ); +} + +#[test] +fn a_handshake_that_trickles_still_ends_at_the_deadline() { + let (mut client, handle) = start(dial()); + let started = Instant::now(); + // One byte every 100 ms: each read is quick, but the handshake as a whole is not. + for byte in [5u8, 1, 0, 5, 1, 0, 3, 11] { + if client.write_all(&[byte]).is_err() { + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + let _ = read_to_close(&mut client); + handle.join().unwrap(); + assert!(started.elapsed() < Duration::from_secs(2)); +} + +#[test] +fn allow_lists_parse_strictly() { + assert!(Allow::parse("example.com").is_ok()); + assert!(Allow::parse("example.com,*.example.org").is_ok()); + for bad in [ + "", + ",", + "example.com,", + "a.com,,b.com", + "Example.com", + "*", + "10.0.0.1", + ] { + assert!(Allow::parse(bad).is_err(), "{bad:?}"); + } + let allow = Allow::parse("*.example.org").unwrap(); + assert!(allow.permits("a.example.org")); + assert!(!allow.permits("example.org")); + assert!(!allow.permits("A.example.org")); +} + +#[test] +fn more_than_the_limit_of_connections_are_closed_at_once() { + let dir = TempDir::new("egress-limit"); + let path = dir.path().join("egress.sock"); + let listener = UnixListener::bind(&path).unwrap(); + let proxy = + Arc::new(Proxy::new(allow(), dial()).with_handshake_timeout(Duration::from_secs(5))); + std::thread::spawn(move || { + let _ = proxy.serve(listener); + }); + // MAX_CONNECTIONS clients that greet and then wait, holding their places. + let mut held = Vec::new(); + for _ in 0..MAX_CONNECTIONS { + let mut c = UnixStream::connect(&path).unwrap(); + c.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + c.write_all(&[5, 1, 0]).unwrap(); + assert_eq!(read_exactly(&mut c, 2), [5, 0]); + held.push(c); + } + let mut extra = UnixStream::connect(&path).unwrap(); + extra + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let _ = extra.write_all(&[5, 1, 0]); + let mut buf = [0u8; 2]; + assert!( + matches!(extra.read(&mut buf), Ok(0) | Err(_)), + "the connection over the limit gets no answer" + ); + // When one place is freed, a new connection is served again. + drop(held.pop()); + std::thread::sleep(Duration::from_millis(200)); + let mut again = UnixStream::connect(&path).unwrap(); + again + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + again.write_all(&[5, 1, 0]).unwrap(); + assert_eq!(read_exactly(&mut again, 2), [5, 0]); +} + +#[test] +fn the_program_listens_where_it_is_told_and_refuses_what_is_not_allowed() { + let dir = TempDir::new("egress-prog"); + let path = dir.path().join("egress.sock"); + let mut child = std::process::Command::new(env!("CARGO_BIN_EXE_toolkit")) + .args([ + "egress-proxy", + "--socket", + path.to_str().unwrap(), + "--allow", + "example.com", + ]) + .spawn() + .unwrap(); + let until = Instant::now() + Duration::from_secs(5); + let mut client = loop { + if let Ok(c) = UnixStream::connect(&path) { + break c; + } + assert!(Instant::now() < until, "the proxy never listened"); + std::thread::sleep(Duration::from_millis(20)); + }; + client + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + client.write_all(&[5, 1, 0]).unwrap(); + assert_eq!(read_exactly(&mut client, 2), [5, 0]); + client.write_all(&request("evil.test", 443)).unwrap(); + assert_eq!(read_exactly(&mut client, 10)[1], NOT_ALLOWED); + child.kill().unwrap(); + child.wait().unwrap(); +} + +#[test] +fn the_program_refuses_bad_arguments() { + let dir = TempDir::new("egress-args"); + let path = dir.path().join("egress.sock"); + let sock = path.to_str().unwrap(); + std::fs::write(dir.path().join("taken"), "").unwrap(); + let taken = dir.path().join("taken"); + for args in [ + vec!["egress-proxy", "--socket", sock, "--allow", "Example.com"], + vec!["egress-proxy", "--socket", sock, "--allow", ""], + vec![ + "egress-proxy", + "--socket", + taken.to_str().unwrap(), + "--allow", + "example.com", + ], + vec!["egress-proxy", "--socket", sock], + vec!["egress-proxy", "--allow", "example.com", "--socket", sock], + ] { + let out = std::process::Command::new(env!("CARGO_BIN_EXE_toolkit")) + .args(&args) + .output() + .unwrap(); + assert_eq!(out.status.code(), Some(2), "{args:?}"); + } +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 073f15b..1878e5a 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M3b/08-toolkit-egress-proxy | 2026-09-23 | done | 2 | fail | none | Copied `docs/plans/M3b/files/crates/toolkit/tests/egress.rs` to `crates/toolkit/tests/`. Wrote `crates/toolkit/src/egress.rs`: the reply-code and timeout constants; `Allow { patterns: Vec }` with `parse` splitting on ',' and rejecting any piece that fails `proto::hosts::valid_host_pattern` (an empty piece like `","`/`"x,"`/`""` errors) and `permits` = `valid_host(host) && any host_matches`; the `Dial` trait and `SystemDial` (`to_socket_addrs()?.collect()` and `connect_timeout`); `Proxy { allow, dial: Arc, handshake }` with `new`/`with_handshake_timeout`/`serve`/`handle`. `handle` shares one deadline across the whole handshake: `read_n` reads exactly the byte count the protocol gives, setting the read timeout to `deadline.checked_duration_since(now)` before each read and returning `None` (stop, no reply) on no time left, a timeout, a failed read, or 0 bytes — so a client trickling one byte per 100 ms is still cut at the deadline. The 11 exits are in order (egress.rs:133 version, :144 methods 0, :154 version/reserved, :160 command, :164 kind without reading the address, :172 zero length, :187 non-UTF-8, :190 port/host, :195 resolve / :203 no public addr, :208 connect, :211 success reply); step 9 takes the first address where `crate::addr::is_public` is true via `.find`, never trying the skipped non-public ones. `serve` is a single-threaded accept loop with an `AtomicUsize` count: over the limit is dropped at once, otherwise a `std::thread::Builder` thread handles it and uncounts itself on return, and `serve` never joins (so the second client is not starved). After the handshake the read timeout is cleared and two `try_clone`'d threads copy both ways with a half-close each way, then join. `main.rs` gained the `egress-proxy --socket --allow ` form before the tool form via `std::env::args_os().skip(1)` (without `skip(1)` the first element is the program path and the form never matches); `Allow::parse`/bind (no removal first — a pre-existing file is a mistake)/`serve` errors print to stderr and exit 2. `lib.rs` gained `pub mod egress;`. First gate failed on clippy `question_mark` (accept loop → `stream?`) and `manual_contains` (`methods.contains(&0)`); fixed both and `make gate` prints `gate: ok`. 15 egress tests pass ten runs in a row. | ? | | M3b/07-toolkit-addr | 2026-09-23 | done | 1 | pass | none | Wrote `crates/toolkit/src/addr.rs`: `is_public(ip)` matches on `IpAddr` and dispatches to `is_public_v4`/`is_public_v6`. The IPv4 function checks the 13 refused ranges in table order with early returns (lines 16-52) then returns `true`. The IPv6 function checks the two "judge as IPv4" rows first — `is_ipv4_mapped` for `::ffff:0:0/96` (lines 60-63) and `is_nat64` for `64:ff9b::/96` (lines 64-67) — reconstructing the last 32 bits as an `Ipv4Addr` via `(u32::from(s[6]) << 16) | u32::from(s[7])` with no `as` casts, then judging it through `is_public_v4`; the remaining rows `::/96` (line 68), `fc00::/7` (71), `fe80::/10` (74), `ff00::/8` (77) and `2001:db8::/32` (80) follow. `cargo fmt --all` first. Added `pub mod addr;` to `lib.rs` in alphabetical position (before `fetch`). All 4 addr tests pass; `make gate` prints `gate: ok` on the first run. | ? | | M3b/06-toolkit-fetch | 2026-09-23 | done | 3 | fail | none | Wrote `crates/toolkit/src/fetch.rs`: `CURL=/bin/curl`, `PROXY=socks5h://localhost/run/egress/egress.sock`, `CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt`, `MAX_STDERR=65536`. `curl_args(url)` returns the 21 fixed strings in order ending in `--url `. `fetch` calls `fetch_with(Path::new(CURL), args)`. `fetch_with`: spawn curl with `curl_args(&args.url)`, null stdin, stdout/stderr piped (spawn failure -> `"http_fetch: cannot start {}: {e}"`); read stderr on its own thread via the `read_capped` helper (8 KiB buffer, keeps the first 64 KiB and drains/discards the rest so curl never blocks on a full pipe), main thread reads stdout to the end; `wait()` then join the thread (a panicked reader falls back to empty stderr); exit 0 -> `done(from_utf8_lossy(body))` with the `--write-out` status line already in it, else the first non-blank trimmed stderr line as `why` or `"curl exited {code}"` or `"curl was killed"` when there is no code, wrapped as `"http_fetch: {url}: {why}"`, and a wait failure -> `"http_fetch: cannot wait for curl: {e}"`. `lib.rs` gained `pub mod fetch` (between `files` and `input`) and an `"http_fetch"` arm parsing `HttpFetchArgs` like the other three tools. First gate run failed on a rustfmt line-wrap of the spawn-error return; second on clippy `manual_unwrap_or_default` for the stderr-join match, switched to `unwrap_or_default()`. 7 fetch tests pass ten runs; `make gate` prints `gate: ok`. | ? | | M3b/05-toolkit-shell | 2026-09-23 | done | 2 | fail | none | Wrote `crates/toolkit/src/shell.rs` (`SHELL=/bin/sh`, `DEFAULT_CWD=/tmp`, `MAX_OUTPUT=1048576`). `shell()`: `cwd` defaults to `/tmp`, `Path::is_dir` check returns `"shell: {cwd}: no such directory"`; one `io::pipe` with a `try_clone`'d second write end (stdout gets `writer`, stderr gets `writer2`); `Command::new(SHELL).arg("-c").arg(command).current_dir(cwd)` with null stdin, `spawn`, then `drop(command)` so the parent holds no write end and reading reaches EOF (the hang fix); an 8 KiB-buffer read loop keeping the first `MAX_OUTPUT` bytes and draining+discarding the rest with a `dropped` flag, retrying `Interrupted`, decoding with `from_utf8_lossy`; `wait()` then appending the dropped line and the exit/signal/none status in that order via a `match` on `code()`/`signal()` → `Outcome::done`. `lib.rs` gained `pub mod shell` (after `input`) and a `"shell"` arm parsing `ShellArgs` like the other two tools. One logic bug caught by the limit test: when the buffer fills `MAX_OUTPUT` exactly the drain branch never set `dropped`, fixed by marking `dropped` in the drain branch. First gate failed on an unused `Write` import (removed); 8 shell tests pass five runs, `make gate` prints `gate: ok`. | ? |