nordisk-pairing 0.5.27

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

/// **The door, serving** on `bind` (the front's overlay address in an estate), one thread per
/// connection. Returns the bound address. Errors only when it cannot bind.
pub fn serve(bind: SocketAddr, policy: Policy) -> std::io::Result<SocketAddr> {
    let listener = TcpListener::bind(bind)?;
    let at = listener.local_addr()?;
    // A28 M18, stays threads: the door's accept loop (its start failure answered) and one short
    // thread per connection, which ends with it (a connection that cannot get one is closed).
    std::thread::Builder::new().name("egress".into()).spawn(move || {
        for conn in listener.incoming().flatten() {
            let _ = std::thread::Builder::new().name("egress-conn".into()).spawn(move || {
                if let Err(why) = one(conn, policy) {
                    eprintln!("egress: {why}");
                }
            });
        }
    })?;
    Ok(at)
}

fn one(mut conn: TcpStream, policy: Policy) -> 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"));
    }
    conn.set_read_timeout(Some(Duration::from_secs(30))).map_err(|e| e.to_string())?;
    let mut head = Vec::new();
    let mut byte = [0u8; 1];
    while !head.ends_with(b"\r\n\r\n") {
        if head.len() >= MAX_HEAD {
            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"));
        }
        match conn.read(&mut byte) {
            Ok(1) => head.push(byte[0]),
            _ => 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);
    Ok(())
}

/// Copies both ways until either side closes.
fn splice(a: TcpStream, b: TcpStream) {
    let (Ok(a2), Ok(b2)) = (a.try_clone(), b.try_clone()) else { return };
    // A28 M18, stays a thread: one direction of the splice, joined below.
    let up = std::thread::spawn(move || {
        let (mut from, mut to) = (a2, b2);
        let _ = std::io::copy(&mut from, &mut to);
        let _ = to.shutdown(Shutdown::Write);
    });
    let (mut from, mut to) = (b, a);
    let _ = std::io::copy(&mut from, &mut to);
    let _ = to.shutdown(Shutdown::Write);
    let _ = up.join();
}

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));
                    }
                    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 {
    use super::*;

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