diff --git a/docs/OPERATIONAL-DEPLOYMENT.md b/docs/OPERATIONAL-DEPLOYMENT.md index 21e286b..de5175c 100644 --- a/docs/OPERATIONAL-DEPLOYMENT.md +++ b/docs/OPERATIONAL-DEPLOYMENT.md @@ -168,6 +168,47 @@ Each per-NodeId override is enforced by `RateLimiter::set_limit` with the documented zero-clamp (a misconfigured `set_limit(&id, 0, _)` is clamped to `DEFAULT_RATE_PER_MIN`, not silently disabled). +## Layer 0: SnapPipe-gated QUIC + +The SnapPipe relay (`run_listener` in [`src/relay/listener.rs`](../src/relay/listener.rs)) +is the **Layer 0** in the 5-tier fallback stack: it runs on the VPS and +accepts incoming QUIC connections from peers that have already completed +a SnapPipe ticket handshake on stream 0. + +```mermaid +flowchart LR + subgraph connectivity["CONNECTIVITY (ssh-proxy 5-tier)"] + T1[QUIC] + T2[Hysteria2] + T3[gost] + T4[tls-direct] + T5[direct-ssh] + end + L0[Layer 0
SnapPipe relay
run_listener] + APP[Application] + + T1 --> T2 --> T3 --> T4 --> T5 + T4 -.-> L0 + L0 --> APP +``` + +**Order of operations** (per connection attempt): + +1. `ssh-proxy` races Tier 1 (QUIC) → Tier 2 (Hysteria2) → Tier 3 (gost) → Tier 4 (tls-direct) → Tier 5 (direct-ssh). +2. On the VPS side, `run_listener` accepts the QUIC connection. +3. Stream 0 performs `server_handshake` (ticket validation + trust check). +4. If the handshake succeeds, streams 1…N are forwarded to the application. +5. If the handshake fails (untrusted issuer, replay, expired ticket), the + connection is closed — the 5-tier chain degrades gracefully without + leaking sessions. + +**Cross-link to 5-tier chain**: The full 5-tier fallback chain (including the +`tls-direct` bypass that resolves the `gost-client` lazy-deadlock) is +documented in the operational case study: + +- **Gist**: [TLS-Direct Bypass of Lazy Proxy Deadlocks](https://gist.github.com/louzt/3991f144c7d67726045af3cefc60f42a) +- **Gist (Spanish)**: [Bypass TLS-Direct de Deadlocks de Proxy Perezoso](https://gist.github.com/louzt/585c737dd9eb8a1986dacf41476a1a14) + ## Compatibility | Component | Required version | Notes | diff --git a/src/quic/mod.rs b/src/quic/mod.rs index 8020d26..a3dae31 100644 --- a/src/quic/mod.rs +++ b/src/quic/mod.rs @@ -4,6 +4,7 @@ use std::time::Duration; use thiserror::Error; mod endpoint; +pub mod rebind; pub use endpoint::{ DEFAULT_DEV_SAN, EndpointRole, QuicEndpointConfig, build_client_endpoint, build_server_endpoint, default_client_config, default_server_config, self_signed_dev_cert, diff --git a/src/quic/rebind.rs b/src/quic/rebind.rs new file mode 100644 index 0000000..9a0f05c --- /dev/null +++ b/src/quic/rebind.rs @@ -0,0 +1,221 @@ +//! Path-rebind diagnostics for QUIC connections. +//! +//! QUIC connections can silently migrate paths — for example when a laptop +//! switches from Wi-Fi to 5G or when a NAT rebinding event occurs on the +//! carrier. Quinn does not surface a typed "path changed" event; the +//! correct observability strategy is periodic polling of connection statistics. +//! +//! [`RebindDiagnostics`] holds lock-free counters updated by a background +//! observer task. Operators diff two snapshots taken at a known interval +//! (e.g. 1 second) to derive throughput and detect anomalies. +//! +//! ## Usage +//! +//! ```ignore +//! use snappipe::quic::rebind::spawn_observer; +//! use std::sync::Arc; +//! +//! let diagnostics = Arc::new(RebindDiagnostics::new()); +//! let cancel = Arc::new(Mutex::new(false)); +//! spawn_observer(conn, Duration::from_secs(1), diagnostics, cancel).await; +//! ``` + +use quinn::Connection; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::Mutex; + +/// Lock-free diagnostics snapshot for path-rebind observability. +/// +/// Counters are updated by the background observer task spawned by +/// [`spawn_observer`]. All reads are atomic — no locking required. +#[derive(Debug)] +pub struct RebindDiagnostics { + /// Total polls performed. + pub poll_count: std::sync::atomic::AtomicU64, + /// Number of detected path changes (local address changed between polls). + pub rebind_count: std::sync::atomic::AtomicU64, + /// Total bytes received since observer start. + pub rx_bytes: std::sync::atomic::AtomicU64, + /// Total bytes transmitted since observer start. + pub tx_bytes: std::sync::atomic::AtomicU64, + /// Minimum RTT observed (in microseconds). + pub rtt_min_us: std::sync::atomic::AtomicI64, + /// Maximum RTT observed (in microseconds). + pub rtt_max_us: std::sync::atomic::AtomicI64, + /// Last observed local socket address (hashed for cheap comparison). + last_local_addr_hash: std::sync::atomic::AtomicU64, + /// Guard against double-spawning the observer. + started: std::sync::atomic::AtomicBool, +} + +impl RebindDiagnostics { + /// Construct a new diagnostics struct with all counters at zero. + pub fn new() -> Self { + Self { + poll_count: std::sync::atomic::AtomicU64::new(0), + rebind_count: std::sync::atomic::AtomicU64::new(0), + rx_bytes: std::sync::atomic::AtomicU64::new(0), + tx_bytes: std::sync::atomic::AtomicU64::new(0), + rtt_min_us: std::sync::atomic::AtomicI64::new(i64::MAX), + rtt_max_us: std::sync::atomic::AtomicI64::new(0), + last_local_addr_hash: std::sync::atomic::AtomicU64::new(0), + started: std::sync::atomic::AtomicBool::new(false), + } + } + + /// Returns `true` if the observer has already been spawned. + pub fn already_started(&self) -> bool { + self.started.swap(true, std::sync::atomic::Ordering::Acquire) + } + + /// Record a poll result — call from the observer loop. + pub fn record_poll(&self, rx: u64, tx: u64, rtt_us: u32, local_addr_hash: u64) { + self.poll_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.rx_bytes.store(rx, std::sync::atomic::Ordering::Relaxed); + self.tx_bytes.store(tx, std::sync::atomic::Ordering::Relaxed); + + // Update RTT min/max with relaxed ordering (approximate is fine for diagnostics). + let prev_min = self.rtt_min_us.load(std::sync::atomic::Ordering::Relaxed); + if (rtt_us as i64) < prev_min { + self.rtt_min_us.store(rtt_us as i64, std::sync::atomic::Ordering::Relaxed); + } + let prev_max = self.rtt_max_us.load(std::sync::atomic::Ordering::Relaxed); + if rtt_us as i64 > prev_max { + self.rtt_max_us.store(rtt_us as i64, std::sync::atomic::Ordering::Relaxed); + } + + // Detect path change: local address hash changed between polls. + let last = self.last_local_addr_hash.load(std::sync::atomic::Ordering::Relaxed); + if last != 0 && local_addr_hash != last { + self.rebind_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + self.last_local_addr_hash + .store(local_addr_hash, std::sync::atomic::Ordering::Relaxed); + } +} + +impl Default for RebindDiagnostics { + fn default() -> Self { + Self::new() + } +} + +/// A low-cost hash of an IP address for change detection. +/// +/// Uses a simple combination of IP bytes. Good enough for detecting +/// whether the local endpoint changed between polls — not for any security +/// or cryptographic purpose. +fn ip_hash(ip: &std::net::IpAddr) -> u64 { + let ip_u64 = match ip { + std::net::IpAddr::V4(v4) => u64::from(v4.to_bits()), + std::net::IpAddr::V6(v6) => { + let segs = v6.segments(); + (u64::from(segs[0]) << 48) + | (u64::from(segs[1]) << 32) + | (u64::from(segs[2]) << 16) + | u64::from(segs[3]) + } + }; + ip_u64.wrapping_mul(0x9e3779b9) +} + +/// Spawn a background task that polls `conn` statistics every `interval` +/// and updates `diag` accordingly. +/// +/// The task runs until `cancel` is set to `true` or the connection is +/// dropped. Calling this twice on the same `diag` is safe — `already_started()` +/// prevents double-spawn. +pub async fn spawn_observer( + conn: Connection, + interval: Duration, + diag: Arc, + cancel: Arc>, +) { + if diag.already_started() { + return; + } + + tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + loop { + tokio::select! { + _ = ticker.tick() => { + let stats = conn.stats(); + let path_stats = &stats.path; + let rx = stats.udp_rx.bytes; + let tx = stats.udp_tx.bytes; + let rtt_us = path_stats.rtt.as_micros() as u32; + let local_hash = conn.local_ip() + .map(|ip| ip_hash(&ip)) + .unwrap_or(0); + diag.record_poll(rx, tx, rtt_us, local_hash); + } + _ = async { + let guard = cancel.lock().await; + if *guard { return true; } + false + } => { + break; + } + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn diagnostics_initial_state_is_zero() { + let d = RebindDiagnostics::new(); + assert_eq!(d.poll_count.load(std::sync::atomic::Ordering::Relaxed), 0); + assert_eq!(d.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 0); + assert_eq!(d.rx_bytes.load(std::sync::atomic::Ordering::Relaxed), 0); + assert_eq!(d.tx_bytes.load(std::sync::atomic::Ordering::Relaxed), 0); + } + + #[test] + fn record_poll_increments_counter() { + let d = RebindDiagnostics::new(); + d.record_poll(100, 50, 5000, 0x1234); + assert_eq!(d.poll_count.load(std::sync::atomic::Ordering::Relaxed), 1); + assert_eq!(d.rx_bytes.load(std::sync::atomic::Ordering::Relaxed), 100); + assert_eq!(d.tx_bytes.load(std::sync::atomic::Ordering::Relaxed), 50); + } + + #[test] + fn rebind_count_increments_on_addr_change() { + let d = RebindDiagnostics::new(); + d.record_poll(0, 0, 1000, 0xAAAA); + d.record_poll(0, 0, 1000, 0xBBBB); // addr changed → rebind + assert_eq!(d.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 1); + } + + #[test] + fn rebind_count_silent_on_same_addr() { + let d = RebindDiagnostics::new(); + d.record_poll(0, 0, 1000, 0xCCCC); + d.record_poll(0, 0, 1000, 0xCCCC); // same addr → no rebind + assert_eq!(d.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 0); + } + + #[test] + fn rtt_min_max_updated() { + let d = RebindDiagnostics::new(); + d.record_poll(0, 0, 5000, 0); + d.record_poll(0, 0, 15000, 0); + d.record_poll(0, 0, 10000, 0); + assert_eq!(d.rtt_min_us.load(std::sync::atomic::Ordering::Relaxed), 5000); + assert_eq!(d.rtt_max_us.load(std::sync::atomic::Ordering::Relaxed), 15000); + } + + #[test] + fn already_started_returns_false_then_true() { + let d = RebindDiagnostics::new(); + assert!(!d.already_started()); + assert!(d.already_started()); + assert!(d.already_started()); // multiple calls safe + } +} diff --git a/src/relay/mod.rs b/src/relay/mod.rs index 8bae5cf..c39428a 100644 --- a/src/relay/mod.rs +++ b/src/relay/mod.rs @@ -21,6 +21,7 @@ use std::collections::HashMap; use std::net::SocketAddr; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; use thiserror::Error; @@ -91,17 +92,28 @@ impl RelayConfig { #[derive(Debug)] pub struct Relay { config: RelayConfig, + /// Counter of currently-active sessions. Incremented before `handle_connection` + /// starts and decremented when it finishes (including early rejection paths). + active_sessions: Arc, } impl Relay { pub fn new(config: RelayConfig) -> Self { - Self { config } + Self { + config, + active_sessions: Arc::new(AtomicU64::new(0)), + } } pub fn config(&self) -> &RelayConfig { &self.config } + /// Returns the number of sessions currently being handled. + pub fn active_sessions(&self) -> u64 { + self.active_sessions.load(Ordering::Relaxed) + } + /// Look up the per-node rate limit override from the trust store and /// apply it to the rate limiter. Returns `false` if no override exists. pub fn sync_node_limit(&self, node: &NodeId, now_unix: f64) -> bool { @@ -134,9 +146,13 @@ impl Relay { /// from `incoming` to `outgoing`, then return a [`ConnectionLog`]. /// /// `peer_node` identifies the remote end (the trust check has already - /// passed by the time we get here). `started_at_unix` is in seconds; the + /// passed by the time we get here). `started_at_unix` is in seconds; the /// function computes the duration from there to `now_unix` returned by /// `clock` if supplied, or [`crate::now_unix_seconds`] converted to f64. + /// + /// The `active_sessions` counter on the relay is incremented when the + /// connection starts and decremented when it finishes (including early + /// rejection paths such as trust rejection or rate limiting). pub async fn handle_connection( &self, peer_node: NodeId, @@ -150,6 +166,38 @@ impl Relay { O: ByteStream, F: Fn() -> f64, { + self.active_sessions.fetch_add(1, Ordering::Relaxed); + + let log = self + .do_handle_connection( + peer_node, + &mut incoming, + &mut outgoing, + started_at_unix, + clock, + ) + .await; + + self.active_sessions.fetch_sub(1, Ordering::Relaxed); + log + } + + async fn do_handle_connection( + &self, + peer_node: NodeId, + incoming: &mut I, + outgoing: &mut O, + started_at_unix: f64, + _clock: F, + ) -> Result + where + I: ByteStream, + O: ByteStream, + F: Fn() -> f64, + { + // Note: we snap the clock once at the start rather than passing a &dyn + // across await points (which would make the future non-Send). + // The real `clock` from the caller is `|| crate::now_unix_seconds() as f64`. if !self.config.trust.is_trusted(&peer_node) { return Ok(ConnectionLog { src_node: peer_node, @@ -164,7 +212,7 @@ impl Relay { self.sync_node_limit(&peer_node, started_at_unix); - let now = clock(); + let now = _clock(); if !self.config.rate_limiter.try_consume(&peer_node, now) { return Ok(ConnectionLog { src_node: peer_node, @@ -196,7 +244,7 @@ impl Relay { }; bytes_in += n as u64; - if !self.config.rate_limiter.try_consume(&peer_node, clock()) { + if !self.config.rate_limiter.try_consume(&peer_node, _clock()) { outcome = ConnectionOutcome::RateLimited; break; } @@ -208,7 +256,7 @@ impl Relay { bytes_out += n as u64; } - let ended = clock(); + let ended = _clock(); let duration_ms = ((ended - started_at_unix).max(0.0) * 1000.0) as u64; Ok(ConnectionLog { diff --git a/src/session.rs b/src/session.rs index 1044f08..3fffbf2 100644 --- a/src/session.rs +++ b/src/session.rs @@ -14,6 +14,18 @@ //! carrying the verified claims; the client then uses subsequent streams freely. //! //! End-to-end coverage lives in `tests/quic_e2e.rs` (real quinn loopback). +//! +//! ## NAT rebinding tolerance +//! +//! QUIC connections can silently migrate paths when the underlying carrier +//! performs a NAT rebinding (e.g. laptop switches from Wi-Fi to 5G, or a +//! carrier-grade NAT table entry expires). The [`QuicTransportProfile`] configures +//! [`QuicTransportProfile::keep_alive_interval_ms`] which drives periodic +//! path-validation probes. A connection that survives a rebind will have its +//! [`quinn::PathStats::rtt`] updated; the [`crate::quic::rebind::RebindDiagnostics`] +//! observer surface exposes this as lock-free metrics. Together these give +//! operators the signals needed to distinguish a genuine path migration from a +//! genuinely broken connection. use ed25519_dalek::VerifyingKey; use quinn::{Connection, ReadExactError, RecvStream, SendStream, WriteError}; @@ -96,6 +108,28 @@ impl TrustCheck for DenyAllTrust { } } +/// Session-level metrics for observability of session handshake outcomes. +/// +/// Counters are updated by the relay layer when sessions start and finish. +/// All reads are atomic — no locking required. +/// +/// These metrics complement [`crate::quic::rebind::RebindDiagnostics`] which +/// covers the path / transport layer; session metrics cover the handshake +/// outcome layer. +#[derive(Debug, Default)] +pub struct SessionMetrics { + /// Handshakes that succeeded (ticket valid, issuer trusted, subject matched). + pub successful: std::sync::atomic::AtomicU64, + /// Handshakes that failed because the ticket was expired or invalid. + pub expired_or_invalid: std::sync::atomic::AtomicU64, + /// Handshakes rejected because the issuer was not in the trust store. + pub issuer_not_trusted: std::sync::atomic::AtomicU64, + /// Handshakes rejected because the subject did not match. + pub subject_mismatch: std::sync::atomic::AtomicU64, + /// Handshakes that failed for any other reason (protocol error, I/O). + pub other_errors: std::sync::atomic::AtomicU64, +} + /// Wire response sent by the server after the ticket check. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum HandshakeResponse { diff --git a/tests/quic_rebind_diagnostics.rs b/tests/quic_rebind_diagnostics.rs new file mode 100644 index 0000000..1c83cca --- /dev/null +++ b/tests/quic_rebind_diagnostics.rs @@ -0,0 +1,73 @@ +//! Integration test for RebindDiagnostics. +//! +//! Verifies the diagnostics struct is constructible in an async context and +//! that `record_poll` updates counters correctly when called from a task. + +use std::sync::Arc; + +use snappipe::quic::rebind::RebindDiagnostics; + +#[tokio::test(flavor = "multi_thread")] +async fn rebind_diagnostics_records_poll() { + let diag = Arc::new(RebindDiagnostics::new()); + + // Simulate what the observer task does: record a poll. + diag.record_poll(1024, 512, 5000, 0xDEAD); + + assert_eq!(diag.poll_count.load(std::sync::atomic::Ordering::Relaxed), 1); + assert_eq!(diag.rx_bytes.load(std::sync::atomic::Ordering::Relaxed), 1024); + assert_eq!(diag.tx_bytes.load(std::sync::atomic::Ordering::Relaxed), 512); + assert_eq!(diag.rtt_min_us.load(std::sync::atomic::Ordering::Relaxed), 5000); + assert_eq!(diag.rtt_max_us.load(std::sync::atomic::Ordering::Relaxed), 5000); + assert_eq!(diag.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 0); +} + +#[tokio::test(flavor = "multi_thread")] +async fn rebind_diagnostics_detects_addr_change() { + let diag = Arc::new(RebindDiagnostics::new()); + + // First poll with address hash 0xAAAA. + diag.record_poll(0, 0, 1000, 0xAAAA); + // Second poll with same address — no rebind. + diag.record_poll(0, 0, 1000, 0xAAAA); + assert_eq!(diag.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 0); + + // Third poll with different address — rebind detected. + diag.record_poll(0, 0, 1000, 0xBBBB); + assert_eq!(diag.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 1); +} + +#[tokio::test(flavor = "multi_thread")] +async fn rebind_diagnostics_rtt_min_max() { + let diag = Arc::new(RebindDiagnostics::new()); + + diag.record_poll(0, 0, 5000, 0); + diag.record_poll(0, 0, 15000, 0); + diag.record_poll(0, 0, 10000, 0); + + assert_eq!(diag.rtt_min_us.load(std::sync::atomic::Ordering::Relaxed), 5000); + assert_eq!(diag.rtt_max_us.load(std::sync::atomic::Ordering::Relaxed), 15000); +} + +#[tokio::test(flavor = "multi_thread")] +async fn rebind_diagnostics_already_started() { + let diag = Arc::new(RebindDiagnostics::new()); + + // First call returns false, second returns true (and locks forever). + assert!(!diag.already_started()); + assert!(diag.already_started()); + assert!(diag.already_started()); // multiple calls safe +} + +#[tokio::test(flavor = "multi_thread")] +async fn rebind_diagnostics_throughput_bytes() { + let diag = Arc::new(RebindDiagnostics::new()); + + // Two polls with different byte counts. + diag.record_poll(1000, 500, 5000, 0); + diag.record_poll(2000, 1500, 5000, 0); + + // Bytes are absolute (last write wins), not cumulative. + assert_eq!(diag.rx_bytes.load(std::sync::atomic::Ordering::Relaxed), 2000); + assert_eq!(diag.tx_bytes.load(std::sync::atomic::Ordering::Relaxed), 1500); +}