sn-exit v0: quota-controlled uplink proxy (byte relay, CONNECT, SSRF guard)

server/crates/sn-exit/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "sn-exit"
+description = "SpiderNet exit service: quota-controlled uplink proxy for neighbors over the overlay"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+tokio = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }
+clap = { workspace = true }
+tracing-subscriber = { workspace = true }
+
+[lints]
+workspace = true

server/crates/sn-exit/src/config.rs

@@ -0,0 +1,285 @@
+//! Typed, validated configuration for the exit service.
+
+use std::collections::HashSet;
+use std::net::{IpAddr, SocketAddr};
+use std::time::Duration;
+
+use crate::error::ExitError;
+
+/// Quota: how many bytes a neighbor may push through the exit per window.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct QuotaPolicy {
+    max_bytes: u64,
+    window: Duration,
+}
+
+impl QuotaPolicy {
+    pub fn try_new(max_bytes: u64, window: Duration) -> Result<Self, ExitError> {
+        if max_bytes == 0 {
+            return Err(ExitError::InvalidQuota("max_bytes must be non-zero"));
+        }
+        if window.is_zero() {
+            return Err(ExitError::InvalidQuota("window must be non-zero"));
+        }
+        Ok(Self { max_bytes, window })
+    }
+
+    pub const fn max_bytes(self) -> u64 {
+        self.max_bytes
+    }
+
+    pub const fn window(self) -> Duration {
+        self.window
+    }
+}
+
+/// Which source addresses (neighbors) may use this exit. Empty list =
+/// loopback only (development mode); real deployments always list the
+/// neighbors' overlay addresses explicitly.
+#[derive(Debug, Clone)]
+pub struct SourceAllowlist {
+    allowed: HashSet<IpAddr>,
+    allow_loopback: bool,
+}
+
+impl SourceAllowlist {
+    /// Loopback-only allowlist (development mode).
+    pub fn loopback_only() -> Self {
+        Self {
+            allowed: HashSet::new(),
+            allow_loopback: true,
+        }
+    }
+
+    /// Explicit allowlist; loopback is allowed additionally in dev mode
+    /// (when no explicit entries exist).
+    pub fn new(allowed: Vec<IpAddr>) -> Self {
+        Self {
+            allowed: allowed.into_iter().collect(),
+            allow_loopback: false,
+        }
+    }
+
+    pub fn is_allowed(&self, source: IpAddr) -> bool {
+        source.is_loopback() && self.allow_loopback || self.allowed.contains(&source)
+    }
+
+    pub fn len(&self) -> usize {
+        self.allowed.len()
+    }
+
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.allowed.is_empty()
+    }
+
+    pub fn is_loopback_only(&self) -> bool {
+        self.allow_loopback
+    }
+}
+
+/// Which origins a neighbor may reach through this exit.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OriginPolicy {
+    /// Production: only public internet addresses (see
+    /// [`is_public_origin`]). Protects the exit household's LAN, its
+    /// admin sockets and overlay addresses from neighbor requests.
+    PublicOnly,
+    /// Test-only: any origin, including loopback. Never use in
+    /// production — it removes the SSRF protection entirely.
+    AnyForTesting,
+}
+
+/// Everything the exit needs to run.
+#[derive(Debug, Clone)]
+pub struct ExitConfig {
+    bind: SocketAddr,
+    allowlist: SourceAllowlist,
+    quota: Option<QuotaPolicy>,
+    origin_policy: OriginPolicy,
+}
+
+impl ExitConfig {
+    pub fn new(bind: SocketAddr, allowlist: SourceAllowlist, quota: Option<QuotaPolicy>) -> Self {
+        Self {
+            bind,
+            allowlist,
+            quota,
+            origin_policy: OriginPolicy::PublicOnly,
+        }
+    }
+
+    /// Override the origin policy (`AnyForTesting` only in tests).
+    #[must_use]
+    pub const fn with_origin_policy(mut self, policy: OriginPolicy) -> Self {
+        self.origin_policy = policy;
+        self
+    }
+
+    pub const fn bind(&self) -> SocketAddr {
+        self.bind
+    }
+
+    pub fn allowlist(&self) -> &SourceAllowlist {
+        &self.allowlist
+    }
+
+    pub const fn quota(&self) -> Option<QuotaPolicy> {
+        self.quota
+    }
+
+    pub const fn origin_policy(&self) -> OriginPolicy {
+        self.origin_policy
+    }
+}
+
+/// SSRF protection: an origin is only fetchable if it is *public*
+/// internet. Loopback, private/LAN, link-local, CGNAT, ULA, multicast,
+/// unspecified and Yggdrasil overlay ranges are refused — neighbors must
+/// not be able to reach the exit household's own LAN or admin sockets,
+/// and the exit is for internet egress, not overlay-internal traffic.
+pub fn is_public_origin(ip: IpAddr) -> bool {
+    match ip {
+        IpAddr::V4(v4) => {
+            !(v4.is_loopback()
+                || v4.is_private()
+                || v4.is_link_local()
+                || v4.is_unspecified()
+                || v4.is_broadcast()
+                || v4.is_multicast()
+                || v4.is_documentation()
+                // 100.64/10 CGNAT (`is_shared()` is still unstable)
+                || (v4.octets()[0] == 100 && (v4.octets()[1] & 0xc0) == 64)
+                || v4.octets()[0] == 0) // 0.0.0.0/8 "this network"
+        }
+        IpAddr::V6(v6) => {
+            let segments = v6.segments();
+            !(v6.is_loopback()
+                || v6.is_unspecified()
+                || v6.is_multicast()
+                // fe80::/10 link-local
+                || (segments[0] & 0xffc0) == 0xfe80
+                // fc00::/7 unique-local (also would cover Yggdrasil's
+                // 200::/7? No: 200::/7 is 0x0200-0x03ff, fc00::/7 is
+                // 0xfc00-0xfdff — distinct. Checked separately below.)
+                || (segments[0] & 0xfe00) == 0xfc00
+                // 200::/7 Yggdrasil overlay (0x0200-0x03ff)
+                || (segments[0] & 0xfe00) == 0x0200
+                // ::ffff:x.y.z.w IPv4-mapped — judge by the v4 part
+                || (segments[0] == 0 && segments[1] == 0 && segments[2] == 0 && segments[3] == 0 && segments[4] == 0 && segments[5] == 0xffff && {
+                    let mapped = IpAddr::V4(std::net::Ipv4Addr::new(
+                        (segments[6] >> 8) as u8,
+                        (segments[6] & 0xff) as u8,
+                        (segments[7] >> 8) as u8,
+                        (segments[7] & 0xff) as u8,
+                    ));
+                    !is_public_origin(mapped)
+                }))
+        }
+    }
+}
+
+/// Parse an address from CLI (`--bind`, `--allow`). Returns a typed error
+/// instead of panicking on junk input.
+pub fn parse_bind(input: &str) -> Result<SocketAddr, ExitError> {
+    input
+        .parse()
+        .map_err(|_| ExitError::InvalidBind(input.to_string()))
+}
+
+pub fn parse_source(input: &str) -> Result<IpAddr, ExitError> {
+    input
+        .parse()
+        .map_err(|_| ExitError::InvalidSource(input.to_string()))
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used)]
+    use super::*;
+
+    #[test]
+    fn quota_rejects_zero() {
+        assert!(QuotaPolicy::try_new(0, Duration::from_secs(60)).is_err());
+        assert!(QuotaPolicy::try_new(100, Duration::ZERO).is_err());
+        assert!(QuotaPolicy::try_new(100, Duration::from_secs(60)).is_ok());
+    }
+
+    #[test]
+    fn allowlist_loopback_default_and_explicit_entries() {
+        let dev = SourceAllowlist::loopback_only();
+        assert!(dev.is_allowed(std::net::IpAddr::from([127, 0, 0, 1])));
+        assert!(!dev.is_allowed(std::net::IpAddr::from([192, 168, 1, 2])));
+
+        let explicit = SourceAllowlist::new(vec![std::net::IpAddr::from([
+            0x200u16, 1, 0, 0, 0, 0, 0, 1,
+        ])]);
+        assert!(!explicit.is_allowed(std::net::IpAddr::from([127, 0, 0, 1])));
+        assert!(explicit.is_allowed(std::net::IpAddr::from([0x200, 1, 0, 0, 0, 0, 0, 1])));
+    }
+
+    #[test]
+    fn public_origin_rejects_private_and_local_targets() {
+        // Public internet: allowed.
+        assert!(is_public_origin(std::net::IpAddr::from([93, 184, 216, 34])));
+        assert!(is_public_origin(
+            "2606:2800:220:1:248:1893:25c8:1946"
+                .parse::<std::net::Ipv6Addr>()
+                .unwrap()
+                .into()
+        ));
+
+        // Refused: loopback, RFC1918, link-local, CGNAT, docs, multicast.
+        for bad_v4 in [
+            [127, 0, 0, 1],
+            [10, 0, 0, 5],
+            [172, 16, 3, 4],
+            [192, 168, 1, 1],
+            [169, 254, 9, 9],
+            [100, 64, 0, 1],
+            [192, 0, 2, 55],
+            [224, 0, 0, 1],
+            [0, 0, 0, 0],
+        ] {
+            assert!(
+                !is_public_origin(std::net::IpAddr::from(bad_v4)),
+                "v4 {bad_v4:?} must be refused"
+            );
+        }
+
+        // Refused: v6 loopback, link-local, ULA, yggdrasil 200::/7,
+        // IPv4-mapped loopback.
+        for bad_v6 in [
+            "::1",
+            "fe80::1",
+            "fc00::1",
+            "200:1234::1",
+            "::ffff:127.0.0.1",
+        ] {
+            let v6 = bad_v6.parse::<std::net::Ipv6Addr>().unwrap();
+            assert!(
+                !is_public_origin(std::net::IpAddr::V6(v6)),
+                "v6 {bad_v6} must be refused"
+            );
+        }
+    }
+
+    #[test]
+    fn parse_bind_and_source() {
+        assert_eq!(
+            parse_bind("[::1]:8080").unwrap(),
+            SocketAddr::new(std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), 8080)
+        );
+        assert_eq!(
+            parse_bind("127.0.0.1:8080").unwrap(),
+            SocketAddr::from(([127, 0, 0, 1], 8080))
+        );
+        assert!(parse_bind("nope").is_err());
+
+        assert_eq!(
+            parse_source("200:1:2:3:4:5:6:7").unwrap(),
+            std::net::IpAddr::V6(std::net::Ipv6Addr::new(0x200, 1, 2, 3, 4, 5, 6, 7))
+        );
+        assert!(parse_source("banana").is_err());
+    }
+}

server/crates/sn-exit/src/error.rs

@@ -0,0 +1,19 @@
+//! Error type for all sn-exit operations.
+
+#[derive(Debug, thiserror::Error)]
+pub enum ExitError {
+    #[error("io error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("invalid bind address `{0}`")]
+    InvalidBind(String),
+
+    #[error("invalid allowed source address `{0}`")]
+    InvalidSource(String),
+
+    #[error("invalid quota: {0}")]
+    InvalidQuota(&'static str),
+
+    #[error("protocol violation: {0}")]
+    Protocol(&'static str),
+}

server/crates/sn-exit/src/lib.rs

@@ -0,0 +1,40 @@
+//! sn-exit — the `SpiderNet` exit service.
+//!
+//! An exit is one household's opt-in offer to its neighbors: this binary
+//! listens on the household's Yggdrasil overlay address and forwards
+//! neighbor requests through the household's own internet uplink, under
+//! a per-neighbor quota.
+//!
+//! Design (v0):
+//!
+//! - **Transparent byte relay.** CONNECT tunnels are relayed byte for
+//!   byte (TLS stays end-to-end between neighbor and origin — the exit
+//!   never sees plaintext HTTPS), plain proxy requests are relayed the
+//!   same way. Range headers and retries keep working because headers
+//!   are never rewritten; segmentation stays a client-side concern
+//!   (`sn-fetch`).
+//! - **SSRF protection.** Origins that resolve to loopback, private,
+//!   link-local, ULA or Yggdrasil ranges are refused — a neighbor must
+//!   not be able to reach the exit household's LAN, its admin sockets
+//!   or other overlay nodes through this service.
+//! - **Quota.** Per source address, per fixed window. Enforcement
+//!   happens when a connection starts; the traffic a connection actually
+//!   carries is accounted when it ends. A single very large connection
+//!   can overshoot the quota — documented v0 limitation.
+//!
+//! Sharing is opt-in by running this binary; stopping it is the
+//! kill-switch. Receiving without offering an exit is the explicitly
+//! allowed leech mode — no punishment, see `[[Ethos]]` in the docs.
+//!
+//! `SpiderNet` software carries no warranty and accepts no responsibility
+//! for its usage or for transported content (AGPL-3.0, sections 15/16).
+
+#![forbid(unsafe_code)]
+#![warn(clippy::pedantic)]
+// PoC scope: doc-annotation noise, not correctness. Revisit at release.
+#![allow(clippy::must_use_candidate, clippy::missing_errors_doc)]
+
+pub mod config;
+pub mod error;
+pub mod parse;
+pub mod proxy;

server/crates/sn-exit/src/main.rs

@@ -0,0 +1,107 @@
+//! CLI entry point for sn-exit.
+//!
+//! KEINE GARANTIE — this software is provided "as is" without warranty of
+//! any kind (AGPL-3.0, sections 15/16). Users are solely responsible for
+//! what they transmit or receive with it.
+
+use std::time::Duration;
+
+use clap::Parser;
+use sn_exit::config::{self, ExitConfig, QuotaPolicy, SourceAllowlist};
+
+/// Per-neighbor quota is disabled when 0 is passed (no limit).
+const QUOTA_DISABLED: u64 = 0;
+
+#[derive(Debug, Parser)]
+#[command(
+    name = "sn-exit",
+    about = "SpiderNet exit service: quota-controlled uplink proxy for neighbors"
+)]
+struct Args {
+    /// Address to listen on — ideally the household's Yggdrasil address,
+    /// e.g. `[200:1234::1]:8080`.
+    #[arg(long, default_value = "127.0.0.1:8080")]
+    bind: String,
+
+    /// Neighbor addresses allowed to use this exit (repeatable, e.g.
+    /// `200:5678::2`). Without any `--allow`, only loopback clients are
+    /// accepted (development mode).
+    #[arg(long = "allow", value_delimiter = ',')]
+    allow: Vec<String>,
+
+    /// Quota per neighbor per window, in MiB. `0` disables the quota.
+    #[arg(long, default_value_t = 1024)]
+    quota_mib: u64,
+
+    /// Quota window in hours (fixed window per neighbor).
+    #[arg(long, default_value_t = 24)]
+    window_hours: u64,
+}
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+
+    if let Err(report) = run().await {
+        eprintln!("NO WARRANTY — SpiderNet sn-exit (AGPL-3.0, no liability)");
+        eprintln!("error: {report}");
+        std::process::exit(1);
+    }
+}
+
+async fn run() -> Result<(), String> {
+    let args = Args::parse();
+
+    let bind = config::parse_bind(&args.bind).map_err(|err| err.to_string())?;
+    let allowlist = if args.allow.is_empty() {
+        SourceAllowlist::loopback_only()
+    } else {
+        let sources = args
+            .allow
+            .iter()
+            .map(|addr| config::parse_source(addr).map_err(|err| err.to_string()))
+            .collect::<Result<Vec<_>, String>>()?;
+        SourceAllowlist::new(sources)
+    };
+    let quota = if args.quota_mib == QUOTA_DISABLED {
+        None
+    } else {
+        let max_bytes = args.quota_mib.saturating_mul(1024 * 1024);
+        Some(
+            QuotaPolicy::try_new(max_bytes, Duration::from_secs(args.window_hours * 3600))
+                .map_err(|err| err.to_string())?,
+        )
+    };
+
+    let config = ExitConfig::new(bind, allowlist, quota);
+    let listener = tokio::net::TcpListener::bind(config.bind())
+        .await
+        .map_err(|err| format!("cannot bind {}: {err}", config.bind()))?;
+
+    println!("NO WARRANTY — SpiderNet sn-exit (AGPL-3.0, no liability)");
+    println!();
+    if config.allowlist().is_loopback_only() {
+        println!("DEV MODE: no --allow given — only loopback clients are accepted.");
+    }
+    match config.quota() {
+        Some(policy) => println!(
+            "quota:    {} MiB per {}h per neighbor (enforced at connection start)",
+            policy.max_bytes() / (1024 * 1024),
+            policy.window().as_secs() / 3600,
+        ),
+        None => println!("quota:    disabled — neighbors may use unlimited uplink traffic"),
+    }
+    println!(
+        "sharing:  this household now offers its uplink (stop sn-exit to revoke = kill-switch)"
+    );
+    println!("listen:   {}", config.bind());
+
+    sn_exit::proxy::serve(config, listener)
+        .await
+        .map_err(|err| err.to_string())
+}

server/crates/sn-exit/src/parse.rs

@@ -0,0 +1,167 @@
+//! Minimal, strict HTTP/1.1 proxy-head parsing.
+//!
+//! The exit is a byte relay, so it only needs to understand the request
+//! line (method + target) to find the origin — everything else is
+//! forwarded verbatim inside the pipe. Strictness beats flexibility
+//! here: anything that is not a clean proxy request line is refused.
+
+use crate::error::ExitError;
+
+/// How the client addressed the origin.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RequestTarget {
+    /// CONNECT authority form: `host:port` (TLS and any other TCP).
+    Tunnel,
+    /// Absolute URI: `http://host[:port]/…` (plain HTTP via proxy).
+    Forward,
+}
+
+/// One parsed proxy request head.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RequestHead {
+    pub target: RequestTarget,
+    pub host: String,
+    pub port: u16,
+}
+
+/// Largest accepted request head; anything above is refused instead of
+/// buffered without bound.
+pub const MAX_HEAD_BYTES: usize = 16 * 1024;
+
+/// Parse the request line out of a raw head buffer. The buffer may
+/// include the trailing `\r\n\r\n`; only the first line is examined.
+///
+/// `CONNECT host:port HTTP/1.1` → tunnel to `host:port`.
+/// `GET http://host[:port]/path HTTP/1.1` → forward to `host:port`.
+pub fn parse_request_head(head: &[u8]) -> Result<RequestHead, ExitError> {
+    let head_text = std::str::from_utf8(head)
+        .map_err(|_| ExitError::Protocol("request head is not valid utf-8"))?;
+    let request_line = head_text
+        .lines()
+        .next()
+        .ok_or(ExitError::Protocol("empty request head"))?;
+
+    let mut parts = request_line.split_whitespace();
+    let method = parts.next().unwrap_or_default();
+    let raw_target = parts.next().unwrap_or_default();
+
+    match method {
+        "CONNECT" => {
+            let (host, port) = parse_authority(raw_target)?;
+            Ok(RequestHead {
+                target: RequestTarget::Tunnel,
+                host,
+                port,
+            })
+        }
+        "GET" | "HEAD" | "POST" | "PUT" | "DELETE" | "OPTIONS" => {
+            let rest = raw_target.strip_prefix("http://").ok_or(ExitError::Protocol(
+                "non-CONNECT requests must use absolute http:// URIs (proxy-capable client required)",
+            ))?;
+            let authority = rest.split('/').next().unwrap_or_default();
+            let (host, port) = parse_authority(authority)?;
+            Ok(RequestHead {
+                target: RequestTarget::Forward,
+                host,
+                port,
+            })
+        }
+        _ => Err(ExitError::Protocol("unsupported method")),
+    }
+}
+
+/// Parse `host:port` or `[ipv6]:port`. Default port 80 when absent.
+fn parse_authority(authority: &str) -> Result<(String, u16), ExitError> {
+    let authority = authority.trim();
+    if authority.is_empty() {
+        return Err(ExitError::Protocol("empty authority"));
+    }
+
+    // [ipv6]:port
+    if let Some(rest) = authority.strip_prefix('[') {
+        let close = rest
+            .find(']')
+            .ok_or(ExitError::Protocol("unterminated [ipv6] authority"))?;
+        let host = &rest[..close];
+        host.parse::<std::net::Ipv6Addr>()
+            .map_err(|_| ExitError::Protocol("bad ipv6 authority"))?;
+        let port = rest[close + 1..].strip_prefix(':').map_or(Ok(80), |p| {
+            p.parse::<u16>()
+                .map_err(|_| ExitError::Protocol("bad port"))
+        })?;
+        return Ok((host.to_string(), port));
+    }
+
+    // No brackets — IPv4 or hostname. rsplit, not split: only the last
+    // `:` separates the port.
+    let Some((host, port)) = authority.rsplit_once(':') else {
+        if authority.contains(':') {
+            return Err(ExitError::Protocol("bare ipv6 needs brackets and a port"));
+        }
+        return Ok((authority.to_string(), 80));
+    };
+    let port: u16 = port.parse().map_err(|_| ExitError::Protocol("bad port"))?;
+    if host.is_empty() {
+        return Err(ExitError::Protocol("empty host"));
+    }
+    Ok((host.to_string(), port))
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used)]
+    use super::*;
+
+    fn head(line: &str) -> Vec<u8> {
+        format!("{line}\r\nHost: example.com\r\n\r\n").into_bytes()
+    }
+
+    #[test]
+    fn parses_connect_authority() {
+        let parsed = parse_request_head(&head("CONNECT example.com:443 HTTP/1.1")).unwrap();
+        assert_eq!(
+            parsed,
+            RequestHead {
+                target: RequestTarget::Tunnel,
+                host: "example.com".to_string(),
+                port: 443,
+            }
+        );
+    }
+
+    #[test]
+    fn parses_connect_ipv6_authority() {
+        let parsed = parse_request_head(&head("CONNECT [200:1234::1]:8080 HTTP/1.1")).unwrap();
+        assert_eq!(parsed.host, "200:1234::1");
+        assert_eq!(parsed.port, 8080);
+    }
+
+    #[test]
+    fn parses_absolute_uri_forward() {
+        let parsed =
+            parse_request_head(&head("GET http://example.com/dir/file.bin HTTP/1.1")).unwrap();
+        assert_eq!(
+            parsed,
+            RequestHead {
+                target: RequestTarget::Forward,
+                host: "example.com".to_string(),
+                port: 80,
+            }
+        );
+
+        let parsed = parse_request_head(&head("GET http://example.com:8080/a HTTP/1.1")).unwrap();
+        assert_eq!(parsed.host, "example.com");
+        assert_eq!(parsed.port, 8080);
+    }
+
+    #[test]
+    fn rejects_origin_form_and_junk() {
+        // Origin-form (no absolute URI) is unusable for a proxy.
+        assert!(parse_request_head(&head("GET /file.bin HTTP/1.1")).is_err());
+        assert!(parse_request_head(&head("GET https://example.com/x HTTP/1.1")).is_err());
+        assert!(parse_request_head(&head("FTP http://example.com/ HTTP/1.1")).is_err());
+        assert!(parse_request_head(&head("CONNECT :443 HTTP/1.1")).is_err());
+        assert!(parse_request_head(&head("")).is_err());
+        assert!(parse_request_head(&[0xff, 0xfe, 0xfd]).is_err()); // not utf-8
+    }
+}

server/crates/sn-exit/src/proxy.rs

@@ -0,0 +1,339 @@
+//! The exit server: accept neighbor connections, admit them (allowlist,
+//! quota), and relay bytes between neighbor and origin.
+//!
+//! Relay is verbatim: the exit never rewrites or inspects payload bytes.
+//! TLS stays end-to-end between the neighbor's client and the origin —
+//! the exit only sees which host is contacted and how many bytes flow.
+
+use std::collections::HashMap;
+use std::net::IpAddr;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+
+use crate::config::{is_public_origin, ExitConfig, OriginPolicy, QuotaPolicy};
+use crate::error::ExitError;
+use crate::parse::{self, RequestTarget, MAX_HEAD_BYTES};
+
+/// Time budget for reading the request head and connecting to the origin.
+const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(15);
+
+const RESP_FORBIDDEN: &[u8] =
+    b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
+const RESP_BAD_REQUEST: &[u8] =
+    b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
+const RESP_TUNNEL_OK: &[u8] = b"HTTP/1.1 200 Connection established\r\n\r\n";
+
+/// Per-neighbor quota ledger: fixed windows anchored at the neighbor's
+/// first use, accounting applied after each finished connection.
+///
+/// v0 limitation: enforcement happens when a connection starts; the
+/// bytes a connection carries are accounted when it ends. One very large
+/// connection can therefore overshoot the quota.
+#[derive(Debug, Default)]
+pub struct QuotaLedger {
+    clients: Mutex<HashMap<IpAddr, ClientWindow>>,
+}
+
+#[derive(Debug)]
+struct ClientWindow {
+    window_started: Instant,
+    used_bytes: u64,
+}
+
+/// Admission decision for one connection attempt.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AdmitDecision {
+    Allowed,
+    /// The neighbor's quota is exhausted; retry after this many seconds.
+    QuotaExhausted {
+        retry_after_secs: u64,
+    },
+}
+
+impl QuotaLedger {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Check whether a connection from `source` may start now.
+    pub fn admit(&self, source: IpAddr, quota: Option<QuotaPolicy>) -> AdmitDecision {
+        let Some(policy) = quota else {
+            return AdmitDecision::Allowed;
+        };
+        let mut clients = Self::lock_recovered(&self.clients);
+        let window = roll_window(clients.entry(source), policy, Instant::now());
+        if window.used_bytes >= policy.max_bytes() {
+            let elapsed = window.window_started.elapsed();
+            let remaining = policy
+                .window()
+                .checked_sub(elapsed)
+                .unwrap_or(Duration::ZERO);
+            AdmitDecision::QuotaExhausted {
+                retry_after_secs: remaining.as_secs().max(1),
+            }
+        } else {
+            AdmitDecision::Allowed
+        }
+    }
+
+    /// Account the traffic of one finished connection.
+    pub fn record(&self, source: IpAddr, bytes: u64, quota: Option<QuotaPolicy>) {
+        let Some(policy) = quota else {
+            return;
+        };
+        let mut clients = Self::lock_recovered(&self.clients);
+        let window = roll_window(clients.entry(source), policy, Instant::now());
+        window.used_bytes = window.used_bytes.saturating_add(bytes);
+    }
+
+    /// Total accounted bytes for one neighbor (visible in metrics later).
+    pub fn used_bytes(&self, source: IpAddr) -> u64 {
+        Self::lock_recovered(&self.clients)
+            .get(&source)
+            .map_or(0, |w| w.used_bytes)
+    }
+
+    /// Lock without panicking on a poisoned mutex: the ledger is
+    /// self-healing (windows roll over), so recovering the guard is
+    /// always safe here.
+    fn lock_recovered(
+        clients: &Mutex<HashMap<IpAddr, ClientWindow>>,
+    ) -> std::sync::MutexGuard<'_, HashMap<IpAddr, ClientWindow>> {
+        clients
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+}
+
+/// Get the client's window, rolling it over when the policy's window has
+/// elapsed since its start.
+fn roll_window(
+    entry: std::collections::hash_map::Entry<'_, IpAddr, ClientWindow>,
+    policy: QuotaPolicy,
+    now: Instant,
+) -> &mut ClientWindow {
+    let window = entry.or_insert_with(|| ClientWindow {
+        window_started: now,
+        used_bytes: 0,
+    });
+    if now.duration_since(window.window_started) >= policy.window() {
+        window.window_started = now;
+        window.used_bytes = 0;
+    }
+    window
+}
+
+/// Accept loop. Runs until the listener errors fatally.
+pub async fn serve(config: ExitConfig, listener: TcpListener) -> Result<(), ExitError> {
+    let config = Arc::new(config);
+    let ledger = Arc::new(QuotaLedger::new());
+    loop {
+        let (stream, peer) = listener.accept().await?;
+        let config = Arc::clone(&config);
+        let ledger = Arc::clone(&ledger);
+        tokio::spawn(async move {
+            if let Err(error) = handle_connection(stream, peer.ip(), config, ledger).await {
+                tracing::debug!(%peer, "connection ended: {error}");
+            }
+        });
+    }
+}
+
+async fn handle_connection(
+    mut neighbor: TcpStream,
+    source: IpAddr,
+    config: Arc<ExitConfig>,
+    ledger: Arc<QuotaLedger>,
+) -> Result<(), ExitError> {
+    // 1. Read the request head (bounded, with timeout).
+    let raw_head = read_head(&mut neighbor).await?;
+
+    // 2. Admission: allowlist, then quota.
+    if !config.allowlist().is_allowed(source) {
+        neighbor.write_all(RESP_FORBIDDEN).await.ok();
+        tracing::info!(%source, "refused: source not allowed");
+        return Ok(());
+    }
+    if let AdmitDecision::QuotaExhausted { retry_after_secs } = ledger.admit(source, config.quota())
+    {
+        let body = format!(
+            "HTTP/1.1 503 Service Unavailable\r\nRetry-After: {retry_after_secs}\r\n\
+             Content-Length: 0\r\nConnection: close\r\n\r\n"
+        );
+        neighbor.write_all(body.as_bytes()).await.ok();
+        tracing::info!(%source, "refused: quota exhausted, retry in {retry_after_secs}s");
+        return Ok(());
+    }
+
+    // 3. Parse and verify the origin is public (SSRF protection).
+    let Ok(head) = parse::parse_request_head(&raw_head) else {
+        neighbor.write_all(RESP_BAD_REQUEST).await.ok();
+        return Err(ExitError::Protocol("unparseable request head"));
+    };
+    ensure_origin_is_public(&head, config.origin_policy()).await?;
+
+    // 4. Ack tunnels, then relay — verbatim, both directions.
+    if head.target == RequestTarget::Tunnel {
+        neighbor.write_all(RESP_TUNNEL_OK).await?;
+    }
+
+    let origin = connect_origin(&head).await?;
+
+    // Counted relay: byte counters live in the wrappers, so traffic is
+    // accounted even when the connection errors or is aborted midway.
+    let mut neighbor_counted = CountingIo::new(neighbor);
+    let mut origin_counted = CountingIo::new(origin);
+    if head.target == RequestTarget::Forward {
+        // The neighbor's head is forwarded as the first bytes of the
+        // pipe; origin servers accept absolute-URI request targets.
+        origin_counted.write_all(&raw_head).await?;
+    }
+
+    let relay = tokio::io::copy_bidirectional(&mut neighbor_counted, &mut origin_counted).await;
+    // Uplink traffic = what the household's ISP sees: bytes to the
+    // origin (egress) + bytes to the neighbor (ingress from origin).
+    let up = origin_counted.written_bytes();
+    let down = neighbor_counted.written_bytes();
+    ledger.record(source, up + down, config.quota());
+    relay?;
+    tracing::debug!(
+        %source,
+        "target={:?} origin={}:{} closed ({up}+{down} bytes)",
+        head.target,
+        head.host,
+        head.port
+    );
+    Ok(())
+}
+
+/// Stream wrapper that counts all bytes read from and written to the
+/// inner stream, so quota accounting does not depend on the relay's
+/// outcome.
+struct CountingIo<T> {
+    inner: T,
+    read: std::sync::atomic::AtomicU64,
+    written: std::sync::atomic::AtomicU64,
+}
+
+impl<T> CountingIo<T> {
+    fn new(inner: T) -> Self {
+        Self {
+            inner,
+            read: std::sync::atomic::AtomicU64::new(0),
+            written: std::sync::atomic::AtomicU64::new(0),
+        }
+    }
+
+    fn written_bytes(&self) -> u64 {
+        self.written.load(std::sync::atomic::Ordering::Relaxed)
+    }
+}
+
+impl<T: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for CountingIo<T> {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        let before = buf.filled().len();
+        let result = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
+        if matches!(result, std::task::Poll::Ready(Ok(()))) {
+            let after = buf.filled().len();
+            self.read.fetch_add(
+                (after - before) as u64,
+                std::sync::atomic::Ordering::Relaxed,
+            );
+        }
+        result
+    }
+}
+
+impl<T: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for CountingIo<T> {
+    fn poll_write(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        let result = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf);
+        if let std::task::Poll::Ready(Ok(n)) = result {
+            self.written
+                .fetch_add(n as u64, std::sync::atomic::Ordering::Relaxed);
+        }
+        result
+    }
+
+    fn poll_flush(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        std::pin::Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+}
+
+/// Read up to [`MAX_HEAD_BYTES`] bytes until the head terminator
+/// `\r\n\r\n`. Returns the raw head *including* the terminator.
+async fn read_head(stream: &mut TcpStream) -> Result<Vec<u8>, ExitError> {
+    let mut head = Vec::new();
+    let mut byte = [0u8; 1];
+    let deadline = tokio::time::Instant::now() + HANDSHAKE_TIMEOUT;
+    loop {
+        if head.len() >= MAX_HEAD_BYTES {
+            return Err(ExitError::Protocol("request head too large"));
+        }
+        match tokio::time::timeout_at(deadline, stream.read(&mut byte)).await {
+            Err(_) => return Err(ExitError::Io(std::io::ErrorKind::TimedOut.into())),
+            Ok(Ok(0)) => return Err(ExitError::Protocol("connection closed before head ended")),
+            Ok(Ok(_)) => {
+                head.push(byte[0]);
+                if head.ends_with(b"\r\n\r\n") {
+                    return Ok(head);
+                }
+            }
+            Ok(Err(error)) => return Err(ExitError::Io(error)),
+        }
+    }
+}
+
+/// DNS resolution happens here so the blocklist sees the *resolved* IPs,
+/// not just the hostname the neighbor typed.
+async fn ensure_origin_is_public(
+    head: &parse::RequestHead,
+    policy: OriginPolicy,
+) -> Result<(), ExitError> {
+    if policy == OriginPolicy::AnyForTesting {
+        return Ok(());
+    }
+    let resolved: Vec<std::net::SocketAddr> =
+        tokio::net::lookup_host((head.host.as_str(), head.port))
+            .await?
+            .collect();
+    if resolved.is_empty() {
+        return Err(ExitError::Protocol("origin could not be resolved"));
+    }
+    for addr in resolved {
+        if !is_public_origin(addr.ip()) {
+            return Err(ExitError::Protocol("origin is not a public address"));
+        }
+    }
+    Ok(())
+}
+
+async fn connect_origin(head: &parse::RequestHead) -> Result<TcpStream, ExitError> {
+    let stream = tokio::time::timeout(
+        HANDSHAKE_TIMEOUT,
+        TcpStream::connect((head.host.as_str(), head.port)),
+    )
+    .await
+    .map_err(|_| ExitError::Io(std::io::ErrorKind::TimedOut.into()))??;
+    Ok(stream)
+}

server/crates/sn-exit/tests/proxy.rs

@@ -0,0 +1,276 @@
+//! Integration tests against a running exit server (loopback): plain
+//! HTTP forwarding, CONNECT tunneling, quota enforcement and the source
+//! allowlist.
+
+#![forbid(unsafe_code)]
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+use std::net::{IpAddr, Ipv6Addr, SocketAddr};
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+
+use sn_exit::config::{ExitConfig, QuotaPolicy, SourceAllowlist};
+use sn_exit::proxy::serve;
+
+/// Spawn the exit server on a free loopback port; returns its address.
+async fn spawn_exit(config: ExitConfig) -> SocketAddr {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        let _ = serve(config, listener).await;
+    });
+    addr
+}
+
+/// Origin server that supports ranges: answers `GET http://origin/…`
+/// absolute-URI requests with (parts of) a fixed body.
+async fn spawn_origin(data: Arc<Vec<u8>>) -> SocketAddr {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let Ok((mut socket, _)) = listener.accept().await else {
+                break;
+            };
+            let data = Arc::clone(&data);
+            tokio::spawn(async move {
+                let mut head = Vec::new();
+                let mut byte = [0u8; 1];
+                while !head.ends_with(b"\r\n\r\n") {
+                    if socket.read(&mut byte).await.unwrap_or(0) == 0 {
+                        return;
+                    }
+                    head.push(byte[0]);
+                }
+                let head_text = String::from_utf8_lossy(&head);
+                let range = head_text.lines().find_map(|line| {
+                    let (name, value) = line.split_once(':')?;
+                    if !name.eq_ignore_ascii_case("range") {
+                        return None;
+                    }
+                    let value = value.trim().strip_prefix("bytes=")?;
+                    let (start, end) = value.trim().split_once('-')?;
+                    Some((start.parse::<u64>().ok()?, end.parse::<u64>().ok()?))
+                });
+
+                let (start, end_inclusive, status) = match range {
+                    Some((start, end)) => (start, end, "206 Partial Content"),
+                    None => (0, data.len() as u64 - 1, "200 OK"),
+                };
+                let slice = &data[start as usize..=(end_inclusive as usize).min(data.len() - 1)];
+                let response = format!(
+                    "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+                    slice.len()
+                );
+                socket.write_all(response.as_bytes()).await.ok();
+                socket.write_all(slice).await.ok();
+                socket.shutdown().await.ok();
+            });
+        }
+    });
+    addr
+}
+
+/// Echo TCP server for the CONNECT tunnel test (stands in for a TLS
+/// origin from the relay's perspective: opaque bytes both ways).
+async fn spawn_echo_server() -> SocketAddr {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let Ok((mut socket, _)) = listener.accept().await else {
+                break;
+            };
+            tokio::spawn(async move {
+                let mut buffer = [0u8; 512];
+                loop {
+                    match socket.read(&mut buffer).await {
+                        Ok(0) | Err(_) => break,
+                        Ok(n) => {
+                            if socket.write_all(&buffer[..n]).await.is_err() {
+                                break;
+                            }
+                        }
+                    }
+                }
+            });
+        }
+    });
+    addr
+}
+
+async fn read_response_head(stream: &mut TcpStream) -> String {
+    let mut head = Vec::new();
+    let mut byte = [0u8; 1];
+    while !head.ends_with(b"\r\n\r\n") {
+        let n = stream.read(&mut byte).await.unwrap_or(0);
+        assert!(n > 0, "connection closed before response head ended");
+        head.push(byte[0]);
+    }
+    String::from_utf8(head).unwrap()
+}
+
+fn dev_config(quota: Option<QuotaPolicy>) -> ExitConfig {
+    ExitConfig::new(
+        SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0),
+        SourceAllowlist::loopback_only(),
+        quota,
+    )
+    .with_origin_policy(sn_exit::config::OriginPolicy::AnyForTesting)
+}
+
+#[tokio::test]
+async fn forward_request_relays_absolute_uri_get() {
+    let data: Vec<u8> = (0..32_000u32).map(|i| (i % 253) as u8).collect();
+    let origin = spawn_origin(Arc::new(data.clone())).await;
+    let exit = spawn_exit(dev_config(None)).await;
+
+    let mut neighbor = TcpStream::connect(exit).await.unwrap();
+    let request = format!("GET http://{origin}/file.bin HTTP/1.1\r\nHost: origin\r\n\r\n");
+    neighbor.write_all(request.as_bytes()).await.unwrap();
+
+    let response_head = read_response_head(&mut neighbor).await;
+    assert!(
+        response_head.starts_with("HTTP/1.1 200 OK"),
+        "{response_head}"
+    );
+
+    let mut body = Vec::new();
+    neighbor.read_to_end(&mut body).await.unwrap();
+    assert_eq!(body, data);
+}
+
+#[tokio::test]
+async fn forward_request_passes_range_headers_through() {
+    let data: Vec<u8> = (0..16_000u32).map(|i| (i % 241) as u8).collect();
+    let origin = spawn_origin(Arc::new(data.clone())).await;
+    let exit = spawn_exit(dev_config(None)).await;
+
+    let mut neighbor = TcpStream::connect(exit).await.unwrap();
+    let request = format!("GET http://{origin}/file.bin HTTP/1.1\r\nRange: bytes=100-199\r\n\r\n");
+    neighbor.write_all(request.as_bytes()).await.unwrap();
+
+    let response_head = read_response_head(&mut neighbor).await;
+    assert!(
+        response_head.starts_with("HTTP/1.1 206 Partial Content"),
+        "{response_head}"
+    );
+    let mut body = Vec::new();
+    neighbor.read_to_end(&mut body).await.unwrap();
+    assert_eq!(&body, &data[100..=199]);
+}
+
+#[tokio::test]
+async fn connect_tunnel_relays_bytes_both_ways() {
+    let echo = spawn_echo_server().await;
+    let exit = spawn_exit(dev_config(None)).await;
+
+    let mut neighbor = TcpStream::connect(exit).await.unwrap();
+    let request = format!("CONNECT {echo} HTTP/1.1\r\nHost: {echo}\r\n\r\n");
+    neighbor.write_all(request.as_bytes()).await.unwrap();
+
+    let response_head = read_response_head(&mut neighbor).await;
+    assert!(
+        response_head.starts_with("HTTP/1.1 200"),
+        "tunnel must be established: {response_head}"
+    );
+
+    // The tunnel is open: echo some opaque "TLS-like" bytes through it.
+    let payload: Vec<u8> = (0..=255u8).cycle().take(700).collect();
+    neighbor.write_all(&payload).await.unwrap();
+    let mut echoed = vec![0u8; payload.len()];
+    neighbor.read_exact(&mut echoed).await.unwrap();
+    assert_eq!(echoed, payload);
+    let _ = neighbor.shutdown().await;
+}
+
+#[tokio::test]
+async fn quota_exhaustion_returns_503_with_retry_after() {
+    let data = Arc::new(vec![42u8; 8_000]);
+    let origin = spawn_origin(Arc::clone(&data)).await;
+    // 1 KiB quota, 1 h window: the first connection (~8 KB) exhausts it.
+    let quota = QuotaPolicy::try_new(1024, Duration::from_secs(3600)).unwrap();
+    let exit = spawn_exit(dev_config(Some(quota))).await;
+
+    let fetch = |exit: SocketAddr, origin: SocketAddr| async move {
+        let mut neighbor = TcpStream::connect(exit).await.unwrap();
+        let request = format!("GET http://{origin}/file.bin HTTP/1.1\r\nHost: o\r\n\r\n");
+        neighbor.write_all(request.as_bytes()).await.unwrap();
+        let head = read_response_head(&mut neighbor).await;
+        let mut body = Vec::new();
+        let _ = neighbor.read_to_end(&mut body).await;
+        (head, body)
+    };
+
+    let (first, body) = fetch(exit, origin).await;
+    assert!(
+        first.starts_with("HTTP/1.1 200 OK"),
+        "first request must pass: {first}"
+    );
+    assert_eq!(body.len(), 8_000, "first response body must be complete");
+
+    // `record()` runs after the first connection is fully torn down —
+    // poll briefly until the ledger has accounted the traffic.
+    let mut second = String::new();
+    for _ in 0..100 {
+        let (head, _body) = fetch(exit, origin).await;
+        second = head;
+        if second.starts_with("HTTP/1.1 503") {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(
+        second.starts_with("HTTP/1.1 503"),
+        "second request must be refused: {second}"
+    );
+    assert!(second.contains("Retry-After:"), "{second}");
+}
+
+#[tokio::test]
+async fn disallowed_source_gets_403() {
+    // Explicit allowlist WITHOUT loopback → loopback clients refused.
+    let config = ExitConfig::new(
+        SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 0),
+        SourceAllowlist::new(vec![IpAddr::V6(Ipv6Addr::new(0x200, 1, 0, 0, 0, 0, 0, 2))]),
+        None,
+    );
+    let exit = spawn_exit(config).await;
+
+    let mut neighbor = TcpStream::connect(exit).await.unwrap();
+    neighbor
+        .write_all(b"GET http://93.184.216.34/x HTTP/1.1\r\nHost: o\r\n\r\n")
+        .await
+        .unwrap();
+    let response = read_response_head(&mut neighbor).await;
+    assert!(response.starts_with("HTTP/1.1 403"), "{response}");
+}
+
+#[tokio::test]
+async fn origin_form_request_gets_400() {
+    let exit = spawn_exit(dev_config(None)).await;
+    let mut neighbor = TcpStream::connect(exit).await.unwrap();
+    neighbor
+        .write_all(b"GET /file.bin HTTP/1.1\r\nHost: o\r\n\r\n")
+        .await
+        .unwrap();
+    let response = read_response_head(&mut neighbor).await;
+    assert!(response.starts_with("HTTP/1.1 400"), "{response}");
+}
+
+#[tokio::test]
+async fn oversized_head_gets_no_hang_and_no_panic() {
+    let exit = spawn_exit(dev_config(None)).await;
+    let mut neighbor = TcpStream::connect(exit).await.unwrap();
+    let big_line = "x".repeat(20 * 1024);
+    let request = format!("GET http://{big_line}/ HTTP/1.1\r\n\r\n");
+    neighbor.write_all(request.as_bytes()).await.unwrap();
+    // The server must close or answer — either way, no hang.
+    let mut buffer = Vec::new();
+    let _ = tokio::time::timeout(Duration::from_secs(5), neighbor.read_to_end(&mut buffer))
+        .await
+        .expect("no hang on oversized head");
+}