nordisk-pairing 0.5.29

//! **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");
    }
}