diff --git a/README.md b/README.md index e13f18d..003d9f4 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ Datensammlung, keine Bezahlung. | Ordner/Datei | Inhalt | Einstieg | |---|---|---| | `documentation/` | Idee, Ethos, Architektur, Rechtslage, Roadmap (Deutsch, Obsidian-freundlich mit `[[Wikilinks]]`; `documentation/Home.md` ist die Map of Content) | [README](documentation/README.md) | -| `server/` | Die Software (Rust): `sn-fetch`, `sn-node`, geplant `sn-exit`, `sn-fair`, `sn-cache`, `sn-freenet` | [README](server/README.md) | +| `server/` | Die Software (Rust): `sn-fetch`, `sn-node`, `sn-exit`; geplant `sn-fair`, `sn-cache`, `sn-freenet` | [README](server/README.md) | | `scripts/` | Repo-Setup (`bootstrap-gitea.sh`) | — | | `AGENTS.md` | Arbeitsanweisungen für Coding-Agents und Contributor:innen | [AGENTS.md](AGENTS.md) | | `LICENSE` | Code-Lizenz: AGPL-3.0-only | [LICENSE](LICENSE) | @@ -50,8 +50,11 @@ Prototyp (Konzeptphase + erster Code): lokalen Proxy für Nachbar-Exits. Details: [server/README.md](server/README.md) - **`sn-node` v0** funktioniert: Yggdrasil-Config-Generierung (`genconf`) + Overlay-Status (`status` via Admin-Socket) -- **Nächste Schritte:** `sn-exit` (kontingentierter Uplink-Proxy auf der - Yggdrasil-Adresse), Pilot mit 2–3 Haushalten — siehe +- **`sn-exit` v0** funktioniert: kontingentierter Uplink-Proxy (Byte-Relay, + CONNECT-Tunnel für HTTPS, SSRF-Schutz, Source-Allowlist) — End-to-End + mit `sn-fetch` gegen einen echten Server verifiziert +- **Nächste Schritte:** Pilot mit 2–3 Haushalten (Yggdrasil-Overlay + verkabeln, Messwerte), Fairness-Scheduler — siehe [Roadmap](documentation/ideen/roadmap.md) ## Doku lesen (Obsidian) diff --git a/documentation/README.md b/documentation/README.md index ebe111f..b716d65 100644 --- a/documentation/README.md +++ b/documentation/README.md @@ -28,10 +28,11 @@ ist — ohne Mehrkosten und mit besserer Versorgung schwacher Haushalte. ## Status -Prototypphase (siehe [Roadmap]): `sn-fetch` (segmentierter Multi-Exit- -Download mit Retry + Egress) und `sn-node` v0 (Yggdrasil-Config + Status) -sind implementiert und getestet — Details in [`server/README.md`](../server/README.md). -Nächster Meilenstein: `sn-exit` + Pilot mit 2–3 Haushalten. +Prototypphase (siehe [[Roadmap]]): `sn-fetch` (segmentierter Multi-Exit- +Download mit Retry + Egress), `sn-node` v0 (Yggdrasil-Config + Status) +und `sn-exit` v0 (kontingentierter Uplink-Proxy mit SSRF-Schutz) sind +implementiert und getestet — Details in [`server/README.md`](../server/README.md). +Nächster Meilenstein: Pilot mit 2–3 Haushalten (echtes Overlay, Messwerte). ## Verwandte Projekte (Prior Art) diff --git a/documentation/ideen/roadmap.md b/documentation/ideen/roadmap.md index aae59de..61c0bee 100644 --- a/documentation/ideen/roadmap.md +++ b/documentation/ideen/roadmap.md @@ -25,7 +25,8 @@ Ziel: Beweis, dass Pooling über zwei Anschlüsse real funktioniert. Client (`status`) — die Overlay-Anbindung beginnt - [ ] Yggdrasil-Overlay zwischen 2 Haushalten (Kabel oder Kurzstrecke WLAN/60-GHz), `sn-node`-Daemon orchestriert den Node -- [ ] `sn-exit` v0: HTTP-Proxy auf der Yggdrasil-Adresse (Kontingente) +- [x] `sn-exit` v0: HTTP-Proxy auf der Yggdrasil-Adresse (Kontingente, + SSRF-Schutz, Kill-Switch = Prozess stoppen) - [ ] `sn-fetch` PoC über 2 echte Exits, plus Baseline-Messung (1-Exit vs. 2-Exit, gleiche Datei) - [ ] Fair-Share-Scheduler v0: zwei gleichzeitige Downloads, faire Aufteilung diff --git a/server/Cargo.toml b/server/Cargo.toml index 3731165..ce411f9 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -1,6 +1,6 @@ [workspace] resolver = "2" -members = ["crates/sn-fetch", "crates/sn-node"] +members = ["crates/sn-fetch", "crates/sn-node", "crates/sn-exit"] [workspace.package] version = "0.1.0" diff --git a/server/README.md b/server/README.md index 03076e0..70f9ecd 100644 --- a/server/README.md +++ b/server/README.md @@ -40,9 +40,28 @@ SHA-256 entspricht der offiziellen `SHA256SUMS`). TCP `localhost:9001` oder Unix-Socket) — Overlay-Adresse, Subnetz, Version, aktive Peer-Verbindungen -**Noch nicht drin:** `sn-exit` (der Proxy-Dienst, der auf der Yggdrasil- -Adresse lauscht und Segmente für Nachbarn holt), `sn-node`-Daemon (Supervisor, -der yggdrasil + sn-exit + sn-fetch orchestriert), Fairness-Scheduler. +**`sn-exit` v0 ist implementiert** (siehe `crates/sn-exit/`): + +- HTTP-Proxy-Dienst, der auf der Yggdrasil-Adresse des Haushalts lauscht + und Nachbar-Requests über den eigenen Uplink holt — Sharing ist opt-in + durch Starten des Dienstes, Stoppen ist der Kill-Switch +- **Transparentes Byte-Relay:** absolute-URI-GETs (HTTP) und CONNECT- + Tunnel (HTTPS) werden unverändert weitergeleitet; TLS bleibt end-to-end + zwischen Neighbor-Client und Origin, Range-Header und Retries + funktionieren (Segmentierung bleibt Sache von `sn-fetch`) +- **Quota pro Neighbor:** Bytes pro Zeitfenster (fixed window), + erschöpft → HTTP 503 mit `Retry-After`; Accounting auch bei abgebrochenen + Verbindungen (Counter-Wrapper, nicht Copy-Ergebnis) +- **SSRF-Schutz:** Origins, die zu Loopback/Private/Link-Local/ULA/ + Yggdrasil-Ranges auflösen, werden abgewiesen — Nachbarn können nicht in + Heim-LAN oder Admin-Sockets des Exit-Haushalts fetchen +- Source-Allowlist: nur gelistete Neighbor-Adressen; ohne Liste nur + Loopback (Dev-Mode) +- Verifiziert end-to-end: `sn-fetch` über einen laufenden `sn-exit` + lädt die Debian-ISO (756 MB, 4 Segmente, SHA-256 korrekt) + +**Noch nicht drin:** `sn-node`-Daemon (Supervisor, der yggdrasil + sn-exit ++ sn-fetch orchestriert), Fairness-Scheduler. Details zum Overlay: [[Overlay]]. @@ -75,7 +94,7 @@ Einplatinenrechner), der: | Crate/Binary | Zweck | |---|---| | `sn-node` | Node-Glue: Yggdrasil-Config (`genconf`), Overlay-Status (`status`), später Daemon/Supervisor | -| `sn-exit` | Exit-Dienst: kontingentierter Uplink-Proxy auf der Yggdrasil-Adresse (geplant) | +| `sn-exit` | Exit-Dienst: kontingentierter Uplink-Proxy auf der Yggdrasil-Adresse (v0: Byte-Relay + Quota + SSRF-Schutz) | | `sn-fetch` | Segmentierter Multi-Exit-Downloader (HTTP-API + CLI) | | `sn-fair` | Fairness-Scheduler (max-min, Exit-Kontingente, Karma, Leech-Modus) | | `sn-cache` | Transparenter Nachbarschafts-Cache (DNS-basiert) | diff --git a/server/crates/sn-exit/Cargo.toml b/server/crates/sn-exit/Cargo.toml new file mode 100644 index 0000000..6043d9e --- /dev/null +++ b/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 diff --git a/server/crates/sn-exit/src/config.rs b/server/crates/sn-exit/src/config.rs new file mode 100644 index 0000000..4989e1d --- /dev/null +++ b/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 { + 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, + 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) -> 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, + origin_policy: OriginPolicy, +} + +impl ExitConfig { + pub fn new(bind: SocketAddr, allowlist: SourceAllowlist, quota: Option) -> 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 { + 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 { + input + .parse() + .map_err(|_| ExitError::InvalidBind(input.to_string())) +} + +pub fn parse_source(input: &str) -> Result { + 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::() + .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::().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()); + } +} diff --git a/server/crates/sn-exit/src/error.rs b/server/crates/sn-exit/src/error.rs new file mode 100644 index 0000000..cd904a5 --- /dev/null +++ b/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), +} diff --git a/server/crates/sn-exit/src/lib.rs b/server/crates/sn-exit/src/lib.rs new file mode 100644 index 0000000..a847e1d --- /dev/null +++ b/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; diff --git a/server/crates/sn-exit/src/main.rs b/server/crates/sn-exit/src/main.rs new file mode 100644 index 0000000..e77ada6 --- /dev/null +++ b/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, + + /// 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::, 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()) +} diff --git a/server/crates/sn-exit/src/parse.rs b/server/crates/sn-exit/src/parse.rs new file mode 100644 index 0000000..dacefae --- /dev/null +++ b/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 { + 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::() + .map_err(|_| ExitError::Protocol("bad ipv6 authority"))?; + let port = rest[close + 1..].strip_prefix(':').map_or(Ok(80), |p| { + p.parse::() + .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 { + 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 + } +} diff --git a/server/crates/sn-exit/src/proxy.rs b/server/crates/sn-exit/src/proxy.rs new file mode 100644 index 0000000..10e4e7f --- /dev/null +++ b/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>, +} + +#[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) -> 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) { + 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>, + ) -> std::sync::MutexGuard<'_, HashMap> { + 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, + ledger: Arc, +) -> 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 { + inner: T, + read: std::sync::atomic::AtomicU64, + written: std::sync::atomic::AtomicU64, +} + +impl CountingIo { + 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 tokio::io::AsyncRead for CountingIo { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + 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 tokio::io::AsyncWrite for CountingIo { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + 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::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::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, 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 = + 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 { + 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) +} diff --git a/server/crates/sn-exit/tests/proxy.rs b/server/crates/sn-exit/tests/proxy.rs new file mode 100644 index 0000000..9d25d3d --- /dev/null +++ b/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>) -> 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::().ok()?, end.parse::().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) -> 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 = (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 = (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 = (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"); +}