diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index b06ee08fe40..441352ce03b 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -24,6 +24,15 @@ implementation = "edge/1" # SETUP name for the relay. Omit for moq-re [listen.reset] secret = "/etc/moq/reset.key" # File holding 64 hex chars. Omit for a random key. +[listen.lb] # QUIC-LB: name this relay in its connection IDs. +id = "01" # Server ID, hex. The same length on every relay. +nonce = 8 # Random bytes after it. Default. + +[listen.lb.forward] # Forward packets that land on the wrong relay to their owner. +bind = "10.0.0.1:4443" # This relay's forward socket, a specific IP its peers list. +key = "/etc/moq/forward.key" # File holding 64 hex chars, the same on every relay. +peers = "/etc/moq/peers.toml" # Server ID to forward socket. Reloaded on change. + [listen.tls] cert = "cert.pem" # Certificate chain and key. Reloaded on change. key = "key.pem" @@ -80,6 +89,27 @@ use the tokio runtime when the listener needs these settings. The `--listen-quic-lb-id` and `--listen-quic-lb-nonce` flags update that table. `unix.allow` needs `unix.bind`. +`[listen.lb.forward]` keeps a connection alive when its packets land on another relay behind +the same anycast address or VIP, after a route change or a rebalance. That relay reads the +server ID in the connection ID and forwards the packet to the owner's forward socket, which +handles it as if the client had sent it there and keeps replying from the shared address. +Handshakes are not forwarded, so the front end must keep each new flow on one relay until it +completes. + +The peers file maps each relay's server ID to its forward `bind`, and may list this relay too, +so one file serves the whole cluster: + +```toml +"01" = "10.0.0.1:4443" +"02" = "10.0.0.2:4443" +``` + +A packet for an unlisted server ID is dropped. A forward is accepted only from a listed address +and under the shared `key` (`openssl rand -hex 32 > forward.key`). A changed file replaces the +map; an invalid one is logged and the last valid map kept, or stops startup. Forwarding needs +`lb.id` and all three settings, which the `--listen-quic-lb-forward-bind`, `-key`, and `-peers` +flags update. + ## \[quic] Transport tuning, applied to accepted and dialed connections alike. diff --git a/quest/m2/README.md b/quest/m2/README.md index 3687d0395c0..a5077df3886 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -33,7 +33,6 @@ lite-07; bitrate enforcement is not part of this release. - [One Ruff pin and Swift formatting on aarch64-linux](/quest/m2/format-gate-followups.md) - only the dev shell pins Ruff, and aarch64-linux checks Swift formatting once nixpkgs builds `swiftc` there - [Weekly mutation testing on critical modules](/quest/m2/mutants.md) - cargo-mutants runs weekly on moq-auth and moq-net's permission, resume, and expiry code - [Video quality](/quest/m2/video-quality/README.md) - a nightly VMAF harness grades every encoder backend and on-demand GoP pops against floors, and moq-boy's encode is fixed under it -- [moq-relay forwards QUIC packets by QUIC-LB server ID](/quest/m2/relay-quic-lb-forward.md) - an established connection that lands on the wrong instance behind an anycast IP or VIP is forwarded to its owner, which replies from the shared address - [P2P](/quest/m2/p2p/README.md) - application-selected peer sessions use shared priorities, routing, and same-epoch handover - [rust2 publishes its UniFFI fork](/quest/m2/rust2-uniffi.md) - waiting on moq-dev/rust2 to import UniFFI and moq's generators and publish them - [Switch to rust2's UniFFI](/quest/m2/uniffi-fork.md) - moq-ffi and every wrapper build from moq-dev/rust2's fork under one pin, and moq's generator code is deleted diff --git a/quest/m2/relay-quic-lb-forward.md b/quest/m2/relay-quic-lb-forward.md deleted file mode 100644 index 7c4476ee0a1..00000000000 --- a/quest/m2/relay-quic-lb-forward.md +++ /dev/null @@ -1,69 +0,0 @@ -# [L] moq-relay forwards QUIC packets by QUIC-LB server ID - -## Goal - -A moq-relay QUIC listener with a QUIC-LB server ID and a peer map forwards -every short-header packet whose connection ID names another server to that -server's unicast address, carrying the client's address. The owning relay -processes it as if it came from the client, and sends that connection's -packets with a configured return source address (direct server return), so -an established connection survives landing on the wrong instance behind an -anycast IP or a load-balancing VIP. Long-header packets are never forwarded: -the handshake relies on the front end keeping a flow on one instance. The -flags are documented inline, with MIGRATE notes. - -## Plan - -Settled with the maintainer, 2026-10-30, for moq.pro's anycast edge cluster -on Cloudflare: - -- Why: Chrome's WebTransport never follows `preferred_address`. Chromium - sends every `https` WebTransport through `DedicatedWebTransportHttp3Client`, - a bare `QuicSpdyClientSession` with no migration path, while its pooled - HTTP/3 client validates the server preferred address. Measured locally - against a moq-relay advertising `--listen-preferred-v4` (Chrome 156: the - net-log shows the address received and never acted on; a forced-QUIC - `fetch()` sends `PATH_CHALLENGE` to it). moq-cli does not migrate either. - So a browser stays on the anycast or VIP address for a connection's life, - and a route change or a VIP rebalance lands its packets on another - instance. -- Scope: established connections only. A handshake split across instances - fails and the client retries. Rejected: a Retry carrying the server ID - (one extra round trip per connection) and a consistent hash of the - client's ID (needs an identical membership view everywhere). -- Connection IDs: the existing plaintext QUIC-LB generator (`listen.lb.id`, - `listen.lb.nonce`, [#848](https://github.com/moq-dev/moq/pull/848)), not the - draft's encrypted form. -- Return path: direct server return. The owner sends a forwarded - connection's packets with the configured return address (the anycast or - VIP address) as source, so only the client-to-server direction takes the - extra hop. Whether the platform allows that source is the deployment's - concern; moq.pro asks Cloudflare. -- Peer map: server ID to unicast address, updated at runtime (the deployment - feeds it, moq.pro from its own API), through config the relay hot-reloads - or the library API; the relay refuses a forward to an unknown ID. - -Open for implementation: - -- The forwarding encapsulation: the original client address and the - address the packet arrived on, prepended to the datagram and sent to the - peer's QUIC port or a dedicated one. Forwards are accepted only from - peer-map addresses and authenticated (a shared key), since a forged - client address would let anyone inject packets or reflect traffic. -- Sending from the return address: a socket that may use a non-local - source (`IP_FREEBIND` or `IP_TRANSPARENT` on Linux) versus binding the - return address where the platform routes it. -- Tests: three relays on loopback addresses, a connection established on - one, its short-header packets delivered to another, and the session - continuing. - -Interactions: per-core shard steering already refuses QUIC-LB connection -IDs (`ShardWithQuicLb`), so if -[Shard the relay by default](/quest/m2/perf/shard-default.md) flips the -default, a listener with `lb.id` stays unsharded. The io_uring listener's -`lb.id` refusal ([io_uring package](/quest/m2/relay-io-uring-package.md)) -can stand: forwarding lives on the tokio backend unless a consumer asks. - -Public API: new listener settings (peer map, return address, forward key) -and a library entry point to update the peer map. Wire: none toward clients; -the peer-to-peer forward format is internal. diff --git a/rs/moq-quic/src/endpoint.rs b/rs/moq-quic/src/endpoint.rs index 4c417fac361..dcf994331ab 100644 --- a/rs/moq-quic/src/endpoint.rs +++ b/rs/moq-quic/src/endpoint.rs @@ -1027,6 +1027,20 @@ impl Endpoint { self.connections.len() } + /// Whether `datagram` from `remote` ends in the stateless reset token a connection's peer + /// issued, so [`handle`](Self::handle) would route it to that connection + /// + /// A stateless reset carries a random connection ID, so a caller that routes datagrams by + /// connection ID before handing them over checks this first. + pub fn is_stateless_reset(&self, remote: SocketAddr, datagram: &[u8]) -> bool { + datagram.len().checked_sub(RESET_TOKEN_SIZE).is_some_and(|start| { + self.index + .connection_reset_tokens + .get(remote, &datagram[start..]) + .is_some() + }) + } + /// Counter for the number of bytes currently used /// in the buffers for Initial messages for pending incoming connections /// and accepts that are still being finalized diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index 1398d7bcf57..3b85b6dd901 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -544,6 +544,13 @@ fn server_stateless_reset() { // Force the server to generate the smallest possible stateless reset pair.client.connections.get_mut(&client_ch).unwrap().ping(); info!("resetting"); + pair.drive_client(); + pair.drive_server(); + // The client recognizes the reset before handling it, and not just any short header. + let (_, _, reset) = pair.client.inbound.back().unwrap(); + assert!(pair.client.endpoint.is_stateless_reset(pair.server.addr, reset)); + assert!(!pair.client.endpoint.is_stateless_reset(pair.server.addr, &[0x40; 64])); + assert!(!pair.client.endpoint.is_stateless_reset(pair.client.addr, reset)); pair.drive(); assert_matches!( pair.client_conn_mut(client_ch).poll(), diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 217e2432788..96cd539073f 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -147,7 +147,13 @@ impl Workers { tls: listen_tls, preferred_v4, preferred_v6, - lb: moq_tokio::listen::LoadBalancer { id: lb_id, nonce: _ }, + lb: + moq_tokio::listen::LoadBalancer { + id: lb_id, + nonce: _, + // Needs an id, refused below. + forward: _, + }, reset_secret, } = server.listen.resolve()?; let quic = &server.quic; diff --git a/rs/moq-tokio/Cargo.toml b/rs/moq-tokio/Cargo.toml index e11723b83f6..194e2dfce93 100644 --- a/rs/moq-tokio/Cargo.toml +++ b/rs/moq-tokio/Cargo.toml @@ -19,7 +19,7 @@ default = ["net", "quic", "aws-lc-rs", "websocket", "tcp", "uds"] net = ["dep:moq-net", "dep:web-transport-trait"] # The async QUIC driver alone (`quic::driver`): moq-quic on tokio, without moq-net or # the MoQ backend. Needs a crypto-provider feature. -driver = ["dep:moq-congestion", "dep:moq-quic", "dep:pin-project-lite", "dep:rustc-hash"] +driver = ["dep:hmac", "dep:moq-congestion", "dep:moq-quic", "dep:pin-project-lite", "dep:rustc-hash", "dep:sha2"] # QUIC and WebTransport over moq-quic, for `https://`, `moqt://`, and `moql://`. quic = ["_transport", "_certs", "driver", "dep:http", "dep:web-transport-proto", "dep:reqwest", "watch", "dep:dns-lookup"] # Server-side certificate handling: chains loaded from disk, self-signed generation, diff --git a/rs/moq-tokio/MIGRATE.md b/rs/moq-tokio/MIGRATE.md index ffbe2a04c7f..1cabb27479f 100644 --- a/rs/moq-tokio/MIGRATE.md +++ b/rs/moq-tokio/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `listen::LoadBalancer` has a `forward` field ([#7539](https://github.com/moq-dev/moq/pull/7539)) + +`listen::LoadBalancer` gained `forward: listen::Forward`, the `[listen.lb.forward]` settings that forward QUIC packets to the server their connection ID names. +The struct is not `#[non_exhaustive]`, so add `..Default::default()` to a `LoadBalancer { id, nonce }` literal, which leaves forwarding off. +An exhaustive destructure, such as a QUIC runtime mapping the resolved listener settings, must name `forward`: honor it, or ignore it with `forward: _` only where `lb.id` is already refused. + ### `tls://` listeners verify client certificates ([#7358](https://github.com/moq-dev/moq/pull/7358)) A `listen.tcp.tls` listener serves the listen policy's client-certificate verifier, so `--listen-tls-root` and pinned `tls.peers` apply to it as they do to QUIC, and a stream-only server with that listener accepts them instead of returning `Error::MtlsUnsupported`. diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index a5d25359d17..c2b4dc2e5bd 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -447,6 +447,34 @@ impl Config { "--listen-quic-lb-nonce requires --listen-quic-lb-id", )); } + let forward = &self.lb.forward; + if forward.configured() { + let refused = [ + ( + self.lb.id.is_none(), + "--listen-quic-lb-forward-* requires --listen-quic-lb-id", + ), + ( + forward.bind.is_none(), + "--listen-quic-lb-forward-* requires --listen-quic-lb-forward-bind", + ), + ( + forward.key.is_none(), + "--listen-quic-lb-forward-bind requires --listen-quic-lb-forward-key", + ), + ( + forward.peers.is_none(), + "--listen-quic-lb-forward-bind requires --listen-quic-lb-forward-peers", + ), + ( + forward.bind.is_some_and(|bind| bind.ip().is_unspecified()), + "--listen-quic-lb-forward-bind needs a specific IP, which its peers list", + ), + ]; + if let Some((_, reason)) = refused.into_iter().find(|(set, _)| *set) { + return Err(crate::Error::NoBackend(reason)); + } + } Ok(()) } @@ -562,9 +590,18 @@ impl Config { #[cfg(feature = "quic")] /// Return the effective QUIC-LB connection-ID encoding. pub(crate) fn load_balancer(&self) -> Option { + let forward = &self.lb.forward; self.lb.id.clone().map(|id| crate::quic::LoadBalancer { id, nonce: self.lb.nonce.unwrap_or(8), + forward: match (forward.bind, &forward.key, &forward.peers) { + (Some(bind), Some(key), Some(peers)) => Some(crate::quic::Forward { + bind, + key: key.clone(), + peers: peers.clone(), + }), + _ => None, + }, }) } } @@ -706,6 +743,7 @@ mod tests { Some(crate::quic::LoadBalancer { id: "ab".parse().unwrap(), nonce: 9, + forward: None, }) ); } @@ -741,6 +779,11 @@ mod tests { [lb] id = "ab" nonce = 8 + +[lb.forward] +bind = "10.0.0.1:4443" +key = "forward.key" +peers = "peers.toml" "#, ) .unwrap(); @@ -749,10 +792,46 @@ nonce = 8 Some(crate::quic::LoadBalancer { id: "ab".parse().unwrap(), nonce: 8, + forward: Some(crate::quic::Forward { + bind: "10.0.0.1:4443".parse().unwrap(), + key: "forward.key".into(), + peers: "peers.toml".into(), + }), }) ); } + /// Forwarding needs a server ID and all three of its settings, and a bind its peers can name. + #[test] + fn quic_lb_forward_refuses_a_partial_config() { + let complete = "[lb]\nid = \"ab\"\n[lb.forward]\nbind = \"10.0.0.1:4443\"\nkey = \"k\"\npeers = \"p\"\n"; + let config: Config = toml::from_str(complete).unwrap(); + config.validate().expect("complete"); + + for (body, needle) in [ + (complete.replace("id = \"ab\"\n", ""), "requires --listen-quic-lb-id"), + ( + complete.replace("bind = \"10.0.0.1:4443\"\n", ""), + "requires --listen-quic-lb-forward-bind", + ), + ( + complete.replace("key = \"k\"\n", ""), + "requires --listen-quic-lb-forward-key", + ), + ( + complete.replace("peers = \"p\"\n", ""), + "requires --listen-quic-lb-forward-peers", + ), + (complete.replace("10.0.0.1", "0.0.0.0"), "needs a specific IP"), + ] { + let config: Config = toml::from_str(&body).unwrap(); + match config.validate() { + Err(crate::Error::NoBackend(reason)) => assert!(reason.contains(needle), "{body}: {reason}"), + other => panic!("{body}: {other:?}"), + } + } + } + /// A stream-only server opens no QUIC listener even with a bind configured, /// which is what lets a worker group hold that address instead. #[tokio::test] @@ -902,4 +981,66 @@ pub struct LoadBalancer { )] #[serde(default, skip_serializing_if = "Option::is_none")] pub nonce: Option, + + /// Forwarding to the other servers behind this one's address (`[listen.lb.forward]`). + #[usage(flatten)] + #[serde(default)] + pub forward: Forward, +} + +/// QUIC-LB forwarding (`[listen.lb.forward]`, `--listen-quic-lb-forward-*`). +/// +/// Behind an anycast address or a load-balanced VIP, a route change or a rebalance can land an +/// established connection's packets on another server. With forwarding, that server hands each +/// packet whose connection ID names another server to the owner, which processes it as if the +/// client had sent it there and keeps replying from the shared address. Only packets after the +/// handshake are forwarded, so the front end must still keep a handshake on one server. +#[derive(Clone, Debug, Default, usage_rs::Args, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +#[non_exhaustive] +pub struct Forward { + /// The UDP address this server receives forwards on, and the one its peers name it by. + /// Needs a specific IP, since a forward is accepted only from a peer's listed address. + #[usage( + name = "listen-quic-lb-forward-bind", + long = "listen-quic-lb-forward-bind", + env = "MOQ_LISTEN_QUIC_LB_FORWARD_BIND", + setting = "listen.lb.forward.bind" + )] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bind: Option, + + /// A file holding the key every server forwarding to this one shares, as 64 hexadecimal + /// characters (`openssl rand -hex 32`). It authenticates each forward, which carries a + /// client address the receiver would otherwise take on trust. + #[usage( + name = "listen-quic-lb-forward-key", + long = "listen-quic-lb-forward-key", + env = "MOQ_LISTEN_QUIC_LB_FORWARD_KEY", + setting = "listen.lb.forward.key" + )] + #[usage(value_hint = usage_rs::ValueHint::FilePath)] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub key: Option, + + /// A TOML file naming each server's forward address by its hex server ID, such as + /// `"02" = "10.0.0.2:4443"`. Reloaded when it changes, keeping the last valid map. + /// This server's own entry is skipped, so every server can share one file. A packet for + /// a server missing from it is dropped. + #[usage( + name = "listen-quic-lb-forward-peers", + long = "listen-quic-lb-forward-peers", + env = "MOQ_LISTEN_QUIC_LB_FORWARD_PEERS", + setting = "listen.lb.forward.peers" + )] + #[usage(value_hint = usage_rs::ValueHint::FilePath)] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub peers: Option, +} + +impl Forward { + /// Whether any forwarding setting is present. + pub fn configured(&self) -> bool { + self.bind.is_some() || self.key.is_some() || self.peers.is_some() + } } diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index e558ab5c7cf..30a55aa6515 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -37,7 +37,7 @@ pub mod web; /// Parsed from, and serialized as, a hex string. Its length must match the load /// balancer's configured server-ID length. #[serde_with::serde_as] -#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)] pub struct ServerId(#[serde_as(as = "serde_with::hex::Hex")] pub(crate) Vec); impl ServerId { @@ -69,6 +69,20 @@ pub(crate) struct LoadBalancer { pub id: ServerId, /// Number of random nonce bytes appended to each connection ID. pub nonce: usize, + /// Forwarding to the other servers behind this one's address. + pub forward: Option, +} + +/// Resolved QUIC-LB forwarding, every setting present. +#[cfg(feature = "quic")] +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct Forward { + /// The forward socket's address. + pub bind: std::net::SocketAddr, + /// The shared key file. + pub key: PathBuf, + /// The peer map file. + pub peers: PathBuf, } /// The congestion control family for a QUIC connection. diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index 7a338e03c85..fc648f13639 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -239,6 +239,27 @@ pub enum Error { /// The flow-control limits are ones a peer would refuse, so none were applied. #[error("invalid quic limits: {0}")] Limits(#[from] moq_quic::LimitsError), + + /// The QUIC-LB forward socket failed to bind. + #[error("failed to bind the QUIC-LB forward socket")] + BindForward(#[source] std::io::Error), + + /// The QUIC-LB forward key file could not be read or does not hold 64 hexadecimal characters. + #[error("invalid QUIC-LB forward key")] + ForwardKey(#[source] crate::secret::Error), + + /// The QUIC-LB peers file could not be read, or names a peer this server cannot forward to. + #[error("invalid QUIC-LB peers file {}: {reason}", path.display())] + ForwardPeers { + /// The peers file. + path: std::path::PathBuf, + /// What is wrong with it. + reason: String, + }, + + /// The QUIC-LB peers file could not be watched for changes. + #[error("failed to watch the QUIC-LB peers file: {0}")] + WatchPeers(String), } impl crate::failover::Aggregate for Error { @@ -554,6 +575,8 @@ fn proto_status(err: &web::proto::ConnectError) -> Option { pub(crate) struct Server { pub quic: driver::endpoint::Endpoint, tls: crate::tls::Serve, + /// Reloads the QUIC-LB peers for as long as the server runs. + _peers: Option, } pub(crate) struct Options { @@ -624,6 +647,7 @@ impl Server { endpoint_config.reset_secret = Some(*secret.as_bytes()); } + let mut forward = None; // Configure connection ID generator with server ID if provided if let Some(shard) = member.as_ref().map(listen::Socket::shard) { if load_balancer.is_some() { @@ -656,8 +680,15 @@ impl Server { nonce_len, "using QUIC-LB compatible connection ID generation" ); + if let Some(config) = &load_balancer.forward { + forward = Some(Forwarding::start( + config, + &server_id, + endpoint_config.max_udp_payload_size, + )?); + } endpoint_config.cid_generator = Some(Arc::new(move || { - Box::new(ServerIdGenerator::new(server_id.clone(), nonce_len)) + Box::new(driver::forward::ServerIdGenerator::new(&server_id.0, nonce_len)) })); } @@ -669,11 +700,22 @@ impl Server { }; let socket = driver::wrap_udp_socket(socket).map_err(|err| Error::CreateEndpoint(err.into()))?; - let quic = - driver::endpoint::Endpoint::new(endpoint_config, demux.wrap(socket)).map_err(Error::CreateEndpoint)?; + let socket = demux.wrap(socket); + let (quic, peers) = match forward { + Some(Forwarding { forward, watcher }) => ( + driver::endpoint::Endpoint::forwarding(endpoint_config, socket, forward), + Some(watcher), + ), + None => (driver::endpoint::Endpoint::new(endpoint_config, socket), None), + }; + let quic = quic.map_err(Error::CreateEndpoint)?; demux.attach(&quic); - Ok(Self { quic, tls: served }) + Ok(Self { + quic, + tls: served, + _peers: peers, + }) } pub fn accept(&self) -> impl std::future::Future> + '_ { @@ -817,38 +859,73 @@ pub(crate) async fn accept( } } -// ── ServerIdGenerator ─────────────────────────────────────────────── +// ── Forwarding ────────────────────────────────────────────────────── -struct ServerIdGenerator { - server_id: ServerId, - nonce_len: usize, +/// A forward socket, and the watcher that keeps its peers current. +struct Forwarding { + forward: driver::forward::Forward, + watcher: notify::RecommendedWatcher, } -impl ServerIdGenerator { - fn new(server_id: ServerId, nonce_len: usize) -> Self { - Self { server_id, nonce_len } - } -} - -impl driver::ConnectionIdGenerator for ServerIdGenerator { - fn generate_cid(&mut self) -> driver::ConnectionId { - use rand::RngExt; - let cid_len = self.cid_len(); - let mut cid = Vec::with_capacity(cid_len); - // First byte has "self-encoded length" of server ID + nonce - cid.push((cid_len - 1) as u8); - cid.extend(self.server_id.0.iter()); - cid.extend(rand::rng().random_iter::().take(self.nonce_len)); - driver::ConnectionId::new(cid.as_slice()) - } +impl Forwarding { + fn start(config: &crate::quic::Forward, id: &ServerId, max_udp_payload_size: u16) -> Result { + let key = crate::secret::Secret::read(&config.key).map_err(Error::ForwardKey)?; + let peers = driver::forward::Peers::default(); + peers.set(read_peers(&config.peers, id, config.bind)?); + + let reload = peers.clone(); + let (path, own, bind) = (config.peers.clone(), id.clone(), config.bind); + let watcher = crate::watch::callback(std::slice::from_ref(&config.peers), move || { + match read_peers(&path, &own, bind) { + Ok(map) => { + reload.set(map); + tracing::info!(path = %path.display(), "reloaded QUIC-LB peers"); + } + Err(err) => tracing::warn!(%err, "failed to reload QUIC-LB peers; retaining the previous ones"), + } + }) + .map_err(|err| Error::WatchPeers(err.to_string()))?; - fn cid_len(&self) -> usize { - 1 + self.server_id.len() + self.nonce_len + let socket = std::net::UdpSocket::bind(config.bind).map_err(Error::BindForward)?; + let forward = driver::forward::Forward::new(socket, &id.0, peers, key.as_bytes(), max_udp_payload_size) + .map_err(Error::BindForward)?; + tracing::info!(bind = %config.bind, "forwarding QUIC-LB packets"); + Ok(Self { forward, watcher }) } +} - fn cid_lifetime(&self) -> Option { - None +/// Read the peers file, skipping this server's own entry. +/// +/// Refuses the whole file for one bad entry, so a typo cannot quietly strand that server's +/// clients: a server ID of another length, which this server's connection IDs could not carry, +/// or an address it cannot send to or accept from. +fn read_peers(path: &std::path::Path, own: &ServerId, bind: net::SocketAddr) -> Result { + let invalid = |reason: String| Error::ForwardPeers { + path: path.to_owned(), + reason, + }; + let text = std::fs::read_to_string(path).map_err(|err| invalid(err.to_string()))?; + let peers: std::collections::HashMap = + toml::from_str(&text).map_err(|err| invalid(err.to_string()))?; + + let mut map = driver::forward::Map::default(); + for (id, addr) in peers { + if id.len() != own.len() { + return Err(invalid(format!( + "{id:?} is not {} bytes like this server's ID", + own.len() + ))); + } + if addr.ip().is_unspecified() || addr.is_ipv4() != bind.is_ipv4() { + return Err(invalid(format!( + "{id:?} has an address {bind} cannot forward to: {addr}" + ))); + } + if id != *own { + map.insert(&id.0, addr); + } } + Ok(map) } #[cfg(test)] @@ -1168,4 +1245,33 @@ mod tests { .await .expect("test timed out"); } + + /// The peers file skips this server's own entry, so every server can share it, and refuses + /// the whole file for one entry this server could not forward to. + #[test] + fn peers_file_is_shared_and_refused_whole() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("peers.toml"); + let own: ServerId = "01".parse().unwrap(); + let bind: net::SocketAddr = "10.0.0.1:4443".parse().unwrap(); + let read = |body: &str| { + std::fs::write(&path, body).unwrap(); + read_peers(&path, &own, bind) + }; + + let map = read("\"01\" = \"10.0.0.1:4443\"\n\"02\" = \"10.0.0.2:4443\"\n").unwrap(); + let expected: std::collections::HashMap, net::SocketAddr> = + [(Box::from(&[2u8][..]), "10.0.0.2:4443".parse().unwrap())].into(); + assert_eq!(map.peers(), &expected); + + for (body, needle) in [ + ("\"0203\" = \"10.0.0.2:4443\"\n", "not 1 bytes"), + ("\"02\" = \"[2001:db8::2]:4443\"\n", "cannot forward to"), + ("\"02\" = \"0.0.0.0:4443\"\n", "cannot forward to"), + ("\"zz\" = \"10.0.0.2:4443\"\n", "Invalid character"), + ] { + let err = read(body).expect_err(body).to_string(); + assert!(err.contains(needle), "{body}: {err}"); + } + } } diff --git a/rs/moq-tokio/src/quic/driver/endpoint.rs b/rs/moq-tokio/src/quic/driver/endpoint.rs index 67b7bcf874c..bf9cf4fceea 100644 --- a/rs/moq-tokio/src/quic/driver/endpoint.rs +++ b/rs/moq-tokio/src/quic/driver/endpoint.rs @@ -31,6 +31,7 @@ pub use crate::quic::driver::runtime::AsyncUdpSocket; use crate::quic::driver::{ ConnectionEvent, IO_LOOP_BOUND, Instant, RECV_TIME_BOUND, VarInt, connection::Connecting, + forward::Forward, runtime::{Slot, wrap_udp_socket}, udp_transmit, work_limiter::WorkLimiter, @@ -100,8 +101,19 @@ impl Endpoint { /// The endpoint's tasks run on the caller's tokio runtime, else moq-tokio's fallback. With /// the `sim` feature, an endpoint made inside `kio::clock::sim::run` runs them on the simulator. pub fn new(config: Config, socket: impl Into) -> Result { + Self::start(config, socket.into(), None) + } + + /// [`new`](Self::new), forwarding each packet whose QUIC-LB server ID names another server + /// and accepting the packets those servers forward here. + #[cfg_attr(not(feature = "quic"), allow(dead_code, reason = "only the QUIC backend forwards"))] + pub(crate) fn forwarding(config: Config, socket: impl Into, forward: Forward) -> Result { + Self::start(config, socket.into(), Some(forward)) + } + + fn start(config: Config, socket: Socket, forward: Option) -> Result { let runtime = crate::runtime::Spawner::current(); - let socket = socket.into().into_async()?; + let socket = socket.into_async()?; let addr = socket.local_addr()?; let allow_mtud = !socket.may_fragment(); let max_receive_segments = socket.max_receive_segments(); @@ -113,6 +125,7 @@ impl Endpoint { max_udp_payload_size, max_receive_segments, runtime.clone(), + forward, ); let driver = EndpointDriver(rc.clone()); runtime.spawn( @@ -230,8 +243,11 @@ impl Endpoint { async move { endpoint.wait_all_draining().await; endpoint.inner.slot.release(); + let state = &mut *endpoint.inner.state.lock().unwrap(); // A rebind keeps the old socket until traffic arrives on the new one. - endpoint.inner.state.lock().unwrap().prev_socket = None; + state.prev_socket = None; + // Nothing reaches the endpoint to forward once its socket is gone. + state.forward = None; } } @@ -381,6 +397,7 @@ impl Drop for EndpointDriver { self.0.slot.release(); let endpoint = &mut *self.0.state.lock().unwrap(); endpoint.prev_socket = None; + endpoint.forward = None; endpoint.driver_lost = true; self.0.shared.incoming.notify_waiters(); // Drop all outgoing channels, signaling the termination of the endpoint to the associated @@ -503,6 +520,8 @@ pub(crate) struct State { driver_lost: bool, /// Connections in the process of being accepted pending_accepts: usize, + /// QUIC-LB forwarding to and from the other servers behind this one's address + forward: Option, } #[derive(Debug)] @@ -523,6 +542,7 @@ impl State { let poll_res = self.recv_state.poll_socket( cx, &mut self.inner, + self.forward.as_mut(), &mut |cx, bufs, meta| socket.poll_recv(cx, bufs, meta), slot, now, @@ -534,10 +554,19 @@ impl State { let poll_res = self.recv_state.poll_socket( cx, &mut self.inner, + self.forward.as_mut(), &mut |cx, bufs, meta| slot.poll_recv(cx, bufs, meta), slot, now, ); + // Polled even when the socket used up the budget, which still admits one forward, so a + // busy socket cannot starve them. + let forwards = match self.forward.as_mut() { + Some(forward) => self + .recv_state + .poll_forwards(cx, &mut self.inner, forward, self.ipv6, slot, now), + None => Ok(false), + }; self.recv_state.recv_limiter.finish_cycle(get_time); let poll_res = poll_res?; if poll_res.received_connection_packet { @@ -545,7 +574,7 @@ impl State { // one anymore. TODO: Account for multiple outgoing connections. self.prev_socket = None; } - Ok(poll_res.keep_going) + Ok(poll_res.keep_going || forwards?) } fn handle_events(&mut self, cx: &mut Context<'_>, shared: &Shared) -> bool { @@ -769,6 +798,7 @@ impl EndpointRef { max_udp_payload_size: u16, max_receive_segments: usize, runtime: crate::runtime::Spawner, + forward: Option, ) -> Self { let (sender, events) = mpsc::unbounded_channel(); let recv_state = RecvState::new(sender, max_receive_segments, max_udp_payload_size); @@ -793,6 +823,7 @@ impl EndpointRef { driver_lost: false, recv_state, pending_accepts: 0, + forward, }), })) } @@ -869,6 +900,7 @@ impl RecvState { &mut self, cx: &mut Context<'_>, endpoint: &mut moq_quic::endpoint::Endpoint, + mut forward: Option<&mut Forward>, recv: &mut PollRecv<'_>, slot: &Slot, now: Instant, @@ -918,7 +950,12 @@ impl RecvState { ); while !data.is_empty() { let buf = data.split_to(meta.stride.min(data.len())); - let mut response_buffer = Vec::new(); + if let Some(forward) = forward.as_deref_mut() + && forward.route(&buf, meta.addr, meta.dst_ip, meta.ecn, || { + endpoint.is_stateless_reset(meta.addr, &buf) + }) { + continue; + } let received = Received { remote: meta.addr, local_ip: meta.dst_ip, @@ -926,30 +963,8 @@ impl RecvState { data: buf, arrival: arrival.take(), }; - match endpoint.handle(now, received, &mut response_buffer) { - Some(DatagramEvent::NewConnection(incoming)) => { - if self.connections.close.is_none() { - self.incoming.push_back(incoming); - } else { - let transmit = endpoint.refuse(incoming, &mut response_buffer); - respond(transmit, &response_buffer, slot); - } - } - Some(DatagramEvent::ConnectionEvent(handle, event)) => { - // Ignoring errors from dropped connections that haven't yet been cleaned up - received_connection_packet = true; - let _ = self - .connections - .senders - .get_mut(&handle) - .unwrap() - .send(ConnectionEvent::Proto(event)); - } - Some(DatagramEvent::Response(transmit)) => { - respond(transmit, &response_buffer, slot); - } - None => {} - } + received_connection_packet |= + Self::handle(&mut self.incoming, &mut self.connections, endpoint, now, received, slot); } } } @@ -985,6 +1000,92 @@ impl RecvState { } } +impl RecvState { + /// Hand one received datagram to the endpoint, returning whether it reached a connection. + fn handle( + incoming_queue: &mut VecDeque, + connections: &mut ConnectionSet, + endpoint: &mut moq_quic::endpoint::Endpoint, + now: Instant, + received: Received, + slot: &Slot, + ) -> bool { + let mut response_buffer = Vec::new(); + match endpoint.handle(now, received, &mut response_buffer) { + Some(DatagramEvent::NewConnection(incoming)) => { + if connections.close.is_none() { + incoming_queue.push_back(incoming); + } else { + let transmit = endpoint.refuse(incoming, &mut response_buffer); + respond(transmit, &response_buffer, slot); + } + } + Some(DatagramEvent::ConnectionEvent(handle, event)) => { + // Ignoring errors from dropped connections that haven't yet been cleaned up + let _ = connections + .senders + .get_mut(&handle) + .unwrap() + .send(ConnectionEvent::Proto(event)); + return true; + } + Some(DatagramEvent::Response(transmit)) => { + respond(transmit, &response_buffer, slot); + } + None => {} + } + false + } + + /// Process the packets other servers forwarded here, returning whether the work limiter + /// stopped before they ran out. + fn poll_forwards( + &mut self, + cx: &mut Context<'_>, + endpoint: &mut moq_quic::endpoint::Endpoint, + forward: &mut Forward, + ipv6: bool, + slot: &Slot, + now: Instant, + ) -> Result { + loop { + match forward.poll_recv(cx) { + Poll::Ready(Ok(Some(unwrapped))) => { + // The client's address as this socket would have reported it, so the + // connection sees the path its handshake took. An IPv6 client cannot have + // shaken hands on an IPv4 socket. + let remote = match (ipv6, unwrapped.remote) { + (true, remote) => Some(SocketAddr::V6(ensure_ipv6(remote))), + (false, remote @ SocketAddr::V4(_)) => Some(remote), + (false, SocketAddr::V6(_)) => None, + }; + if let Some(remote) = remote { + self.datagrams.extend_from_slice(unwrapped.packet); + let received = Received { + remote, + local_ip: unwrapped.local_ip, + ecn: unwrapped.ecn.map(proto_ecn), + data: self.datagrams.split(), + // When the forward reached this server: the hop between servers is + // part of the client's path now, and their clocks are not shared. + arrival: Some(Instant::now()), + }; + Self::handle(&mut self.incoming, &mut self.connections, endpoint, now, received, slot); + } + } + Poll::Ready(Ok(None)) => {} + Poll::Pending => return Ok(false), + Poll::Ready(Err(e)) if e.kind() == io::ErrorKind::ConnectionReset || is_msg_size_err(&e) => {} + Poll::Ready(Err(e)) => return Err(e), + } + self.recv_limiter.record_work(1); + if !self.recv_limiter.allow_work(Instant::now) { + return Ok(true); + } + } + } +} + impl fmt::Debug for RecvState { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("RecvState") @@ -1043,6 +1144,7 @@ mod tests { .poll_socket( &mut cx, &mut endpoint, + None, &mut |cx, bufs, meta| socket.poll_recv(cx, bufs, meta), &slot, now, diff --git a/rs/moq-tokio/src/quic/driver/forward.rs b/rs/moq-tokio/src/quic/driver/forward.rs new file mode 100644 index 00000000000..55dce4bcaec --- /dev/null +++ b/rs/moq-tokio/src/quic/driver/forward.rs @@ -0,0 +1,422 @@ +//! QUIC-LB forwarding between the servers behind one address. +//! +//! A server whose connection IDs carry a QUIC-LB server ID hands each short-header packet naming +//! another server to that server's forward socket, wrapped with the client's address and the +//! address the packet arrived on. The owner unwraps it and processes it as if the client had sent +//! it there. Replies need nothing more: a connection sends from the address its handshake arrived +//! on, which is the shared one, so only the client-to-server direction takes the extra hop. +//! +//! Long headers are never forwarded: a client picks the destination ID of its handshake, so the +//! handshake relies on the front end keeping a flow on one server. +#![cfg_attr(not(feature = "quic"), allow(dead_code, reason = "only the QUIC backend forwards"))] + +use std::{ + collections::{HashMap, HashSet}, + io, + net::{IpAddr, Ipv6Addr, SocketAddr}, + sync::{Arc, PoisonError, RwLock}, + task::{Context, Poll, ready}, +}; + +use hmac::{KeyInit, Mac}; +use moq_sock::udp::EcnCodepoint; + +type Hmac = hmac::Hmac; + +/// Where a QUIC-LB connection ID carries its server ID: after the byte encoding its length. +pub(crate) const ID_OFFSET: usize = 1; + +/// The forward header's version, so a mismatched peer's forwards are dropped rather than misread. +const VERSION: u8 = 0; +/// The truncated HMAC-SHA256 tag that authenticates a forward. +const TAG_LEN: usize = 16; +/// The flag marking a forward that knows the address its packet arrived on. +const ARRIVAL: u8 = 0b100; +/// The header ahead of a forwarded packet: version, flags (ECN and [`ARRIVAL`]), the client's +/// port and IP, the arrival IP, and the tag over all of it and the packet. +pub(crate) const OVERHEAD: usize = 1 + 1 + 2 + 16 + 16 + TAG_LEN; + +/// Generates QUIC-LB connection IDs: a byte encoding the length, the server ID, then a random nonce. +pub(crate) struct ServerIdGenerator { + id: Box<[u8]>, + nonce_len: usize, +} + +impl ServerIdGenerator { + /// Name `id` in every connection ID, ahead of `nonce_len` random bytes. + pub(crate) fn new(id: &[u8], nonce_len: usize) -> Self { + Self { + id: id.into(), + nonce_len, + } + } +} + +impl moq_quic::ConnectionIdGenerator for ServerIdGenerator { + fn generate_cid(&mut self) -> moq_quic::ConnectionId { + use rand::RngExt; + let cid_len = self.cid_len(); + let mut cid = Vec::with_capacity(cid_len); + // The "self-encoded length" of server ID and nonce, which puts the server ID at `ID_OFFSET`. + cid.push((cid_len - 1) as u8); + cid.extend(self.id.iter()); + cid.extend(rand::rng().random_iter::().take(self.nonce_len)); + moq_quic::ConnectionId::new(cid.as_slice()) + } + + fn cid_len(&self) -> usize { + ID_OFFSET + self.id.len() + self.nonce_len + } + + fn cid_lifetime(&self) -> Option { + None + } +} + +/// The other servers behind the shared address, by server ID, replaced whole on each update. +#[derive(Clone, Debug, Default)] +pub(crate) struct Peers(Arc>); + +/// A snapshot of [`Peers`]. +#[derive(Debug, Default)] +pub(crate) struct Map { + by_id: HashMap, SocketAddr>, + /// The forward sockets accepted as sources. + addrs: HashSet, +} + +impl Map { + /// Name `id`'s forward socket. + pub(crate) fn insert(&mut self, id: &[u8], addr: SocketAddr) { + self.by_id.insert(id.into(), addr); + self.addrs.insert(addr); + } + + /// Every peer, by server ID. + #[cfg(test)] + pub(crate) fn peers(&self) -> &HashMap, SocketAddr> { + &self.by_id + } +} + +impl Peers { + /// Replace every peer at once. + pub(crate) fn set(&self, map: Map) { + *self.0.write().unwrap_or_else(PoisonError::into_inner) = map; + } + + fn get(&self, id: &[u8]) -> Option { + self.0 + .read() + .unwrap_or_else(PoisonError::into_inner) + .by_id + .get(id) + .copied() + } + + fn contains(&self, addr: &SocketAddr) -> bool { + self.0 + .read() + .unwrap_or_else(PoisonError::into_inner) + .addrs + .contains(addr) + } +} + +/// A forwarded packet, as the client sent it. +pub(crate) struct Unwrapped<'a> { + pub remote: SocketAddr, + pub local_ip: Option, + pub ecn: Option, + pub packet: &'a [u8], +} + +/// One server's forwarding: its server ID, its peers, and the socket forwards travel on. +#[derive(Debug)] +pub(crate) struct Forward { + socket: tokio::net::UdpSocket, + /// The same socket, sending without waiting on the reactor to report it writable, which a + /// fresh socket is not yet. + send: std::net::UdpSocket, + id: Box<[u8]>, + peers: Peers, + key: Hmac, + /// Scratch for one forward, sent or received. + buf: Box<[u8]>, +} + +impl Forward { + /// Forward on `socket` between the servers in `peers`, this one named `id` in its connection IDs. + /// + /// `max_udp_payload_size` bounds the packets a client sends, so a forward never exceeds it + /// plus [`OVERHEAD`]. The socket may fragment: a full-size packet plus the header can exceed + /// the path between servers, which carries it anyway. + pub(crate) fn new( + socket: std::net::UdpSocket, + id: &[u8], + peers: Peers, + key: &[u8; 32], + max_udp_payload_size: u16, + ) -> io::Result { + socket.set_nonblocking(true)?; + let send = socket.try_clone()?; + let socket = { + let _runtime = crate::runtime::enter(); + tokio::net::UdpSocket::from_std(socket)? + }; + Ok(Self { + socket, + send, + id: id.into(), + peers, + key: Hmac::new_from_slice(key).expect("HMAC accepts any key length"), + buf: vec![0; usize::from(max_udp_payload_size) + OVERHEAD].into(), + }) + } + + /// Forward `packet` if its connection ID names another server, returning whether it was taken. + /// + /// A packet naming a server missing from the peers is taken and dropped: answering it here + /// would reset a connection another server owns. One that `is_local_reset` recognizes stays, + /// since a stateless reset's connection ID is random and may seem to name another server. + pub(crate) fn route( + &mut self, + packet: &[u8], + remote: SocketAddr, + local_ip: Option, + ecn: Option, + is_local_reset: impl FnOnce() -> bool, + ) -> bool { + let Some(id) = self.foreign(packet) else { + return false; + }; + if is_local_reset() { + return false; + } + let Some(peer) = self.peers.get(id) else { + tracing::trace!(id = %hex::encode(id), "dropping a packet for an unknown server"); + return true; + }; + // A client never sends more than the advertised payload size, so a longer datagram is + // not a packet for one of our connections. + let Some(len) = self.seal(packet, remote, local_ip, ecn) else { + return true; + }; + match self.send.send_to(&self.buf[..len], peer) { + Ok(_) => {} + // Lost like any datagram on a full socket; the client retransmits. + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => tracing::debug!(%err, %peer, "failed to forward a packet"), + } + true + } + + /// Wrap `packet` into the scratch buffer, returning the forward's length, unless it would not fit. + fn seal( + &mut self, + packet: &[u8], + remote: SocketAddr, + local_ip: Option, + ecn: Option, + ) -> Option { + let buf = self.buf.get_mut(..OVERHEAD + packet.len())?; + buf[0] = VERSION; + buf[1] = ecn.map_or(0, |ecn| ecn as u8) | local_ip.map_or(0, |_| ARRIVAL); + buf[2..4].copy_from_slice(&remote.port().to_be_bytes()); + buf[4..20].copy_from_slice(&ipv6(remote.ip()).octets()); + buf[20..36].copy_from_slice(&local_ip.map_or(Ipv6Addr::UNSPECIFIED, ipv6).octets()); + buf[OVERHEAD..].copy_from_slice(packet); + let tag = tag(&self.key, buf).finalize().into_bytes(); + buf[OVERHEAD - TAG_LEN..OVERHEAD].copy_from_slice(&tag[..TAG_LEN]); + Some(buf.len()) + } + + /// Receive the next authentic forward, or `None` for a datagram that is not one. + pub(crate) fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll>>> { + let mut read = tokio::io::ReadBuf::new(&mut self.buf); + let from = ready!(self.socket.poll_recv_from(cx, &mut read))?; + let len = read.filled().len(); + Poll::Ready(Ok(self.open(from, len))) + } + + /// Authenticate the forward in the first `len` bytes of the scratch buffer. + fn open(&self, from: SocketAddr, len: usize) -> Option> { + // Checked before the tag, so a stranger costs a lookup and never a hash. + if !self.peers.contains(&from) { + return None; + } + let buf = &self.buf[..len]; + if len < OVERHEAD || buf[0] != VERSION { + return None; + } + tag(&self.key, buf) + .verify_truncated_left(&buf[OVERHEAD - TAG_LEN..OVERHEAD]) + .ok()?; + + let packet = &buf[OVERHEAD..]; + // Never forward again, and never answer for a connection another server owns: a peer + // with a stale view of who owns an ID would otherwise loop or reset it. + if self.foreign(packet).is_some() || packet.first().is_none_or(|first| first & 0x80 != 0) { + return None; + } + + let flags = buf[1]; + let port = u16::from_be_bytes([buf[2], buf[3]]); + let ip = |at: usize| canonical(Ipv6Addr::from(<[u8; 16]>::try_from(&buf[at..at + 16]).unwrap())); + Some(Unwrapped { + remote: SocketAddr::new(ip(4), port), + local_ip: (flags & ARRIVAL != 0).then(|| ip(20)), + ecn: EcnCodepoint::from_bits(flags), + packet, + }) + } + + /// The server ID in `packet`'s connection ID, when it is a short header naming another server. + fn foreign<'a>(&self, packet: &'a [u8]) -> Option<&'a [u8]> { + if packet.first()? & 0x80 != 0 { + return None; + } + let start = 1 + ID_OFFSET; + let id = packet.get(start..start + self.id.len())?; + (id != &*self.id).then_some(id) + } +} + +/// The MAC over `forward`'s header, short of the tag itself, and its packet. +fn tag(key: &Hmac, forward: &[u8]) -> Hmac { + let mut mac = key.clone(); + mac.update(&forward[..OVERHEAD - TAG_LEN]); + mac.update(&forward[OVERHEAD..]); + mac +} + +fn ipv6(ip: IpAddr) -> Ipv6Addr { + match ip { + IpAddr::V4(ip) => ip.to_ipv6_mapped(), + IpAddr::V6(ip) => ip, + } +} + +fn canonical(ip: Ipv6Addr) -> IpAddr { + match ip.to_ipv4_mapped() { + Some(ip) => IpAddr::V4(ip), + None => IpAddr::V6(ip), + } +} + +#[cfg(test)] +mod tests { + use std::{future::poll_fn, net::Ipv4Addr}; + + use super::*; + + const KEY: [u8; 32] = [7; 32]; + + type Opened = (SocketAddr, Option, Option, Vec); + + fn socket() -> std::net::UdpSocket { + std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap() + } + + /// The server named `id`, forwarding on `socket` under `key`. + fn server(id: u8, socket: std::net::UdpSocket, peers: &Peers, key: &[u8; 32]) -> Forward { + Forward::new(socket, &[id], peers.clone(), key, 1200).unwrap() + } + + /// A short header whose connection ID names server `id`. + fn short(id: u8) -> Vec { + vec![0x40, 5, id, 0xa, 0xb, 0xc, 0xd, 0x01, 0x02] + } + + /// The next datagram `forward` receives, opened. + async fn recv(forward: &mut Forward) -> Option { + let next = poll_fn(|cx| { + forward + .poll_recv(cx) + .map_ok(|opened| opened.map(|o| (o.remote, o.local_ip, o.ecn, o.packet.to_vec()))) + }); + kio::clock::timeout(std::time::Duration::from_secs(5), next) + .await + .expect("a datagram") + .unwrap() + } + + /// Servers 1 and 2, each listing both. + fn pair() -> (Forward, Forward, Peers) { + let (a, b) = (socket(), socket()); + let peers = Peers::default(); + let mut map = Map::default(); + map.insert(&[1], a.local_addr().unwrap()); + map.insert(&[2], b.local_addr().unwrap()); + peers.set(map); + (server(1, a, &peers, &KEY), server(2, b, &peers, &KEY), peers) + } + + #[tokio::test] + async fn a_forward_arrives_as_the_client_sent_it() { + let (mut a, mut b, _) = pair(); + let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); + let arrival: IpAddr = "198.51.100.1".parse().unwrap(); + + assert!(a.route(&short(2), client, Some(arrival), Some(EcnCodepoint::Ce), || false)); + assert_eq!( + recv(&mut b).await, + Some((client, Some(arrival), Some(EcnCodepoint::Ce), short(2))) + ); + + // An IPv6 client, through a server that never learned the arrival address. + let client: SocketAddr = "[2001:db8::7]:5000".parse().unwrap(); + assert!(a.route(&short(2), client, None, None, || false)); + assert_eq!(recv(&mut b).await, Some((client, None, None, short(2)))); + } + + /// What stays: this server's own connection IDs, long headers, anything too short to carry a + /// server ID, and a stateless reset for one of this server's connections, whose random + /// connection ID may seem to name another. A short header for a server nobody listed is taken + /// and dropped. + #[tokio::test] + async fn only_another_servers_short_headers_leave() { + let (mut a, _b, _) = pair(); + let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); + assert!(!a.route(&short(1), client, None, None, || false)); + let mut long = short(2); + long[0] = 0xc0; + assert!(!a.route(&long, client, None, None, || false)); + assert!(!a.route(&[0x40, 5], client, None, None, || false)); + assert!(a.route(&short(3), client, None, None, || false)); + assert!(!a.route(&short(2), client, None, None, || true)); + } + + /// A forward opens only from a listed peer, under the shared key, and only for this server: a + /// peer with a stale map must not make it forward again or answer for another server. + #[tokio::test] + async fn a_forward_needs_a_peer_the_key_and_this_server() { + let (mut a, mut b, peers) = pair(); + let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); + let b_addr = b.socket.local_addr().unwrap(); + + // Not listed, with the right key. + let mut stranger = server(1, socket(), &peers, &KEY); + assert!(stranger.route(&short(2), client, None, None, || false)); + assert_eq!(recv(&mut b).await, None); + + // Listed, with the wrong key. + let mut forged = server(3, socket(), &peers, &[8; 32]); + let mut map = Map::default(); + map.insert(&[1], a.socket.local_addr().unwrap()); + map.insert(&[2], b_addr); + map.insert(&[3], forged.socket.local_addr().unwrap()); + peers.set(map); + assert!(forged.route(&short(2), client, None, None, || false)); + assert_eq!(recv(&mut b).await, None); + + // Server 1 sending server 2 a packet of its own, as a stale map would. + let len = a.seal(&short(1), client, None, None).unwrap(); + a.send.send_to(&a.buf[..len], b_addr).unwrap(); + assert_eq!(recv(&mut b).await, None); + + // Still open to a real forward. + assert!(a.route(&short(2), client, None, None, || false)); + assert!(recv(&mut b).await.is_some()); + } +} diff --git a/rs/moq-tokio/src/quic/driver/mod.rs b/rs/moq-tokio/src/quic/driver/mod.rs index f88d0fe1f42..f559fbb5f26 100644 --- a/rs/moq-tokio/src/quic/driver/mod.rs +++ b/rs/moq-tokio/src/quic/driver/mod.rs @@ -47,6 +47,7 @@ pub mod connection; pub mod endpoint; +pub(crate) mod forward; mod incoming; mod mutex; pub mod recv; diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index da2a32cbd10..64da9a12878 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -330,11 +330,16 @@ impl EndpointFactory { self.endpoint_with_config(transport::Config::default()) } - fn endpoint_with_config(&self, transport_config: transport::Config) -> Endpoint { + fn server_config(&self, transport_config: transport::Config) -> server::Config { let key = PrivateKeyDer::Pkcs8(self.cert.signing_key.serialize_der().into()); let mut server_config = crate::quic::driver::server::Config::with_single_cert(vec![self.cert.cert.der().clone()], key).unwrap(); - server_config.transport = transport_config.clone(); + server_config.transport = transport_config; + server_config + } + + fn endpoint_with_config(&self, transport_config: transport::Config) -> Endpoint { + let server_config = self.server_config(transport_config.clone()); let mut roots = RootCertStore::empty(); roots.add(self.cert.cert.der().clone()).unwrap(); @@ -1372,6 +1377,135 @@ async fn remote_address_follows_migration() { client_conn.close(0u32.into(), "done"); } +/// Three servers sit behind one front address. A connection whose handshake reached the first +/// keeps working while the front sends its packets to the second, then the third: each forwards +/// them to the first, which replies to the client directly. +#[tokio::test] +async fn forwarded_packets_reach_their_owner() { + use crate::quic::driver::forward::{Forward, Map, Peers, ServerIdGenerator}; + + let _guard = subscribe(); + let localhost = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let factory = EndpointFactory::new(); + let key = [7; 32]; + + let sockets: Vec = (0..3).map(|_| UdpSocket::bind(localhost).unwrap()).collect(); + let peers = Peers::default(); + let mut map = Map::default(); + for (id, socket) in (1u8..).zip(&sockets) { + map.insert(&[id], socket.local_addr().unwrap()); + } + peers.set(map); + let servers: Vec = (1u8..) + .zip(sockets) + .map(|(id, socket)| { + let mut config = factory.endpoint_config.clone(); + config.cid_generator = Some(Arc::new(move || Box::new(ServerIdGenerator::new(&[id], 4)))); + let forward = Forward::new(socket, &[id], peers.clone(), &key, config.max_udp_payload_size).unwrap(); + Endpoint::forwarding( + config.with_server(factory.server_config(transport::Config::default())), + UdpSocket::bind(localhost).unwrap(), + forward, + ) + .unwrap() + }) + .collect(); + + // The front end: the client's packets go to whichever server it names, from one address, and + // a reply from any server goes back to the client. + let front = tokio::net::UdpSocket::bind(localhost).await.unwrap(); + let front_addr = front.local_addr().unwrap(); + let back = tokio::net::UdpSocket::bind(localhost).await.unwrap(); + let back_addr = back.local_addr().unwrap(); + let target = Arc::new(Mutex::new(servers[0].local_addr())); + let front_task = tokio::spawn({ + let target = target.clone(); + async move { + let (mut up, mut down) = ([0; 1500], [0; 1500]); + let mut client = None; + loop { + tokio::select! { + received = front.recv_from(&mut up) => { + let (len, from) = received.unwrap(); + client = Some(from); + let to = *target.lock().unwrap(); + back.send_to(&up[..len], to).await.unwrap(); + } + received = back.recv_from(&mut down) => { + let (len, _) = received.unwrap(); + front.send_to(&down[..len], client.unwrap()).await.unwrap(); + } + } + } + } + }); + + let owner = servers[0].clone(); + let server_task = tokio::spawn(async move { + let conn = owner.accept().await.unwrap().accept().unwrap().await.unwrap(); + loop { + let Ok(stream) = conn.accept_bi().await else { break }; + // The client's packets keep arriving from the address its handshake took, whichever + // server they reached first. + assert_eq!(conn.remote_address(), back_addr); + echo(stream).await; + } + }); + + let client = factory.endpoint(); + let conn = client.dial(front_addr, "localhost").unwrap().await.unwrap(); + for (round, server) in [0, 1, 2, 0].into_iter().enumerate() { + *target.lock().unwrap() = servers[server].local_addr(); + let message = format!("round {round} through server {server}"); + let (mut send, mut recv) = conn.open_bi(Default::default()).await.unwrap(); + send.write_all(message.as_bytes()).await.unwrap(); + let _ = send.finish(); + let echoed = timeout(Duration::from_secs(10), recv.read_to_end(1024)) + .await + .expect("the owner answered") + .unwrap(); + assert_eq!(echoed, message.as_bytes()); + } + assert_eq!(servers[1].open_connections(), 0); + assert_eq!(servers[2].open_connections(), 0); + + conn.close(0u32.into(), "done"); + server_task.await.unwrap(); + front_task.abort(); +} + +/// Closing a forwarding endpoint frees its forward port with its main one, even while a handle +/// keeps the endpoint alive, so a listener restarted on the same configuration can bind both. +#[tokio::test] +async fn close_releases_the_forward_socket() { + use crate::quic::driver::forward::{Forward, Peers}; + + let _guard = subscribe(); + let localhost = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let factory = EndpointFactory::new(); + let forward_socket = UdpSocket::bind(localhost).unwrap(); + let forward_addr = forward_socket.local_addr().unwrap(); + let config = factory.endpoint_config.clone(); + let forward = Forward::new( + forward_socket, + &[1], + Peers::default(), + &[7; 32], + config.max_udp_payload_size, + ) + .unwrap(); + let endpoint = Endpoint::forwarding( + config.with_server(factory.server_config(transport::Config::default())), + UdpSocket::bind(localhost).unwrap(), + forward, + ) + .unwrap(); + + endpoint.close(0u32.into(), "done").await; + UdpSocket::bind(forward_addr).expect("the forward port is free once close returns"); + drop(endpoint); +} + /// A send group from another connection panics where it is passed in, without poisoning the /// connection's lock: a poisoned lock panics again in every handle's drop, aborting the process. #[tokio::test] diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs index 3b2c565674f..4a51f388826 100644 --- a/rs/moq-tokio/src/settings.rs +++ b/rs/moq-tokio/src/settings.rs @@ -321,4 +321,21 @@ struct ListenLb { #[usage(env = "MOQ_LISTEN_QUIC_LB_NONCE", cli("--listen-quic-lb-nonce"))] nonce: Option, + + #[usage(flatten)] + forward: ListenLbForward, +} + +/// `[listen.lb.forward]` / `--listen-quic-lb-forward-*`. +#[derive(usage_rs::Config)] +#[usage(prefix = "listen.lb.forward")] +struct ListenLbForward { + #[usage(env = "MOQ_LISTEN_QUIC_LB_FORWARD_BIND", cli("--listen-quic-lb-forward-bind"))] + bind: Option, + + #[usage(env = "MOQ_LISTEN_QUIC_LB_FORWARD_KEY", cli("--listen-quic-lb-forward-key"))] + key: Option, + + #[usage(env = "MOQ_LISTEN_QUIC_LB_FORWARD_PEERS", cli("--listen-quic-lb-forward-peers"))] + peers: Option, }