crates/installer/nordisk-pairing/src/egress.rs
//! **The estate's way out, through its front** (owner 2026-10-23, the lean estate: ONE public
//! IPv4 per estate, on the front). Main and the twin are laid with a utility address alone, so
//! nothing they dial leaves the provider by itself: the ONE box is reached over the utility
//! network (`fleet_at`), the front over the tunnel, and everything else on the internet (an
//! off-site copy's object store, an upstream registry, a mail relay over TLS) through this door
//! on the front.
//!
//! The door is an HTTP `CONNECT` proxy on the front's OVERLAY address ([`EGRESS_PORT`] on
//! [`crate::OVERLAY_FRONT`]), so only a box on the estate's tunnel reaches it. It opens a TCP
//! connection to a NAMED host on a TLS port ([`TLS_PORTS`]) and then copies bytes both ways: it
//! sees no plaintext, holds no certificate and no credential, and it names the host to the
//! resolver itself (the boxes need no DNS). A destination that resolves into a private, loopback,
//! link-local or shared network is refused: the door is a way OUT, never into the provider's
//! utility network or back into the tunnel.
//!
//! A box learns the door from its feed ([`FEED_EGRESS`], `http://10.44.0.1:3128`), fed by the
//! shop to boxes laid with no public IPv4, and keeps it process-wide ([`set_proxy`], [`proxy`])
//! for the HTTP clients it builds. A client with no proxy of its own (the mail relay's, lettre)
//! dials a [`Forward`] instead: a loopback port carried through the door ([`dial`]). Std only:
//! every build links it, the open one too.
use std::io::{Read, Write};
use std::net::{IpAddr, Shutdown, SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
use std::sync::OnceLock;
use std::time::Duration;
/// The door's port on the front's overlay address.
pub const EGRESS_PORT: u16 = 3128;
/// The feed key that names the door to a box behind its front.
pub const FEED_EGRESS: &str = "egress";
/// The ports a box may leave on: HTTPS, SMTP over TLS (implicit) and submission (STARTTLS).
pub const TLS_PORTS: [u16; 3] = [443, 465, 587];
/// The longest request head the door reads before it refuses.
pub const MAX_HEAD: usize = 8 * 1024;
/// How long the door waits for a destination to answer its connect.
pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
/// Pure: **the door's URL**, as a box is fed it.
pub fn door_url() -> String {
format!("http://{}:{EGRESS_PORT}", crate::OVERLAY_FRONT)
}
/// Pure: **the destination of a `CONNECT` request head** (`CONNECT host:port HTTP/1.1`), or the
/// refusal that names what is wrong.
pub fn parse_connect(head: &str) -> Result<(String, u16), String> {
let line = head.lines().next().unwrap_or_default();
let mut words = line.split_whitespace();
match (words.next(), words.next(), words.next()) {
(Some("CONNECT"), Some(target), Some(v)) if v.starts_with("HTTP/1.") => {
let (host, port) = target.rsplit_once(':').ok_or_else(|| format!("CONNECT {target}: no port"))?;
let host = host.trim_start_matches('[').trim_end_matches(']');
let port: u16 = port.parse().map_err(|_| format!("CONNECT {target}: the port is not a number"))?;
if host.is_empty() {
return Err(format!("CONNECT {target}: no host"));
}
Ok((host.to_ascii_lowercase(), port))
}
_ => Err(format!("not a CONNECT request: {line:?}")),
}
}
/// Pure: **an address the door never connects to**: private, loopback, link-local, shared
/// (100.64/10), unspecified, broadcast or multicast; IPv6 loopback, unique local and link-local.
pub fn inside(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v) => crate::fleet::bench_address(ip) || v.is_unspecified() || v.is_broadcast() || v.is_multicast(),
IpAddr::V6(v) => crate::fleet::bench_address(ip) || v.is_unspecified() || v.is_multicast() || v.to_ipv4_mapped().is_some_and(|m| inside(IpAddr::V4(m))),
}
}
/// Pure: **whether the door takes this request**: a TLS port, and every address the host
/// resolves to outside ([`inside`]). `resolved` is what the resolver said.
pub fn admits(host: &str, port: u16, resolved: &[SocketAddr]) -> Result<(), String> {
if !TLS_PORTS.contains(&port) {
return Err(format!("REFUSED [egress-port] {host}:{port}: the estate leaves on {TLS_PORTS:?} only"));
}
if resolved.is_empty() {
return Err(format!("REFUSED [egress-unresolved] {host}: the resolver named no address"));
}
if let Some(a) = resolved.iter().find(|a| inside(a.ip())) {
return Err(format!("REFUSED [egress-inside] {host} is {}: the door is a way out, never into a private network", a.ip()));
}
Ok(())
}
/// What the door checks; [`Policy::OUT`] in every build, a test's own over loopback.
#[derive(Clone, Copy)]
pub struct Policy {
/// Whether a peer may use the door at all.
pub peer: fn(IpAddr) -> bool,
/// Whether the door takes a destination ([`admits`]).
pub admits: fn(&str, u16, &[SocketAddr]) -> Result<(), String>,
}
impl Policy {
/// The estate's: peers on the tunnel only, destinations outside only.
pub const OUT: Policy = Policy { peer: on_tunnel, admits };
}
fn on_tunnel(ip: IpAddr) -> bool {
matches!(ip, IpAddr::V4(v) if crate::in_overlay(v))
}
/// At most this many connections through the door at once (A32-14): one thread each, so a
/// peer cannot spend the front's threads without bound.
pub const MAX_CONNS: usize = 64;
/// At most this many of them from ONE peer (RA4-13): a single box on the tunnel can no longer
/// hold every slot and refuse every other box's way out.
pub const PER_PEER: usize = 16;
/// The whole request head must arrive within this, however it is dripped (RA4-13: the old
/// timeout was per 1-byte read, so a byte every 29 s held a slot for ever).
pub const HEAD_DEADLINE: Duration = Duration::from_secs(30);
/// A spliced tunnel that carries no byte either way for this long is closed and its slot given
/// back (RA4-13: before, a tunnel nobody spoke on held its slot until a side closed it).
pub const IDLE: Duration = Duration::from_secs(300);
/// The door says at most one line per this window; the lines it did not say are counted into
/// the next one (RA4-13: each refusal was an eprintln, so a flood of refusals flooded the log).
pub const SAY_EVERY: Duration = Duration::from_secs(60);
/// What the door holds a connection to: the caps, the head deadline and the idle timeout.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Limits {
/// Connections through the door at once.
pub total: usize,
/// Connections from one peer at once.
pub per_peer: usize,
/// The whole request head arrives within this.
pub head: Duration,
/// A tunnel silent both ways for this long is closed.
pub idle: Duration,
}
impl Limits {
/// The estate's door: [`MAX_CONNS`], [`PER_PEER`], [`HEAD_DEADLINE`], [`IDLE`].
pub const DOOR: Limits = Limits { total: MAX_CONNS, per_peer: PER_PEER, head: HEAD_DEADLINE, idle: IDLE };
}
/// **The door's places**, counted for the whole door and per peer (RA4-13). A place is a
/// [`Slot`], given back when it is dropped.
pub struct Slots {
total: usize,
per_peer: usize,
held: std::sync::Mutex<(usize, std::collections::HashMap<IpAddr, usize>)>,
}
impl Slots {
/// A door with `total` places, at most `per_peer` of them for one peer.
pub fn new(total: usize, per_peer: usize) -> std::sync::Arc<Slots> {
std::sync::Arc::new(Slots { total, per_peer, held: std::sync::Mutex::new((0, std::collections::HashMap::new())) })
}
/// A place for one connection from `peer`, or the refusal that names which cap is full:
/// `[egress-full]` for the door's (said first), `[egress-peer-full]` for the peer's own.
pub fn take(self: &std::sync::Arc<Self>, peer: IpAddr) -> Result<Slot, String> {
let peer = peer.to_canonical();
let mut held = self.held.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let mine = held.1.get(&peer).copied().unwrap_or(0);
if held.0 >= self.total {
return Err(format!("REFUSED [egress-full] {} connections are open, the door's most at once", held.0));
}
if mine >= self.per_peer {
return Err(format!("REFUSED [egress-peer-full] {peer} holds {mine} connections, its most at once"));
}
held.0 += 1;
*held.1.entry(peer).or_insert(0) += 1;
Ok(Slot { slots: std::sync::Arc::clone(self), peer })
}
/// How many places are held now (the whole door).
pub fn held(&self) -> usize {
self.held.lock().unwrap_or_else(std::sync::PoisonError::into_inner).0
}
}
/// One connection's place at the door, given back when it ends (or its thread never starts).
pub struct Slot {
slots: std::sync::Arc<Slots>,
peer: IpAddr,
}
impl Drop for Slot {
fn drop(&mut self) {
let mut held = self.slots.held.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
held.0 = held.0.saturating_sub(1);
if let Some(n) = held.1.get_mut(&self.peer) {
*n -= 1;
if *n == 0 {
held.1.remove(&self.peer);
}
}
}
}
/// **The door says a line at most once per window** (RA4-13); what it held back is counted and
/// said with the next line it says.
pub struct Said {
every: Duration,
state: std::sync::Mutex<(Option<std::time::Instant>, u64)>,
}
impl Said {
/// At most one line per `every`.
pub fn new(every: Duration) -> Said {
Said { every, state: std::sync::Mutex::new((None, 0)) }
}
/// Pure on `now`: `Some(held back since the last line)` when a line may be said now,
/// `None` when it is held back (and counted).
pub fn may_say(&self, now: std::time::Instant) -> Option<u64> {
let mut s = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
match s.0 {
Some(last) if now.saturating_duration_since(last) < self.every => {
s.1 += 1;
None
}
_ => {
s.0 = Some(now);
Some(std::mem::take(&mut s.1))
}
}
}
/// Says `line` on stderr when the window allows it.
pub fn say(&self, line: &str) {
match self.may_say(std::time::Instant::now()) {
Some(0) => eprintln!("egress: {line}"),
Some(n) => eprintln!("egress: {line} ({n} more lines held back since the last one)"),
None => {}
}
}
}
/// **The door, serving** on `bind` (the front's overlay address in an estate), one thread per
/// connection, within [`Limits::DOOR`]. Returns the bound address. Errors only when it cannot
/// bind.
pub fn serve(bind: SocketAddr, policy: Policy) -> std::io::Result<SocketAddr> {
serve_with(bind, policy, Limits::DOOR)
}
/// [`serve`] with at most `cap` connections at once (and at most `cap` from one peer).
pub fn serve_capped(bind: SocketAddr, policy: Policy, cap: usize) -> std::io::Result<SocketAddr> {
serve_with(bind, policy, Limits { total: cap, per_peer: cap, ..Limits::DOOR })
}
/// [`serve`] within `limits`.
pub fn serve_with(bind: SocketAddr, policy: Policy, limits: Limits) -> std::io::Result<SocketAddr> {
let listener = TcpListener::bind(bind)?;
let at = listener.local_addr()?;
let slots = Slots::new(limits.total, limits.per_peer);
let said = std::sync::Arc::new(Said::new(SAY_EVERY));
// A28 M18, A32-14, stays threads: the door's accept loop (its start failure answered) and one
// short thread per connection, which ends with it, at most `limits.total` at once and
// `limits.per_peer` from one peer (nordisk-pairing is std-only; a connection that cannot get
// a slot or a thread is closed, the first by name).
std::thread::Builder::new().name("egress".into()).spawn(move || {
for mut conn in listener.incoming().flatten() {
let peer = match conn.peer_addr() {
Ok(p) => p.ip(),
Err(_) => continue,
};
let slot = match slots.take(peer) {
Ok(slot) => slot,
Err(why) => {
let _ = conn.write_all(format!("HTTP/1.1 503 Service Unavailable\r\n\r\n{why}; ask again\r\n").as_bytes());
said.say(&why);
continue;
}
};
let said = std::sync::Arc::clone(&said);
let _ = std::thread::Builder::new().name("egress-conn".into()).spawn(move || {
let _slot = slot;
if let Err(why) = one(conn, policy, limits) {
said.say(&why);
}
});
}
})?;
Ok(at)
}
fn one(mut conn: TcpStream, policy: Policy, limits: Limits) -> Result<(), String> {
let peer = conn.peer_addr().map_err(|e| e.to_string())?;
if !(policy.peer)(peer.ip()) {
let _ = conn.write_all(b"HTTP/1.1 403 Forbidden\r\n\r\n");
return Err(format!("REFUSED [egress-peer] {peer}: not on the estate's tunnel"));
}
// One deadline for the whole head, not one per byte (RA4-13).
let until = std::time::Instant::now() + limits.head;
let mut head = Vec::new();
let mut byte = [0u8; 1];
while !head.ends_with(b"\r\n\r\n") {
if head.len() >= MAX_HEAD {
let _ = conn.write_all(b"HTTP/1.1 431 Request Header Fields Too Large\r\n\r\n");
return Err(format!("{peer}: a request head over {MAX_HEAD} bytes"));
}
let left = until.saturating_duration_since(std::time::Instant::now());
if left.is_zero() {
let _ = conn.write_all(b"HTTP/1.1 408 Request Timeout\r\n\r\n");
return Err(format!("REFUSED [egress-head-slow] {peer}: no whole request head within {:?}", limits.head));
}
conn.set_read_timeout(Some(left)).map_err(|e| e.to_string())?;
match conn.read(&mut byte) {
Ok(1) => head.push(byte[0]),
Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => continue,
_ => return Err(format!("{peer}: closed before a whole request head")),
}
}
let (host, port) = match parse_connect(&String::from_utf8_lossy(&head)) {
Ok(d) => d,
Err(why) => {
let _ = conn.write_all(b"HTTP/1.1 405 Method Not Allowed\r\n\r\n");
return Err(format!("{peer}: {why}"));
}
};
let resolved: Vec<SocketAddr> = (host.as_str(), port).to_socket_addrs().map(|a| a.collect()).unwrap_or_default();
if let Err(why) = (policy.admits)(&host, port, &resolved) {
let _ = conn.write_all(b"HTTP/1.1 403 Forbidden\r\n\r\n");
return Err(format!("{peer}: {why}"));
}
let out = resolved.iter().find_map(|a| TcpStream::connect_timeout(a, CONNECT_TIMEOUT).ok());
let Some(out) = out else {
let _ = conn.write_all(b"HTTP/1.1 502 Bad Gateway\r\n\r\n");
return Err(format!("{peer}: {host}:{port} did not answer"));
};
conn.set_read_timeout(None).map_err(|e| e.to_string())?;
conn.write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n").map_err(|e| e.to_string())?;
splice(conn, out, limits.idle);
Ok(())
}
/// Copies both ways until either side closes, or until no byte has moved either way for `idle`
/// (then both sides are shut).
fn splice(a: TcpStream, b: TcpStream, idle: Duration) {
let (Ok(a2), Ok(b2)) = (a.try_clone(), b.try_clone()) else { return };
let start = std::time::Instant::now();
let last = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
let last2 = std::sync::Arc::clone(&last);
// A28 M18, stays a thread: one direction of the splice, joined below.
let up = std::thread::spawn(move || pump(a2, b2, idle, start, &last2));
pump(b, a, idle, start, &last);
let _ = up.join();
}
/// One direction of a [`splice`]: a read that times out closes both sides only when the
/// OTHER direction was silent too (`last`, in ms since `start`, is the latest byte either way).
fn pump(mut from: TcpStream, mut to: TcpStream, idle: Duration, start: std::time::Instant, last: &std::sync::atomic::AtomicU64) {
use std::sync::atomic::Ordering::SeqCst;
let _ = from.set_read_timeout(Some(idle));
let _ = to.set_write_timeout(Some(idle));
let mut buf = [0u8; 16 * 1024];
loop {
match from.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
if to.write_all(&buf[..n]).is_err() {
break;
}
last.store(start.elapsed().as_millis() as u64, SeqCst);
}
Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {
let silent = start.elapsed().saturating_sub(Duration::from_millis(last.load(SeqCst)));
if silent >= idle {
let _ = from.shutdown(Shutdown::Both);
let _ = to.shutdown(Shutdown::Both);
return;
}
}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(_) => break,
}
}
let _ = to.shutdown(Shutdown::Write);
}
static PROXY: OnceLock<String> = OnceLock::new();
/// **This box leaves through its front's door** (the feed's [`FEED_EGRESS`]): kept for the
/// process's HTTP clients. The first call wins; a blank value is not one.
pub fn set_proxy(url: &str) {
let url = url.trim();
if !url.is_empty() {
let _ = PROXY.set(url.to_string());
}
}
/// The door this box leaves through, when it was fed one (`None`: it has its own way out).
pub fn proxy() -> Option<&'static str> {
PROXY.get().map(String::as_str)
}
// ── the client half: a box that leaves through its door ──────────────────────
/// Pure: **the request a box sends its door** for `host:port` (an IPv6 literal bracketed).
pub fn connect_head(host: &str, port: u16) -> String {
let h = if host.contains(':') { format!("[{host}]") } else { host.to_string() };
format!("CONNECT {h}:{port} HTTP/1.1\r\nHost: {h}:{port}\r\n\r\n")
}
/// Pure: **the door's address from the URL a box was fed** (`http://10.44.0.1:3128`). An
/// address, never a name: a box behind its front has no DNS.
pub fn door_addr(url: &str) -> Result<SocketAddr, String> {
let rest = url.trim().strip_prefix("http://").ok_or_else(|| format!("the door {url:?} is not an http:// URL"))?;
let hostport = rest.trim_end_matches('/');
hostport.parse::<SocketAddr>().map_err(|_| format!("the door {url:?} names no address and port (a box behind its front resolves no names)"))
}
/// Pure: **the door's answer to a box's request**: `200` is a tunnel; anything else is the
/// refusal, named by its status line.
pub fn established(head: &str) -> Result<(), String> {
let line = head.lines().next().unwrap_or_default().trim();
let mut words = line.split_whitespace();
match (words.next(), words.next()) {
(Some(v), Some("200")) if v.starts_with("HTTP/1.") => Ok(()),
_ => Err(format!("the egress door answered {line:?}")),
}
}
/// **A tunnel through the door** to `host:port`: the TCP connection to the door, the request,
/// and its `200` read. What is read and written after is the destination's (end to end).
pub fn dial(door: &str, host: &str, port: u16) -> std::io::Result<TcpStream> {
let other = |why: String| std::io::Error::other(why);
let at = door_addr(door).map_err(other)?;
let mut conn = TcpStream::connect_timeout(&at, CONNECT_TIMEOUT).map_err(|e| other(format!("the egress door {at} did not answer: {e}")))?;
conn.set_read_timeout(Some(CONNECT_TIMEOUT + Duration::from_secs(15)))?;
conn.write_all(connect_head(host, port).as_bytes())?;
let mut head = Vec::new();
let mut byte = [0u8; 1];
while !head.ends_with(b"\r\n\r\n") {
if head.len() >= MAX_HEAD {
return Err(other(format!("the egress door {at} answered a head over {MAX_HEAD} bytes")));
}
match conn.read(&mut byte) {
Ok(1) => head.push(byte[0]),
_ => return Err(other(format!("the egress door {at} closed before it answered for {host}:{port}"))),
}
}
established(&String::from_utf8_lossy(&head)).map_err(|why| other(format!("{host}:{port} through {at}: {why}")))?;
conn.set_read_timeout(None)?;
Ok(conn)
}
/// **A loopback forward through the door, for a client that speaks no proxy** (A19 row 13:
/// lettre, so the sign-in-again mail of a box behind its front). The client dials
/// [`Forward::addr`] and names the DESTINATION in its TLS (the relay's name, so its
/// certificate is checked as ever); each of its connections is carried through the door as
/// `CONNECT host:port`. The first tunnel is dialled at [`Forward::open`], so a refusing door is
/// said there, by its status, before any mail is written. Dropping it stops the forward.
pub struct Forward {
at: SocketAddr,
stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
}
impl Forward {
/// Opens the forward to `host:port` through `door` (the URL the box was fed).
pub fn open(door: &str, host: &str, port: u16) -> std::io::Result<Forward> {
let first = dial(door, host, port)?;
let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))?;
let at = listener.local_addr()?;
let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let (door, host, halt) = (door.to_string(), host.to_string(), stop.clone());
// A28 M18, stays threads: the forward's accept loop (stopped by its `Forward`) and one per connection.
std::thread::Builder::new().name("egress-forward".into()).spawn(move || {
let mut first = Some(first);
for conn in listener.incoming() {
if halt.load(std::sync::atomic::Ordering::SeqCst) {
break;
}
let Ok(conn) = conn else { continue };
let out = match first.take() {
Some(t) => Ok(t),
None => dial(&door, &host, port),
};
match out {
Ok(out) => {
let _ = std::thread::Builder::new().name("egress-forward-conn".into()).spawn(move || splice(conn, out, IDLE));
}
Err(why) => eprintln!("egress forward: {why}"),
}
}
})?;
Ok(Forward { at, stop })
}
/// Where the client dials (loopback only).
pub fn addr(&self) -> SocketAddr {
self.at
}
}
impl Drop for Forward {
fn drop(&mut self) {
self.stop.store(true, std::sync::atomic::Ordering::SeqCst);
// Wakes the accept, which then sees the stop.
let _ = TcpStream::connect_timeout(&self.at, Duration::from_secs(1));
}
}
#[cfg(test)]
mod tests {
/// ★ **The door holds at most its cap of connections; one over is refused by name, and a
/// slot frees when a connection ends** (A32-14: a thread per connection, without bound, let
/// one peer spend the front's threads).
#[test]
fn the_door_refuses_a_connection_over_its_cap_by_name_and_frees_the_slot() {
let door = serve_capped("127.0.0.1:0".parse().unwrap(), Policy { peer: |_| true, admits }, 2).unwrap();
// Two that say nothing hold their slots (the door waits for their heads).
let held: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(door).unwrap()).collect();
std::thread::sleep(Duration::from_millis(200));
let mut over = TcpStream::connect(door).unwrap();
over.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
let mut said = String::new();
let _ = over.read_to_string(&mut said);
assert!(said.starts_with("HTTP/1.1 503") && said.contains("egress-full"), "one over the cap is refused by name: {said:?}");
drop(held);
std::thread::sleep(Duration::from_millis(300));
let mut again = TcpStream::connect(door).unwrap();
again.set_read_timeout(Some(Duration::from_millis(500))).unwrap();
let mut b = [0u8; 1];
let r = again.read(&mut b);
assert!(matches!(&r, Err(e) if e.kind() == std::io::ErrorKind::WouldBlock || e.kind() == std::io::ErrorKind::TimedOut), "a freed slot takes the next connection (it waits for its head): {r:?}");
}
use super::*;
/// ★ **One peer with 64 connections does not block a second** (RA4-13: one count of 64 for
/// the whole door let a single overlay peer hold every slot and 503 every box's way out).
/// A peer is held to [`PER_PEER`], each refusal named by the cap that is full.
#[test]
fn one_peer_with_64_connections_does_not_block_a_second() {
let slots = Slots::new(MAX_CONNS, PER_PEER);
let a: IpAddr = "10.44.0.2".parse().unwrap();
let b: IpAddr = "10.44.0.3".parse().unwrap();
let tries: Vec<Result<Slot, String>> = (0..64).map(|_| slots.take(a)).collect();
let held = tries.iter().filter(|t| t.is_ok()).count();
assert_eq!(held, PER_PEER, "one peer holds at most its own cap");
assert!(tries.iter().filter_map(|t| t.as_ref().err()).all(|e| e.contains("[egress-peer-full]")), "each refusal names the peer's cap");
let second = slots.take(b);
assert!(second.is_ok(), "a second peer still gets a slot: {:?}", second.err());
drop(second);
// The door's own cap still holds: four peers at their caps fill it, a fifth is refused by the door's name.
let mut all: Vec<Slot> = tries.into_iter().flatten().collect();
for p in 3..6u8 {
all.extend((0..PER_PEER).map(|_| slots.take(IpAddr::from([10, 44, 0, p])).unwrap()));
}
assert_eq!(slots.held(), MAX_CONNS);
let fifth = slots.take("10.44.0.9".parse().unwrap()).err().unwrap();
assert!(fifth.contains("[egress-full]"), "{fifth}");
drop(all);
assert_eq!(slots.held(), 0, "every slot is given back");
assert!(slots.take(a).is_ok(), "a peer's count is given back too");
assert_eq!(Limits::DOOR, Limits { total: 64, per_peer: 16, head: Duration::from_secs(30), idle: Duration::from_secs(300) });
}
/// ★ **The door refuses one peer over its own cap by name, over a socket** (RA4-13).
#[test]
fn the_door_refuses_a_peer_over_its_own_cap_by_name() {
let limits = Limits { total: 8, per_peer: 2, ..Limits::DOOR };
let door = serve_with("127.0.0.1:0".parse().unwrap(), Policy { peer: |_| true, admits }, limits).unwrap();
let held: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(door).unwrap()).collect();
std::thread::sleep(Duration::from_millis(200));
let mut over = TcpStream::connect(door).unwrap();
over.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
let mut said = String::new();
let _ = over.read_to_string(&mut said);
assert!(said.starts_with("HTTP/1.1 503") && said.contains("egress-peer-full"), "one over the peer's cap is refused by name: {said:?}");
drop(held);
}
/// ★ **A head dripped a byte at a time is cut at ONE deadline for the whole head** (RA4-13:
/// the 30 s was per 1-byte read, so a byte every 29 s held a slot for ever), and its slot is
/// given back.
#[test]
fn a_head_dripped_a_byte_at_a_time_is_cut_at_its_deadline() {
let limits = Limits { total: 1, per_peer: 1, head: Duration::from_millis(600), ..Limits::DOOR };
let door = serve_with("127.0.0.1:0".parse().unwrap(), Policy { peer: |_| true, admits }, limits).unwrap();
let mut c = TcpStream::connect(door).unwrap();
let mut w = c.try_clone().unwrap();
// Drips "CONNECT x:443 ..." a byte per 100 ms: each read is quick, the head never whole.
std::thread::spawn(move || {
for b in b"CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n" {
if w.write_all(&[*b]).is_err() {
return;
}
std::thread::sleep(Duration::from_millis(100));
}
});
c.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
let begun = std::time::Instant::now();
let mut said = String::new();
let _ = c.read_to_string(&mut said);
assert!(said.starts_with("HTTP/1.1 408"), "a head not whole by its deadline is refused: {said:?}");
assert!(begun.elapsed() < Duration::from_secs(3), "cut at the deadline, not after the drip: {:?}", begun.elapsed());
std::thread::sleep(Duration::from_millis(200));
let mut next = TcpStream::connect(door).unwrap();
next.set_read_timeout(Some(Duration::from_millis(300))).unwrap();
let mut b = [0u8; 1];
let r = next.read(&mut b);
assert!(matches!(&r, Err(e) if e.kind() == std::io::ErrorKind::WouldBlock || e.kind() == std::io::ErrorKind::TimedOut), "its slot was given back: {r:?}");
}
/// ★ **A tunnel silent both ways for its idle time is closed and frees its slot** (RA4-13:
/// a spliced tunnel had no idle timeout); one that keeps talking is not.
#[test]
fn an_idle_tunnel_is_closed_and_frees_its_slot() {
let echo = TcpListener::bind("127.0.0.1:0").unwrap();
let echo_at = echo.local_addr().unwrap();
std::thread::spawn(move || {
for c in echo.incoming().flatten() {
std::thread::spawn(move || {
let mut r = c.try_clone().unwrap();
let mut w = c;
let _ = std::io::copy(&mut r, &mut w);
});
}
});
let open = Policy { peer: |_| true, admits: |_, _, r| if r.is_empty() { Err("none".into()) } else { Ok(()) } };
let limits = Limits { total: 1, per_peer: 1, idle: Duration::from_millis(400), ..Limits::DOOR };
let door = serve_with("127.0.0.1:0".parse().unwrap(), open, limits).unwrap();
let mut c = TcpStream::connect(door).unwrap();
c.write_all(format!("CONNECT 127.0.0.1:{} HTTP/1.1\r\n\r\n", echo_at.port()).as_bytes()).unwrap();
let mut reply = [0u8; 39];
c.read_exact(&mut reply).unwrap();
// Talking keeps it open past its idle time.
for _ in 0..4 {
c.write_all(b"ping!").unwrap();
let mut back = [0u8; 5];
c.read_exact(&mut back).unwrap();
std::thread::sleep(Duration::from_millis(200));
}
// Silence closes it.
c.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
let begun = std::time::Instant::now();
let mut b = [0u8; 1];
let r = c.read(&mut b);
assert!(matches!(r, Ok(0)) || matches!(&r, Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset), "an idle tunnel is closed: {r:?}");
assert!(begun.elapsed() < Duration::from_secs(3), "closed near its idle time: {:?}", begun.elapsed());
std::thread::sleep(Duration::from_millis(200));
let mut next = TcpStream::connect(door).unwrap();
next.set_read_timeout(Some(Duration::from_millis(300))).unwrap();
let r = next.read(&mut b);
assert!(matches!(&r, Err(e) if e.kind() == std::io::ErrorKind::WouldBlock || e.kind() == std::io::ErrorKind::TimedOut), "its slot was given back: {r:?}");
}
/// ★ **A refusal is said once per window** (RA4-13: each refusal was its own eprintln); the
/// lines held back are counted into the next one said.
#[test]
fn the_door_says_a_line_once_per_window_and_counts_the_rest() {
let said = Said::new(Duration::from_secs(60));
let t0 = std::time::Instant::now();
assert_eq!(said.may_say(t0), Some(0), "the first line is said");
assert_eq!(said.may_say(t0 + Duration::from_secs(1)), None);
assert_eq!(said.may_say(t0 + Duration::from_secs(59)), None);
assert_eq!(said.may_say(t0 + Duration::from_secs(60)), Some(2), "the next window says one, with the two held back");
assert_eq!(said.may_say(t0 + Duration::from_secs(61)), None);
assert_eq!(SAY_EVERY, Duration::from_secs(60));
}
/// ★ The door's request is a CONNECT to a named host and a port; anything else is refused
/// by name.
#[test]
fn a_connect_head_names_its_destination_and_anything_else_is_refused() {
assert_eq!(parse_connect("CONNECT objects.example.com:443 HTTP/1.1\r\nHost: objects.example.com:443\r\n\r\n").unwrap(), ("objects.example.com".into(), 443));
assert_eq!(parse_connect("CONNECT [2001:db8::1]:443 HTTP/1.1\r\n\r\n").unwrap(), ("2001:db8::1".into(), 443));
assert!(parse_connect("GET http://x/ HTTP/1.1\r\n\r\n").unwrap_err().contains("not a CONNECT"));
assert!(parse_connect("CONNECT x HTTP/1.1\r\n\r\n").unwrap_err().contains("no port"));
assert!(parse_connect("CONNECT :443 HTTP/1.1\r\n\r\n").unwrap_err().contains("no host"));
assert_eq!(door_url(), "http://10.44.0.1:3128");
}
/// ★ **A way out, never in**: TLS ports only, and a host that resolves into a private,
/// loopback, link-local or shared network (the utility network, the tunnel) is refused.
#[test]
fn the_door_leaves_on_tls_ports_to_outside_addresses_only() {
let out: Vec<SocketAddr> = vec!["192.0.2.10:443".parse().unwrap()];
assert!(admits("objects.example.com", 443, &out).is_ok());
assert!(admits("smtp.example.com", 465, &["192.0.2.10:465".parse().unwrap()]).is_ok());
assert!(admits("x", 22, &out).unwrap_err().contains("[egress-port]"));
assert!(admits("x", 80, &out).unwrap_err().contains("[egress-port]"));
for inner in ["10.6.0.2:443", "10.44.0.2:443", "127.0.0.1:443", "169.254.169.254:443", "100.64.0.1:443", "[::1]:443", "[fd00::1]:443", "[::ffff:10.0.0.1]:443"] {
assert!(admits("x", 443, &[inner.parse().unwrap()]).unwrap_err().contains("[egress-inside]"), "{inner}");
}
assert!(admits("x", 443, &[]).unwrap_err().contains("[egress-unresolved]"));
assert!(on_tunnel("10.44.0.2".parse().unwrap()) && !on_tunnel("10.6.0.2".parse().unwrap()) && !on_tunnel("192.0.2.1".parse().unwrap()));
}
/// ★ **The door carries bytes both ways after `200`** (over loopback, the test's own policy):
/// a peer the policy refuses gets `403`, a refused destination `403`, and nothing is dialled.
#[test]
fn the_door_carries_a_tunnel_and_refuses_what_its_policy_refuses() {
let echo = TcpListener::bind("127.0.0.1:0").unwrap();
let echo_at = echo.local_addr().unwrap();
std::thread::spawn(move || {
for mut c in echo.incoming().flatten() {
let mut b = [0u8; 5];
if c.read_exact(&mut b).is_ok() {
let _ = c.write_all(&b);
}
}
});
let open = Policy { peer: |_| true, admits: |_, _, r| if r.is_empty() { Err("none".into()) } else { Ok(()) } };
let door = serve("127.0.0.1:0".parse().unwrap(), open).unwrap();
let mut c = TcpStream::connect(door).unwrap();
c.write_all(format!("CONNECT 127.0.0.1:{} HTTP/1.1\r\n\r\n", echo_at.port()).as_bytes()).unwrap();
let mut reply = [0u8; 39];
c.read_exact(&mut reply).unwrap();
assert!(String::from_utf8_lossy(&reply).starts_with("HTTP/1.1 200"), "{}", String::from_utf8_lossy(&reply));
c.write_all(b"hello").unwrap();
let mut back = [0u8; 5];
c.read_exact(&mut back).unwrap();
assert_eq!(&back, b"hello");
let shut = Policy { peer: |_| false, admits };
let door = serve("127.0.0.1:0".parse().unwrap(), shut).unwrap();
let mut c = TcpStream::connect(door).unwrap();
let mut reply = String::new();
let _ = c.read_to_string(&mut reply);
assert!(reply.starts_with("HTTP/1.1 403"), "a peer off the tunnel: {reply}");
let door = serve("127.0.0.1:0".parse().unwrap(), Policy { peer: |_| true, admits }).unwrap();
let mut c = TcpStream::connect(door).unwrap();
c.write_all(format!("CONNECT 127.0.0.1:{} HTTP/1.1\r\n\r\n", echo_at.port()).as_bytes()).unwrap();
let mut reply = String::new();
let _ = c.read_to_string(&mut reply);
assert!(reply.starts_with("HTTP/1.1 403"), "the estate's policy refuses loopback and the port: {reply}");
}
/// ★ **A box asks its door by name** (the client half, for a client that has no proxy of its
/// own): the CONNECT head it sends, the door's address from its fed URL, and the door's
/// answer read: `200` is a tunnel, anything else a refusal that names the status.
#[test]
fn a_box_asks_its_door_for_a_named_destination_and_reads_the_answer() {
assert_eq!(connect_head("smtp.example.com", 465), "CONNECT smtp.example.com:465 HTTP/1.1\r\nHost: smtp.example.com:465\r\n\r\n");
assert_eq!(connect_head("2001:db8::1", 587), "CONNECT [2001:db8::1]:587 HTTP/1.1\r\nHost: [2001:db8::1]:587\r\n\r\n");
assert_eq!(parse_connect(&connect_head("smtp.example.com", 587)).unwrap(), ("smtp.example.com".into(), 587), "the door reads what a box asks");
assert_eq!(door_addr(&door_url()).unwrap(), "10.44.0.1:3128".parse::<SocketAddr>().unwrap());
assert_eq!(door_addr("http://10.44.0.1:3128/").unwrap(), "10.44.0.1:3128".parse::<SocketAddr>().unwrap());
assert!(door_addr("socks5://10.44.0.1:1080").unwrap_err().contains("http://"));
assert!(door_addr("http://front:3128").unwrap_err().contains("address"), "a fed door is an address: the box has no DNS");
assert!(established("HTTP/1.1 200 Connection Established\r\n\r\n").is_ok());
let refused = established("HTTP/1.1 403 Forbidden\r\n\r\n").unwrap_err();
assert!(refused.contains("403"), "{refused}");
}
/// ★ **The SMTP gap, closed: a loopback forward through the front's door** (A19 row 13).
/// lettre speaks no proxy, so a box behind its front opens a [`Forward`] to the relay: the
/// mail client dials the forward's loopback address with the RELAY's TLS name, and each of
/// its connections is carried through the door as `CONNECT relay:port`, end to end. A door
/// that refuses is said at [`Forward::open`], by its status, before any mail is written.
#[test]
fn a_forward_carries_a_client_that_speaks_no_proxy_through_the_door() {
let relay = TcpListener::bind("127.0.0.1:0").unwrap();
let relay_at = relay.local_addr().unwrap();
std::thread::spawn(move || {
for mut c in relay.incoming().flatten() {
std::thread::spawn(move || {
let _ = c.write_all(b"220 relay ESMTP\r\n");
let mut b = [0u8; 6];
if c.read_exact(&mut b).is_ok() {
let _ = c.write_all(&b);
}
});
}
});
let open = Policy { peer: |_| true, admits: |_, _, r| if r.is_empty() { Err("none".into()) } else { Ok(()) } };
let door = serve("127.0.0.1:0".parse().unwrap(), open).unwrap();
let door = format!("http://{door}");
let fwd = Forward::open(&door, "127.0.0.1", relay_at.port()).unwrap();
assert!(fwd.addr().ip().is_loopback(), "the forward listens on loopback only");
for round in 0..2 {
let mut c = TcpStream::connect(fwd.addr()).unwrap();
let mut greet = [0u8; 17];
c.read_exact(&mut greet).unwrap();
assert_eq!(&greet, b"220 relay ESMTP\r\n", "round {round}: the relay's own greeting, through the door");
c.write_all(b"EHLO x").unwrap();
let mut back = [0u8; 6];
c.read_exact(&mut back).unwrap();
assert_eq!(&back, b"EHLO x");
}
drop(fwd);
let shut = serve("127.0.0.1:0".parse().unwrap(), Policy { peer: |_| true, admits }).unwrap();
let why = Forward::open(&format!("http://{shut}"), "127.0.0.1", relay_at.port()).err().expect("the estate's policy refuses loopback");
assert!(why.to_string().contains("403"), "{why}");
}
/// The door a box was fed is kept for its clients; a blank one is not one.
#[test]
fn a_fed_door_is_kept_and_a_blank_one_is_not() {
set_proxy(" ");
assert_eq!(proxy(), None);
set_proxy(&door_url());
assert_eq!(proxy(), Some("http://10.44.0.1:3128"));
set_proxy("http://elsewhere:1");
assert_eq!(proxy(), Some("http://10.44.0.1:3128"), "the first wins");
}
}