diff --git a/doc/bin/relay/auth.md b/doc/bin/relay/auth.md index 5b624bc874..9fe0373f35 100644 --- a/doc/bin/relay/auth.md +++ b/doc/bin/relay/auth.md @@ -52,9 +52,10 @@ refuses the grant), `expires` (optional unix seconds; the session closes then), `revalidate` (optional seconds until the relay asks again), `tier` (optional label handed to -[stats](/bin/relay/config#stats)), and `peer` (optional; `true` marks another +[stats](/bin/relay/config#stats)), `peer` (optional; `true` marks another relay, so what it announces counts as entering the cluster elsewhere, not as -ingest here). A 2xx with a grant admits. A 401 or 403 refuses. +ingest here), and `upstream` (optional; `true` marks that peer as an +[upstream link](/bin/relay/cluster#upstream-links), and requires `peer`). A 2xx with a grant admits. A 401 or 403 refuses. Anything else at connect, a timeout, a 5xx, or an unparseable body, refuses and logs an error; nothing is admitted because the server was down. A grant that names nothing refuses, and one with `revalidate` but no `expires` is refused @@ -66,7 +67,7 @@ clock, so later polls, outages, and wall-clock adjustments do not restart it. **Revalidate and outage.** On the cadence the relay POSTs `revalidate` with the same request. A grant applies: a changed `root` or `mounts`, or one that no longer covers what the session holds closes it with `Unauthorized` (the live session is not -resized in place), and so does a flipped `peer`; a changed `tier` keeps the +resized in place), and so does a flipped `peer` or `upstream`; a changed `tier` keeps the session and moves its stats: its presence counts under the new tier from then on, as does each group and subscription it starts afterwards, while one already in flight finishes where it began. A 401 or 403 closes the session now, as does a 2xx @@ -384,7 +385,12 @@ bind = "/run/moq/internal.sock" allow.uid = [1001] ``` -Bind TCP to loopback or a private interface; it carries no peer identity. The -Unix socket is created mode `0666`, so gate it with a restrictive parent -directory or an explicit allowlist. +Bind plaintext TCP to loopback or a private interface; it carries no peer +identity. The Unix socket is created mode `0666`, so gate it with a restrictive +parent directory or an explicit allowlist. These are native-only paths for gateways and stats publishers on the same host. + +`listen.tcp.tls = true` serves the TCP listener over TLS with the listen +certificate instead, for `tls://` dials such as [cluster links](/bin/relay/cluster#tls-links) +that need neither QUIC nor a WebSocket. It still carries no peer identity: it +asks for no client certificate, so a peer on it presents a token, not mTLS. diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index 635f7bbfcb..75d1cd1bee 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -75,6 +75,58 @@ For a full mesh, list every other relay, or serve the list from `connect_api`. A session carries both directions, so one dial per pair is enough; listing a pair on both sides opens a redundant second session. +## Upstream links + +Mark a link `upstream` and this relay never offers it a route learned on +another upstream link. Everything else still transits, and clients still see +every route the relay knows. An edge marks its region's cores upstream, so it +reaches every core and every core reaches its clients, but it never carries +traffic from one core to another. A drone marks its CDN uplink upstream, so its +mesh reaches the CDN and the CDN reaches the mesh, but the drone never carries +traffic between CDN relays. + +```toml +# edge.toml +[cluster] +connect = [ + { url = "https://core-a.example.com/", upstream = true }, + { url = "https://core-b.example.com/", upstream = true }, +] +``` + +The mark belongs to the link, whichever side dialed. A peer this relay dials +is upstream when its `connect` or `connect_api` entry says so. A peer that +dials in is upstream when its [grant](/bin/relay/auth#the-contract) sets +`"upstream": true` beside `"peer": true`, so an edge whose auth server grants +that to core certificates treats a core that dials it as upstream too. A +relay that predates the mark treats every link as transit, so a cluster +migrates one region at a time. + +Which relay dials which is still the peer list's job: there are no roles and +no topology check, so whatever generates the list applies the layout's rules. +Hiding a core from clients is admission, not routing: its auth refuses any +session that is not a cluster peer. + +## TLS links + +A link inside a region is rarely congested, so it can skip QUIC: a `tls://` +peer URL dials qmux over TLS on TCP, verified with the same `connect.tls` +settings as any other dial. The accepting relay serves it from its TCP +listener with TLS on. That listener asks for no client certificate, so a peer +on it authenticates with a token (`cluster.token`, `token`, or `?jwt=`), not +mTLS. + +```toml +# core.toml +[listen.tcp] +bind = "[::]:4443" +tls = true + +# edge.toml +[cluster] +connect = [{ url = "tls://core-a.internal:4443/", upstream = true }] +``` + ## Link costs Add `?cost=N` to a peer URL to route by price instead of hop count. An unpriced @@ -101,7 +153,7 @@ connect = ["https://sibling.same-dc/?cost=0", "https://us-east.example.com/?cost ``` The same policy reads as an object, which is the only form that accepts -`egress` and `token`. A bare URL stays valid, and an object whose `url` still +`egress`, `token`, and [`upstream`](#upstream-links). A bare URL stays valid, and an object whose `url` still carries `?cost=` or `?jwt=` alongside those fields is rejected rather than given a precedence a migration could silently get wrong. diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 0524a10e71..b224436daa 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -29,6 +29,7 @@ root = ["peer-ca.pem"] # Optional: CAs for client certs (mTLS), re [listen.tcp] # Plaintext qmux over TCP for trusted local workers. bind = "127.0.0.1:4444" +# tls = true # Or: qmux over TLS (tls://) with the listen certificate, no client certs. [listen.unix] # Plaintext qmux over a Unix socket, gated by peer credentials. bind = "/run/moq/internal.sock" @@ -151,7 +152,7 @@ See [Authentication](/bin/relay/auth). ```toml [cluster] -connect = ["https://us-east.example.com/?cost=10"] # Peers to dial. ?cost prices the link, or use {url, cost, egress, token} objects. +connect = ["https://us-east.example.com/?cost=10"] # Peers to dial. ?cost prices the link, or use {url, cost, egress, token, upstream} objects. node = "https://us-west.example.com/" # This relay's own URL. connect_api = "https://api.example.com/peers" # Or fetch the peer list (JSON array of URLs and/or objects) live. token = "cluster.jwt" # JWT for dials without an inline ?jwt=. diff --git a/doc/lib/rs/moq-auth.md b/doc/lib/rs/moq-auth.md index 6375356ca5..5060f15d3b 100644 --- a/doc/lib/rs/moq-auth.md +++ b/doc/lib/rs/moq-auth.md @@ -13,7 +13,7 @@ Everything a party needs to ask for or answer an authorization on answers the relay, in a service that mints tokens for clients, or in your own accept loop that decides in process. -- **Request and grant**: `Request` is the JSON a relay POSTs per session event (`connect`, `revalidate`, `end`) with everything it knows: id, node, transport, addresses, SNI and ALPN, the raw path and query, the moq-transport SETUP `token` (its Token Type and bytes, base64url on the wire), the declared role, and the verified certificate facts. `Grant` is the answer: `publish` and `subscribe` pattern unions, an optional `root` alias, `mounts` that read a subtree from elsewhere on the origin, `expires`, `revalidate`, `tier`, and `peer`, which marks the session as a cluster peer so the routes it announces report `Source::Peer`. `Grant::validate` refuses a grant that names nothing, asks to be revalidated without a bound or at no interval, or has already expired; expiry is exact, with no grace for clock skew. +- **Request and grant**: `Request` is the JSON a relay POSTs per session event (`connect`, `revalidate`, `end`) with everything it knows: id, node, transport, addresses, SNI and ALPN, the raw path and query, the moq-transport SETUP `token` (its Token Type and bytes, base64url on the wire), the declared role, and the verified certificate facts. `Grant` is the answer: `publish` and `subscribe` pattern unions, an optional `root` alias, `mounts` that read a subtree from elsewhere on the origin, `expires`, `revalidate`, `tier`, `peer`, which marks the session as a cluster peer so the routes it announces report `Source::Peer`, and `upstream`, which marks that peer an [upstream link](/bin/relay/cluster#upstream-links). `Grant::validate` refuses a grant that names nothing, marks a non-peer upstream, asks to be revalidated without a bound or at no interval, or has already expired; expiry is exact, with no grace for clock skew. - **Lease**: `lease::Producer` and `lease::Consumer` are the handle a session holds for its grant. The consumer reads the current grant, waits for a change, and learns why the lease ended; the producer updates and revokes. Either side's terminal call returns the reason the lease actually ended with, so whichever got there first is what both report. `Consumer::fixed` is a grant nobody drives. `Consumer::revalidate` nudges a re-check now and the producer observes it via `poll_revalidate` or `revalidate_requested`, which is how the relay's session push lands. Whoever runs the accept loop builds the producer, so an embedder decides in process with no trait and no HTTP. Enforcing `expires` is the holder's job; the `Client` driver also revokes at expiry so its `end` event goes out. `Grant::deadline()` (feature `tokio`, also enabled by `client` and `serve`) snapshots expiry on Tokio's clock. Call it once per accepted grant and retain the deadline; an expiry already past is now. - **Client**: `Client::new(url, tls)` and `Client::connect(request)` drive a lease against an auth server over `https://`, `unix://`, or loopback `http://`: revalidate on cadence with jittered backoff through an outage until `expires`, revoke on a 401/403 or an invalid grant, and POST `end` with the reason, duration, and byte totals the session reported through `lease::Consumer::close` when it ended. Dropping the consumer reports zero bytes. `end.reason` is `dropped`, `expired`, `refused`, `invalid`, `narrowed`, `shutdown`, or the session's own classification. - **Server**: `serve::Policy` and `serve::Server` (feature `serve`) are the reference auth server behind `moq auth serve`: a JWT from the `jwt` query or a type-0 SETUP token, verified against a key file, a `{kid}.jwk` directory, or a JWK Set file (a token of another type, or two different ones, is refused), an explicit grant for verified certificates, the anonymous permissions, a tier, the revalidation cadence, a default `expires`, and live session caps per token and per remote address. A token is authorized at the dialed path with `Claims::authorize`; residuals become the grant. `Server::router` is an axum `POST /` you can mount in your own service. diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 5fa5089f8b..94932fdbea 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -23,7 +23,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this. - **Groups** are written frame by frame and delivered on independent streams. Old groups are cached for fetch-by-sequence; stale groups are skipped per the subscriber's budget. - **Track ends**: `finish()` ends a track at its live edge, while `finish_at(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber awaits it with `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF. - **Datagrams** send a single small frame unreliably on moq-lite 05+ and moq-transport. -- **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one. `Route::source()` says where a delivered route entered: `Source::Local`, or `Source::Peer(hop)` when a handle marked `origin::Producer::peer()` announced it. `origin::Consumer::local()` sees only the local ones. +- **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one. `Route::source()` says where a delivered route entered: `Source::Local`, or `Source::Peer(hop)` when a handle marked `origin::Producer::peer()` announced it. `origin::Consumer::local()` sees only the local ones. A handle marked `origin::Producer::upstream()` announces as a peer, and a consumer taken from it never sees a route another upstream handle announced, so a relay never transits between two upstream links. - **Stats** counters per broadcast and session, drained by [`moq-stats`](https://docs.rs/moq-stats). It runs over anything implementing `web_transport_trait::poll::Session`: noq, the diff --git a/js/auth/src/contract.ts b/js/auth/src/contract.ts index 19b31a9603..39e4ff238e 100644 --- a/js/auth/src/contract.ts +++ b/js/auth/src/contract.ts @@ -137,6 +137,8 @@ export const GrantSchema = z tier: z.optional(z.string()), /** The session is a cluster peer (another relay): what it announces entered the cluster elsewhere. */ peer: z.optional(z.boolean()), + /** The peer is upstream: the relay never offers it a route learned from another upstream. Requires `peer`. */ + upstream: z.optional(z.boolean()), }) .check( z.refine((grant) => (grant.publish?.length ?? 0) > 0 || (grant.subscribe?.length ?? 0) > 0, { @@ -145,5 +147,8 @@ export const GrantSchema = z z.refine((grant) => grant.revalidate === undefined || grant.expires !== undefined, { message: "a grant that asks to be revalidated must expire", }), + z.refine((grant) => !grant.upstream || grant.peer === true, { + message: "an upstream grant must be a peer", + }), ); export type Grant = z.infer; diff --git a/js/auth/src/interop.test.ts b/js/auth/src/interop.test.ts index 85c10ad522..bd2b4b3d22 100644 --- a/js/auth/src/interop.test.ts +++ b/js/auth/src/interop.test.ts @@ -261,7 +261,7 @@ describe("Rust-serialized contract", () => { const RUST_INVALID_END = '{"id":"00ff","event":"end","reason":"invalid","duration":1.5,"bytes":{"sent":10,"received":20},"node":"relay-1","transport":"websocket","remote":"203.0.113.9:4433","path":"/demo/room","query":"jwt=abc"}'; const RUST_GRANT = - '{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true}'; + '{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true,"upstream":true}'; test("parse a request a Rust relay sent", () => { const request = RequestSchema.parse(JSON.parse(RUST_REQUEST)); @@ -298,6 +298,7 @@ describe("Rust-serialized contract", () => { revalidate: 60, tier: "websocket", peer: true, + upstream: true, }); expect(JSON.stringify(grant)).toBe(RUST_GRANT); }); diff --git a/quest/m1/cluster-routing/README.md b/quest/m1/cluster-routing/README.md index c562957913..61fa4232f7 100644 --- a/quest/m1/cluster-routing/README.md +++ b/quest/m1/cluster-routing/README.md @@ -51,8 +51,9 @@ vector the 2026-09-30 cache-tiers audit kept: routes learned on another upstream link, and a route learned upstream keeps a down-only mark across other links so a mesh with two uplinks cannot leak CDN routes back into the CDN. An edge marks its core links upstream; a - Starlink drone marks its CDN link upstream. See - [Upstream links](/quest/m1/cluster-routing/transit.md). + Starlink drone marks its CDN link upstream. The per-link mark landed on + today's path vector (`doc/bin/relay/cluster.md`); the down-only mark lands + with [Routes and announces](/quest/m1/cluster-routing/routes.md). - **One node id space, no cluster ids.** A customer cluster is nodes behind upstream or trusted links. Loop detection by cluster id would drop a partitioned swarm's traffic to itself through the CDN (BGP's partitioned-AS @@ -127,7 +128,6 @@ Once every child has landed: ## Required -- [Upstream links](/quest/m1/cluster-routing/transit.md) - a link marked upstream never receives routes learned on another upstream link, which builds edge tiers and drone uplinks from one rule - [Selection](/quest/m1/cluster-routing/selection.md) - a broadcast under overlapping prefixes routes to one origin deterministically, and same-path origins are one source - [Multi-CDN endpoints](/quest/m1/cluster-routing/multi-cdn.md) - an endpoint holds sessions to several CDNs, uses its preferred one, and fails over to the next - [Simulate the split](/quest/m1/cluster-routing/sim.md) - moq.pro's simulator compares the route layer with path vector before the wire is written diff --git a/quest/m1/cluster-routing/routes.md b/quest/m1/cluster-routing/routes.md index 3e656df8ed..5dfd41668c 100644 --- a/quest/m1/cluster-routing/routes.md +++ b/quest/m1/cluster-routing/routes.md @@ -67,7 +67,7 @@ Decided 2026-10-01 (see the [line's decisions](/quest/m1/cluster-routing/README. No per-announce seqno. - Down-only bit: set on a route learned on an upstream link, kept across other links, and a route carrying it is never sent on an upstream link - ([Upstream links](/quest/m1/cluster-routing/transit.md)). + (extending the `upstream` link mark in `doc/bin/relay/cluster.md`). - A plain client with one link advertises a ROUTE for itself and the ANNOUNCEs it publishes; its node id is scoped to its session (shared identity across sessions is [Route trust](/quest/m3/route-trust.md)). diff --git a/quest/m1/cluster-routing/selection.md b/quest/m1/cluster-routing/selection.md index 7cf5d58a6b..53c7ba7ef0 100644 --- a/quest/m1/cluster-routing/selection.md +++ b/quest/m1/cluster-routing/selection.md @@ -59,5 +59,4 @@ no timestamp rewind. ## Required -- [Upstream links](/quest/m1/cluster-routing/transit.md) - the edges and cores this spreads over - [Wildcard](/quest/m0/wildcard/README.md) - the longest-prefix rule and pool spread this builds on diff --git a/quest/m1/cluster-routing/transit.md b/quest/m1/cluster-routing/transit.md deleted file mode 100644 index 46bc11f2f6..0000000000 --- a/quest/m1/cluster-routing/transit.md +++ /dev/null @@ -1,64 +0,0 @@ -# [M] Upstream links - -## Goal - -A cluster link can be marked upstream, and a relay never re-advertises a -route learned on one upstream link to another upstream link. Everything else -still transits. That one rule builds moq.pro's edge tier (an edge marks its -region's cores upstream, so it is never transit between them) and a drone -uplink (a Starlink drone marks its CDN link upstream: the mesh reaches the -CDN and the CDN reaches the mesh, but the drone never carries CDN traffic -between CDN relays). Edge, core, and one-node region are an operator's names -for configurations, not relay modes. Clients still see every route a relay -knows. - -## Plan - -Decided 2026-10-01, replacing the per-node `cluster.transit` setting (#4683) -with a per-link attribute, because a drone must transit for its mesh but never -for its CDN uplink, which no per-node switch can express (BGP's -customer/provider export rule, RFC 9234, reduced to one bit): - -- `upstream` is an attribute of a peer entry, from config or the connect - API's peer list, off by default. It applies to the link whichever side - dialed: an edge that is dialed by a core still treats that core as - upstream, so accepted peer sessions need a way to be marked (the peer's - grant classification or identity is the likely source; pick the simplest - shape that covers the reverse-dial test). -- An upstream link is sent only routes that did not arrive on another - upstream link. On today's path vector that is the routes that entered - locally or over non-upstream links. -- The down-only mark that stops a two-uplink mesh from leaking CDN routes - back into the CDN needs a wire bit, so it lands with - [Routes and announces](/quest/m1/cluster-routing/routes.md), not here. -- No roles and no topology check. Who dials whom is the peer list, and - whatever generates it applies the operator's rules (an edge dials its - region's cores, cores dial only lower-named cores of linked regions); the - generator is where that is tested. -- Refusing clients on a hidden core is admission, not routing: the auth - server or embedder refuses a session that is not a cluster peer. -- Relays from before this attribute treat every link as transit, so a cluster - migrates one region at a time with no legacy mode. -- Spreading paths over equally ranked upstream links by the path-keyed - rendezvous hash is [Selection](/quest/m1/cluster-routing/selection.md)'s; - it is what makes every edge of a region send a given path to the same core. -- Edge links run qmux over TLS on TCP, not WebSocket and not QUIC: - intra-region links are not congested. qmux over TCP exists today only in - plaintext (`tcp://`, moq-tokio's `tcp.rs`); add `tls://` beside it on - moq-tokio's `tls` helpers. (Scheme name decided 2026-09-30.) The closed - [#4731](https://github.com/moq-dev/moq/pull/4731) is prior art for the - `tls://` qmux piece in moq-tokio; its edge/core roles are not wanted. - -Test: a two-region cluster in `rs/moq-relay/tests`. One region has two edges -whose links to two cores are upstream; the other is one relay that also -serves clients. Subscribers on both edges cause one cross-region subscription -per broadcast, a route from one core never reaches the other through an edge, -a client on the far relay gets a broadcast published on an edge, a core that -dials an edge still gets no routes through it, and losing a core moves only -its paths. Add a three-node line (mesh node, uplink node, CDN relay) where the -mesh node's broadcast reaches the CDN and a CDN broadcast reaches the mesh -node. - -Public API: an `upstream` peer attribute and a TLS qmux URL scheme. -`cluster.tier` is unrelated: it stays the billing and stats label. -Wire: none. diff --git a/rs/moq-auth/src/error.rs b/rs/moq-auth/src/error.rs index 6825f06308..3a6e154f35 100644 --- a/rs/moq-auth/src/error.rs +++ b/rs/moq-auth/src/error.rs @@ -99,6 +99,9 @@ pub enum Error { #[error("grant names nothing; the session is refused")] UselessGrant, + #[error("grant marks an upstream that is not a peer")] + UpstreamWithoutPeer, + #[error("grant asks to be revalidated but never expires")] UnboundedRevalidate, diff --git a/rs/moq-auth/src/grant.rs b/rs/moq-auth/src/grant.rs index bb414a56c0..9142d58664 100644 --- a/rs/moq-auth/src/grant.rs +++ b/rs/moq-auth/src/grant.rs @@ -49,6 +49,12 @@ pub struct Grant { /// the cluster elsewhere, not here. #[serde(skip_serializing_if = "std::ops::Not::not")] pub peer: bool, + + /// The cluster peer is upstream: the relay never offers it a route learned + /// from another upstream peer, so it never carries traffic between two. + /// Requires `peer`. + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub upstream: bool, } impl Grant { @@ -69,12 +75,15 @@ impl Grant { tokio::time::Instant::now().checked_add(remaining) } - /// Refuse a grant that admits nothing, asks to be revalidated without a bound or - /// at no interval, or has already expired. + /// Refuse a grant that admits nothing, marks a non-peer upstream, asks to be + /// revalidated without a bound or at no interval, or has already expired. pub fn validate(&self) -> crate::Result<()> { if self.publish.is_empty() && self.subscribe.is_empty() { return Err(crate::Error::UselessGrant); } + if self.upstream && !self.peer { + return Err(crate::Error::UpstreamWithoutPeer); + } if self.revalidate.is_some() && self.expires.is_none() { return Err(crate::Error::UnboundedRevalidate); } @@ -108,12 +117,14 @@ mod tests { revalidate: Some(Duration::from_secs(60)), tier: Some("websocket".into()), peer: true, + upstream: true, }; let json = serde_json::to_value(&grant).unwrap(); assert_eq!(json["expires"], 4_102_444_800_i64); assert_eq!(json["revalidate"], 60); assert_eq!(json["publish"], serde_json::json!(["alice/**"])); assert_eq!(json["peer"], true); + assert_eq!(json["upstream"], true); assert_eq!(serde_json::from_value::(json).unwrap(), grant); } @@ -130,10 +141,11 @@ mod tests { revalidate: Some(Duration::from_secs(60)), tier: Some("websocket".into()), peer: true, + upstream: true, }; assert_eq!( serde_json::to_string(&grant).unwrap(), - r#"{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true}"# + r#"{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true,"upstream":true}"# ); } @@ -177,6 +189,15 @@ mod tests { assert!(matches!(grant.validate(), Err(crate::Error::ZeroRevalidate))); } + #[test] + fn validate_refuses_an_upstream_that_is_not_a_peer() { + let mut grant = Grant::new(patterns(&["**"]), Patterns::new()); + grant.upstream = true; + assert!(matches!(grant.validate(), Err(crate::Error::UpstreamWithoutPeer))); + grant.peer = true; + grant.validate().unwrap(); + } + #[cfg(feature = "tokio")] #[tokio::test(start_paused = true)] async fn deadline_is_the_exact_expiry() { diff --git a/rs/moq-auth/src/request.rs b/rs/moq-auth/src/request.rs index 2e7d20504e..1c5bcfd78c 100644 --- a/rs/moq-auth/src/request.rs +++ b/rs/moq-auth/src/request.rs @@ -135,7 +135,7 @@ pub enum Transport { Iroh, /// A WebSocket connection using qmux framing. WebSocket, - /// A plaintext TCP connection using qmux framing. + /// A TCP connection using qmux framing, plaintext or TLS. Tcp, /// A Unix domain socket using qmux framing. Unix, diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 00cd050446..040198ea1c 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -858,9 +858,8 @@ struct RouteEntry { /// Whether this is a broadcast published on this origin, which wins a cost tie and /// keeps its own cache. local: bool, - /// Whether a handle marked [`Producer::peer`] inserted the entry, so it - /// entered from a cluster peer rather than here. - peer: bool, + /// The link the entry arrived on, from the handle that inserted it. + link: Link, /// The queue requests under this route are served from, when the announcer /// serves content on demand (a [`Dynamic`]). `None` for an advertise-only /// announcement ([`Producer::announce`]) and for a local broadcast. @@ -897,9 +896,9 @@ impl RouteEntry { /// Where the entry entered this origin. fn entered(&self) -> Source { - match self.peer { - true => Source::Peer(self.via), - false => Source::Local, + match self.link { + Link::Local => Source::Local, + Link::Peer | Link::Upstream => Source::Peer(self.via), } } @@ -977,15 +976,33 @@ struct Horizon { exclude: Option, /// Hide the routes that entered from a cluster peer ([`Consumer::local`]). local: bool, + /// Hide the routes learned on an upstream link, because this reader is one + /// ([`Producer::upstream`]): a relay never transits between two upstreams. + upstream: bool, } impl Horizon { /// Whether `entry` may be observed or served through this horizon. fn admits(&self, entry: &RouteEntry) -> bool { - !(self.local && entry.peer) && entry.visible_to(self.exclude) + !(self.local && entry.link != Link::Local) + && !(self.upstream && entry.link == Link::Upstream) + && entry.visible_to(self.exclude) } } +/// The link a handle's routes arrive on; see [`Producer::peer`] and +/// [`Producer::upstream`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] +enum Link { + /// Here: an in-process producer or a client session. + #[default] + Local, + /// A cluster peer. + Peer, + /// A cluster peer marked upstream. + Upstream, +} + /// One remotely-served front in [`OriginState::fronts`]: the shared broadcast at a /// path plus the channel requesters resolve through. #[derive(Clone)] @@ -1407,9 +1424,9 @@ pub struct Producer { // session tagged this handle via [`Self::with_stats`]. stats: stats::Session, - // Whether routes announced through this handle entered from a cluster peer - // (see [`Self::peer`]). - peer: bool, + // The link routes announced through this handle arrive on (see + // [`Self::peer`] and [`Self::upstream`]). + link: Link, // Submission handle to the origin's [`Driver`]: source watchers, fronts, and // serve tasks queued here run when the driver is polled. Closed once the @@ -1441,7 +1458,7 @@ impl Producer { pool: config.pool, cache_duration: config.cache_duration, stats: stats::Session::default(), - peer: false, + link: Link::Local, tasks, timers: timers.clone(), }; @@ -1473,7 +1490,22 @@ impl Producer { /// entered here from what a peer forwarded. The hop chain cannot: a client /// and a peer each append one hop. pub fn peer(mut self) -> Self { - self.peer = true; + if self.link == Link::Local { + self.link = Link::Peer; + } + self + } + + /// Mark this handle (and any handle derived from it) as an upstream peer's: + /// a [`Self::peer`] whose routes are never offered to another upstream. + /// A [`Consumer`] taken from it hides every route an upstream handle + /// announced, so this origin never carries traffic between two upstreams. + /// + /// Hand it to a session with a peer that should reach everything here + /// without this relay carrying its traffic to other upstreams, such as an + /// edge's link to a core or a drone's link to its CDN. + pub fn upstream(mut self) -> Self { + self.link = Link::Upstream; self } @@ -1508,7 +1540,7 @@ impl Producer { pool: cache::Pool::default(), cache_duration: Duration::MAX, stats: stats::Session::default(), - peer: false, + link: Link::Local, tasks, timers: Clock::default(), } @@ -1575,7 +1607,7 @@ impl Producer { prefixes: vec![(full.clone(), claim)], scope: self.scope.allowed.clone(), local: true, - peer: self.peer, + link: self.link, stats: self.stats.clone(), }; let info = broadcast::Info { @@ -1731,7 +1763,7 @@ impl Producer { pool: self.pool.clone(), cache_duration: self.cache_duration, stats: self.stats.clone(), - peer: self.peer, + link: self.link, tasks: self.tasks.clone(), timers: self.timers.clone(), }) @@ -1843,8 +1875,8 @@ struct Announcing { /// The absolute paths the producer is authorized to serve. scope: Patterns, local: bool, - /// Whether the producer was marked [`Producer::peer`]. - peer: bool, + /// The link the producer's routes arrive on. + link: Link, stats: stats::Session, } @@ -1866,7 +1898,7 @@ impl Announcing { prefixes: vec![(requested, claim)], scope: producer.scope.allowed.clone(), local: false, - peer: producer.peer, + link: producer.link, stats: producer.stats.clone(), }) } @@ -1898,7 +1930,7 @@ impl Announcing { cost: route.cost, via, local: self.local, - peer: self.peer, + link: self.link, server: serving.server.clone(), source: serving.source.clone(), advertised: serving.advertised, @@ -3827,7 +3859,8 @@ pub struct Consumer { // Split horizon: routes whose hop chain or announcing session (`via`) is the // excluded peer are invisible to `announced` and skipped by // `request_broadcast`, so a peer is never served (or advertised) its own - // content back. A local view (`Self::local`) hides peer routes the same way. + // content back. A local view (`Self::local`) hides peer routes the same way, + // and a view taken from an upstream handle hides upstream routes. horizon: Horizon, // Which routes beneath a hidden (`.`-prefixed) segment `announced` reports. @@ -3856,7 +3889,10 @@ impl Consumer { root: producer.root.clone(), shared: producer.shared.clone(), stats, - horizon: Horizon::default(), + horizon: Horizon { + upstream: producer.link == Link::Upstream, + ..Horizon::default() + }, hidden: Hidden::default(), pool: producer.pool.clone(), cache_duration: producer.cache_duration, @@ -3883,7 +3919,7 @@ impl Consumer { } /// A view of the routes that entered here: every route a handle marked - /// [`Producer::peer`] announced is hidden from [`Self::announced`] and never + /// [`Producer::peer`] or [`Producer::upstream`] announced is hidden from [`Self::announced`] and never /// resolved by [`Self::request_broadcast`]. /// /// On a relay, this is what the relay ingests itself, from clients and @@ -6815,6 +6851,70 @@ mod tests { ); } + /// An upstream link is offered what entered here and what other peers + /// forwarded, but never a route learned on another upstream link, and the + /// best route it sees skips a better upstream one. + #[tokio::test] + async fn upstream_view_hides_upstream_routes() { + let producer = origin(1).produce(); + let peer = producer.clone().peer(); + let core = producer.clone().upstream(); + let mut toward_core = core.consume().announced(); + let mut everyone = producer.consume().announced(); + + let _core = core + .dynamic("core", Route::default().with_hops(hops(&[7])).with_via(origin(7))) + .unwrap(); + assert_eq!(everyone.assert_next_active("core").source(), Source::Peer(origin(7))); + toward_core.assert_next_wait(); + + let _mesh = peer + .dynamic("mesh", Route::default().with_hops(hops(&[8])).with_via(origin(8))) + .unwrap(); + assert_eq!(toward_core.assert_next_active("mesh").source(), Source::Peer(origin(8))); + + // A path both an upstream and a mesh peer reach: the upstream route is + // cheaper, yet the upstream view selects the mesh peer's. + let _cheap = core + .dynamic("both", Route::default().with_hops(hops(&[7])).with_via(origin(7))) + .unwrap(); + let mesh = peer + .dynamic( + "both", + Route::default().with_hops(hops(&[8])).with_via(origin(8)).with_cost(9), + ) + .unwrap(); + assert_eq!(toward_core.assert_next_active("both").source(), Source::Peer(origin(8))); + drop(mesh); + toward_core.assert_next_ended("both"); + + // `peer` never demotes an upstream handle. + let _still = core + .clone() + .peer() + .dynamic("again", Route::default().with_hops(hops(&[7])).with_via(origin(7))) + .unwrap(); + toward_core.assert_next_wait(); + + // Resolution agrees with the cursor: an upstream-only path is unroutable + // toward an upstream, while the full view queues it on the core's route. + let err = core + .consume() + .request_broadcast("core/alice") + .now_or_never() + .expect("unroutable") + .err() + .unwrap(); + assert!(matches!(err, Error::Unroutable)); + assert!( + producer + .consume() + .request_broadcast("core/alice") + .now_or_never() + .is_none() + ); + } + /// A handler that rejects a path with `Unroutable` while its route stands /// gives the requester that answer; the front must not re-ask the same route /// forever, which would spin the origin driver. diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index f3cee4285c..8e85251ad7 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -385,6 +385,8 @@ pub struct Token { pub tier: Tier, /// Whether the session is a cluster peer, so its routes entered elsewhere. pub peer: bool, + /// Whether the peer is upstream, so it is never offered another upstream's routes. + pub upstream: bool, } impl Token { @@ -407,6 +409,7 @@ impl Token { publish: grant.publish.clone(), tier: crate::configured_tier(grant.tier.clone()), peer: grant.peer, + upstream: grant.upstream, } } @@ -480,8 +483,9 @@ impl Lease { /// /// A changed root or mounts, or a narrower grant, ends it: origin handles cannot yet narrow /// a live scope in place (tracked by `quest/m1/auth/narrowing.md`). A flipped - /// `peer` ends it too, since the routes it already announced would be - /// misreported as entering here or from a peer. A changed tier keeps the + /// `peer` or `upstream` ends it too, since the routes it already announced + /// would be misreported as entering here or from a peer, or offered to the + /// wrong links. A changed tier keeps the /// session and moves its [stats](Self::with_stats) to the new tier. pub async fn ended(&mut self) -> lease::Reason { loop { @@ -506,6 +510,9 @@ impl Lease { if fresh.peer != self.token.peer { return "peer changed".into(); } + if fresh.upstream != self.token.upstream { + return "upstream changed".into(); + } if !self.token.covered_by(&fresh) { return lease::Reason::Narrowed; } @@ -1054,7 +1061,7 @@ mod tests { /// Routes a session announced were recorded as a peer's or not; a re-check that /// flips it closes the session rather than misreport them. #[tokio::test] - async fn a_recheck_that_flips_peer_closes() { + async fn a_recheck_that_flips_the_link_closes() { let grant = Grant::new(patterns(&["**"]), patterns(&["**"])); let (producer, consumer) = lease::Producer::new(grant.clone()); let mut lease = Lease::new("/", consumer); @@ -1062,8 +1069,17 @@ mod tests { let mut peer = grant; peer.peer = true; - producer.update(peer); + producer.update(peer.clone()); assert_eq!(lease.ended().await.to_string(), "peer changed"); + + // Routes already offered to (or withheld from) the session were filtered + // by whether it is upstream, so flipping that closes it too. + let (producer, consumer) = lease::Producer::new(peer.clone()); + let mut lease = Lease::new("/", consumer); + let mut upstream = peer; + upstream.upstream = true; + producer.update(upstream); + assert_eq!(lease.ended().await.to_string(), "upstream changed"); } #[test] diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 30c5ff0b18..72ed81e7fa 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -34,9 +34,10 @@ const CONNECT_API_POLL_INTERVAL: Duration = Duration::from_secs(30); /// /// Accepts a URL string or an object with `url` plus /// policy: `cost` prices the link, `egress` declares this relay's own price to -/// the peer, and `token` carries the peer's credential. `egress` defaults to -/// `cost`; a different effective value is rejected until asymmetric routing -/// lands. An object `token` behaves exactly like an inline `?jwt=` and is +/// the peer, `token` carries the peer's credential, and `upstream` marks the +/// link upstream, so it is never offered a route learned on another upstream +/// link. `egress` defaults to `cost`; a different effective value is rejected +/// until asymmetric routing lands. An object `token` behaves exactly like an inline `?jwt=` and is /// redacted the same way. Unknown object fields are rejected, as is an object /// that sets policy while its `url` still carries `?cost=` or `?jwt=`. #[derive(Clone, PartialEq, Eq)] @@ -46,6 +47,7 @@ pub struct Peer { cost: Option, egress: Option, token: Option, + upstream: bool, } impl std::fmt::Debug for Peer { @@ -57,6 +59,7 @@ impl std::fmt::Debug for Peer { .field("cost", &self.cost) .field("egress", &self.egress) .field("token", &self.token.as_ref().map(|_| "...")) + .field("upstream", &self.upstream) .finish() } } @@ -69,6 +72,7 @@ impl Peer { cost: None, egress: None, token: None, + upstream: false, } } @@ -94,6 +98,12 @@ impl Peer { self.token.as_deref() } + /// Whether the link is upstream: it is never offered a route learned on + /// another upstream link. + pub fn upstream(&self) -> bool { + self.upstream + } + /// Price the link this peer is dialed on. pub fn with_cost(mut self, cost: u64) -> Self { self.cost = Some(cost); @@ -111,6 +121,12 @@ impl Peer { self.token = Some(token.into()); self } + + /// Mark the link upstream. + pub fn with_upstream(mut self, upstream: bool) -> Self { + self.upstream = upstream; + self + } } impl std::str::FromStr for Peer { @@ -134,6 +150,8 @@ struct PeerObject { egress: Option, #[serde(default)] token: Option, + #[serde(default)] + upstream: bool, } /// A bare string is the URL form; a map is the object form. Dispatching on the @@ -150,7 +168,7 @@ impl<'de> serde::Deserialize<'de> for Peer { type Value = Peer; fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("a peer URL string or an object with url, cost, egress, token") + f.write_str("a peer URL string or an object with url, cost, egress, token, upstream") } fn visit_str(self, url: &str) -> Result { @@ -165,6 +183,7 @@ impl<'de> serde::Deserialize<'de> for Peer { cost: object.cost, egress: object.egress, token: object.token, + upstream: object.upstream, }) } } @@ -178,7 +197,7 @@ impl serde::Serialize for Peer { where S: serde::Serializer, { - if self.cost.is_none() && self.egress.is_none() && self.token.is_none() { + if self.cost.is_none() && self.egress.is_none() && self.token.is_none() && !self.upstream { return serializer.serialize_str(&self.url); } use serde::ser::SerializeStruct as _; @@ -192,6 +211,9 @@ impl serde::Serialize for Peer { if self.token.is_some() { len += 1; } + if self.upstream { + len += 1; + } let mut state = serializer.serialize_struct("Peer", len)?; state.serialize_field("url", &self.url)?; if let Some(cost) = &self.cost { @@ -203,6 +225,9 @@ impl serde::Serialize for Peer { if let Some(token) = &self.token { state.serialize_field("token", token)?; } + if self.upstream { + state.serialize_field("upstream", &true)?; + } state.end() } } @@ -222,6 +247,8 @@ struct DialTarget { /// Present the mDNS credential on `/.cluster/` and skip /// [`Config::token`]. LAN only. lan: bool, + /// Never offer this peer a route learned on another upstream link. + upstream: bool, } impl DialTarget { @@ -276,6 +303,7 @@ impl DialTarget { cost, fingerprint: None, lan: false, + upstream: peer.upstream, }) } @@ -310,6 +338,7 @@ impl DialTarget { cost, fingerprint: peer.fingerprint.clone(), lan: true, + upstream: false, }) } } @@ -555,7 +584,7 @@ pub struct Config { /// Connect to one or more other cluster nodes. Each entry is a full URL, e.g. /// `https://host/?jwt=TOKEN`, or an object with `url`, `cost`, `egress`, - /// and `token`; see [`Peer`]. A bare host or `host:port` is refused; pass the + /// `token`, and `upstream`; see [`Peer`]. A bare host or `host:port` is refused; pass the /// full URL instead. Accepts a comma-separated list on the CLI or repeat the /// flag; in config files use a TOML array of URLs and/or objects. /// @@ -579,7 +608,7 @@ pub struct Config { /// Fetch the list of peers to dial from an HTTP(S) URL or a local file, /// reloading at runtime without a restart. The source returns a JSON array /// of peers: bare URL strings and/or objects with `url`, `cost`, `egress`, - /// and `token`, exactly as [`Self::connect`] accepts, e.g. + /// `token`, and `upstream`, exactly as [`Self::connect`] accepts, e.g. /// `["https://a.pop.example/?cost=1", {"url": "https://b.pop.example/", "cost": 2}]`. /// An http(s) URL is re-checked on a fixed cadence, with caching, conditional revalidation /// (`ETag` / `Last-Modified`), and stale-if-error handled by the shared HTTP @@ -1163,26 +1192,29 @@ impl Cluster { /// Returns an [`origin::Producer`] scoped to this session's subscribe permissions. /// /// Passed by reference to [`moq_net::Server::with_publisher`] (or the - /// equivalent per-request setter), which derives the read handle. + /// equivalent per-request setter), which derives the read handle. On an + /// upstream grant that handle never sees a route learned from another upstream. pub fn subscriber(&self, token: &auth::Token) -> Option { self.mounted(token)?.scope(&token.root, &token.subscribe).ok() } /// Returns an [`origin::Producer`] scoped to this session's publish permissions, - /// marked [`origin::Producer::peer`] when the grant names a cluster peer. + /// marked [`origin::Producer::peer`] when the grant names a cluster peer and + /// [`origin::Producer::upstream`] when that peer is upstream. /// Nothing is published beneath the grant's mounts. pub fn publisher(&self, token: &auth::Token) -> Option { - let publisher = self.mounted(token)?.scope(&token.root, &token.publish).ok()?; - Some(match token.peer { - true => publisher.peer(), - false => publisher, - }) + self.mounted(token)?.scope(&token.root, &token.publish).ok() } - /// The origin with the grant's mounts applied, before it is scoped to the - /// session. A mount the origin refuses (overlapping another) admits nothing. + /// The origin marked with the grant's link and with its mounts applied, + /// before it is scoped to the session. A mount the origin refuses + /// (overlapping another) admits nothing. fn mounted(&self, token: &auth::Token) -> Option { - let mut origin = self.origin.clone(); + let mut origin = match (token.upstream, token.peer) { + (true, _) => self.origin.clone().upstream(), + (false, true) => self.origin.clone().peer(), + (false, false) => self.origin.clone(), + }; for (at, target) in &token.mounts { origin = match origin.mount(token.root.join(at), target) { Ok(origin) => origin, @@ -1639,7 +1671,6 @@ impl Cluster { #[tracing::instrument("remote", skip_all, err, fields(remote = %target.key))] async fn run_remote(self, target: &DialTarget, token: String) -> anyhow::Result<()> { let mut urls = target.addrs(); - let cost = target.cost; // Apply the shared cluster token unless the URL already carries its own // non-empty `?jwt=` (a per-peer inline token or object `token` wins; the // shared token still covers peers that have none). An empty @@ -1666,9 +1697,7 @@ impl Cluster { loop { let started = tokio::time::Instant::now(); - let result = self - .run_remote_once(&addrs, cost, target.lan, target.fingerprint.as_deref()) - .await; + let result = self.run_remote_once(&addrs, target).await; let elapsed = started.elapsed(); match result { @@ -1687,37 +1716,25 @@ impl Cluster { } } - async fn run_remote_once( - &self, - addrs: &moq_tokio::Addrs, - cost: Option, - lan: bool, - fingerprint: Option<&str>, - ) -> anyhow::Result<()> { + /// One session to `target` over `addrs`, which carry any credential. + async fn run_remote_once(&self, addrs: &moq_tokio::Addrs, target: &DialTarget) -> anyhow::Result<()> { // Each attempt is its own session, so it gets its own id. Matches the span an // accepted connection runs under, so both directions log the same way. let id = self.next_connection_id(); - self.run_remote_session(id, addrs, cost, lan, fingerprint) + self.run_remote_session(id, addrs, target) .instrument(tracing::info_span!("conn", id)) .await } - async fn run_remote_session( - &self, - id: u64, - addrs: &moq_tokio::Addrs, - cost: Option, - lan: bool, - fingerprint: Option<&str>, - ) -> anyhow::Result<()> { + async fn run_remote_session(&self, id: u64, addrs: &moq_tokio::Addrs, target: &DialTarget) -> anyhow::Result<()> { // The peer URL may carry the cluster JWT in its query, so neither the log // line nor the node label below may show the raw URL. let first = addrs.as_slice().first().expect("Addrs is non-empty").url(); let redacted = moq_tokio::RedactedUrl::new(first); tracing::info!(url = %redacted, "dialing cluster peer"); - let client = if lan { - self.lan_client(fingerprint)? + let client = if target.lan { + self.lan_client(target.fingerprint.as_deref())? } else { self.client .clone() @@ -1726,14 +1743,18 @@ impl Cluster { // Cluster dials use their configured stats tier. Cluster peers carry no auth // root, so presence is keyed under the empty root within the cluster tier. - // The peer's routes entered the cluster elsewhere. A peer that predates the - // hidden opt-in still discovers our hidden routes; see `Cluster::scope`. - let origin = self.origin.clone().peer(); + // The peer's routes entered the cluster elsewhere, and an upstream peer is + // never offered another upstream's. A peer that predates the hidden opt-in + // still discovers our hidden routes; see `Cluster::scope`. + let origin = match target.upstream { + true => self.origin.clone().upstream(), + false => self.origin.clone().peer(), + }; let mut client = client .with_publisher(origin.consume().with_hidden(true)) .with_subscriber(origin) .with_stats(self.stats.tier(self.cluster_tier()).session("")); - if let Some(cost) = cost { + if let Some(cost) = target.cost { client = client.with_cost(cost); } // The GOAWAY lifecycle lives in the reconnect loop: on an upstream GOAWAY it @@ -2165,6 +2186,7 @@ mod tests { cost: None, fingerprint: None, lan: false, + upstream: false, } } @@ -2440,6 +2462,28 @@ mod tests { ); } + /// `upstream` is object-only policy: it round-trips, reaches the dial target, + /// and flipping it is a different configuration, so the API redials. + #[test] + fn peer_object_upstream() { + let object: Peer = + serde_json::from_str(r#"{"url": "https://core.example/", "upstream": true}"#).expect("parse upstream peer"); + assert_eq!(object, Peer::new("https://core.example/").with_upstream(true)); + assert_eq!( + serde_json::to_string(&object).unwrap(), + r#"{"url":"https://core.example/","upstream":true}"# + ); + let upstream = DialTarget::from_peer(&object).unwrap(); + assert!(upstream.upstream); + let plain = DialTarget::from_peer(&Peer::new("https://core.example/")).unwrap(); + assert_eq!(upstream.key, plain.key); + assert_ne!(upstream, plain); + assert_eq!( + serde_json::to_string(&Peer::new("https://core.example/")).unwrap(), + r#""https://core.example/""# + ); + } + /// `egress` defaults to `cost`, so a matching value is accepted and prices /// the link exactly like the bare URL form. An unpriced link costs 1, so /// `egress = 1` alone is symmetric too. @@ -2738,7 +2782,7 @@ mod tests { // that mutate it. let _env = crate::test_env::EnvGuard::lock(); - let toml = "[cluster]\nconnect = [\"https://a.example/?cost=1\", { url = \"https://b.example/\", cost = 2, egress = 2, token = \"secret\" }]\n"; + let toml = "[cluster]\nconnect = [\"https://a.example/?cost=1\", { url = \"https://b.example/\", cost = 2, egress = 2, token = \"secret\", upstream = true }]\n"; let dir = std::env::temp_dir().join("moq-relay-cluster-test"); std::fs::create_dir_all(&dir).unwrap(); let path = dir.join("cluster-connect-object-toml.toml"); @@ -2753,7 +2797,8 @@ mod tests { Peer::new("https://b.example/") .with_cost(2) .with_egress(2) - .with_token("secret"), + .with_token("secret") + .with_upstream(true), ] ); // The object normalizes exactly like its URL spellings. @@ -3127,6 +3172,7 @@ mod tests { cost: None, fingerprint: Some(fp), lan: true, + upstream: false, }; let _dial = fingerprint.dial_lan_target(&target).expect("dial"); @@ -3201,6 +3247,48 @@ mod tests { assert!(try_next_announced(&mut local).is_none()); } + /// An upstream grant's session is a peer that never reads another upstream's + /// routes: two accepted cores never transit through this relay. + #[tokio::test] + async fn upstream_grant_hides_other_upstreams() { + let cluster = new_cluster(Config::default()).expect("cluster"); + let all: moq_net::Patterns = [moq_net::Pattern::all()].into_iter().collect(); + + let client = auth::Token::new("/", &moq_auth::Grant::new(all.clone(), all.clone())); + let mut grant = moq_auth::Grant::new(all.clone(), all); + grant.peer = true; + let peer = auth::Token::new("/", &grant); + grant.upstream = true; + let core = auth::Token::new("/", &grant); + + let _ingest = cluster + .publisher(&client) + .expect("client") + .publish("ingest", Default::default()); + let _mesh = cluster + .publisher(&peer) + .expect("peer") + .publish("mesh", Default::default()); + let _core = cluster + .publisher(&core) + .expect("core") + .publish("core", Default::default()); + + let mut everyone = cluster.subscriber(&client).expect("client").consume().announced(); + let mut prefixes = std::iter::from_fn(|| try_next_announced(&mut everyone)) + .map(|announce| announce.prefix.to_string()) + .collect::>(); + prefixes.sort(); + assert_eq!(prefixes, ["core", "ingest", "mesh"]); + + let mut toward_core = cluster.subscriber(&core).expect("core").consume().announced(); + let mut prefixes = std::iter::from_fn(|| try_next_announced(&mut toward_core)) + .map(|announce| announce.prefix.to_string()) + .collect::>(); + prefixes.sort(); + assert_eq!(prefixes, ["ingest", "mesh"]); + } + /// A `/.cluster` request on a cluster without LAN discovery is refused. #[cfg(feature = "cluster-lan")] #[tokio::test] diff --git a/rs/moq-relay/tests/upstream_links.rs b/rs/moq-relay/tests/upstream_links.rs new file mode 100644 index 0000000000..32cf6f2fac --- /dev/null +++ b/rs/moq-relay/tests/upstream_links.rs @@ -0,0 +1,361 @@ +//! Upstream links over real TCP sessions: a relay never offers a route learned on +//! one upstream link to another, whichever side dialed, and everything else +//! still transits. +//! +//! Each relay is a [`cluster::Cluster`] serving qmux over TLS on TCP (`tls://`), +//! the way edge links run, through an embedded auth decider that reads the +//! dialed path: `/upstream` admits the dialer as an upstream peer, `/peer` as a +//! plain one. + +use std::collections::BTreeMap; +use std::time::Duration; + +use moq_auth::{Grant, Pattern, lease}; +use moq_net::{announce, origin}; +use moq_relay::{ + Connection, auth, + cluster::{self, Peer}, +}; + +const TIMEOUT: Duration = Duration::from_secs(15); + +/// Bound `fut` by [`TIMEOUT`], naming the step that hung. +async fn within(step: &str, fut: impl std::future::Future) -> T { + tokio::time::timeout(TIMEOUT, fut) + .await + .unwrap_or_else(|_| panic!("timed out: {step}")) +} + +/// Run on a big-stack thread: several relays' transport types make the test +/// future large enough to overflow libtest's default stack, as in +/// `goaway_cluster.rs`. +fn run_cluster_test(fut: F) +where + F: std::future::Future + Send + 'static, +{ + std::thread::Builder::new() + .stack_size(32 * 1024 * 1024) + .spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build test runtime") + .block_on(fut); + }) + .expect("spawn test thread") + .join() + .expect("test thread panicked"); +} + +/// One relay: its origin, listen port, and every task it runs. Dropping it +/// aborts them all, closing its sessions like a crash. +struct Node { + origin: origin::Producer, + port: u16, + _tasks: tokio::task::JoinSet<()>, +} + +impl Node { + /// Dial this node from another. `upstream` is the dialer's own mark; `as_upstream` + /// asks this node to admit the dialer as its upstream. + fn dial(&self, upstream: bool, as_upstream: bool) -> Peer { + let path = if as_upstream { "upstream" } else { "peer" }; + Peer::new(format!("tls://127.0.0.1:{}/{path}", self.port)).with_upstream(upstream) + } +} + +/// What the embedded decider grants a dial at `path`: everything, as a peer, +/// upstream when asked. +fn grant(path: &str) -> Grant { + let all: moq_net::Patterns = [Pattern::all()].into_iter().collect(); + let mut grant = Grant::new(all.clone(), all); + grant.root = Some(String::new()); + grant.peer = true; + grant.upstream = path == "/upstream"; + grant +} + +async fn node(id: u64, connect: Vec) -> Node { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let mut tasks = tokio::task::JoinSet::new(); + + let mut listen = moq_tokio::listen::Config::default(); + listen.tcp.bind = Some("127.0.0.1:0".parse().expect("parse addr")); + listen.tcp.tls = Some(true); + listen.tls.generate = vec!["localhost".into()]; + let mut listener = listen + .init(Default::default()) + .expect("server init") + .listen() + .await + .expect("listen"); + let port = listener.tcp_local_addr().expect("tcp listener bound").port(); + + let mut client = moq_tokio::connect::Config::default(); + client.tls.insecure = Some(true); + let client = client.init(Default::default()).expect("client init"); + + let mut config = cluster::Config::default(); + config.id = Some(id); + config.connect = connect; + let cluster = cluster::Cluster::new(cluster::Options::new(config)) + .expect("cluster init") + .with_client(client); + + let (auth, mut admissions) = auth::Auth::embedded(format!("node-{id}")); + tasks.spawn(async move { + while let Some(admission) = admissions.next().await { + let grant = grant(&admission.request.path); + admission.grant(lease::Consumer::fixed(grant)); + } + }); + + let serving = cluster.clone(); + tasks.spawn(async move { + let mut sessions = tokio::task::JoinSet::new(); + while let Some(request) = listener.accept().await { + let connection = Connection::new(request, serving.clone(), auth.clone()); + sessions.spawn(async move { + let _ = connection.run().await; + }); + } + }); + + let started = cluster.clone().start().await.expect("cluster start"); + tasks.spawn(async move { + let _ = started.run().await; + }); + + Node { + origin: cluster.origin.clone(), + port, + _tasks: tasks, + } +} + +/// A broadcast writing a group every 50ms on `origin`, so a subscriber attaching +/// at any point receives one. Dropping it ends the broadcast. +struct Publisher { + _broadcast: moq_net::broadcast::Producer, + writer: tokio::task::AbortHandle, +} + +impl Drop for Publisher { + fn drop(&mut self) { + self.writer.abort(); + } +} + +fn publish(origin: &origin::Producer, path: &str) -> Publisher { + let broadcast = origin.publish(path, Default::default()).expect("publish"); + let track = broadcast.create_track("video", None).expect("create track"); + let writer = tokio::spawn(async move { + loop { + let Ok(mut group) = track.append_group() else { break }; + if group.write_frame(moq_net::Timestamp::ZERO, b"hello".as_ref()).is_err() || group.finish().is_err() { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .abort_handle(); + Publisher { + _broadcast: broadcast, + writer, + } +} + +/// Subscribe to `path` through `origin`'s routes and read one frame. +async fn read_frame(origin: &origin::Producer, path: &str) { + let consumer = origin.consume(); + let broadcast = within(&format!("route to {path}"), consumer.routed_broadcast(path)) + .await + .unwrap_or_else(|err| panic!("{path} unroutable: {err}")); + let subscription = moq_net::track::Subscription::default().with_max_age(Duration::from_secs(1)); + let mut track = within( + &format!("subscribe to {path}"), + broadcast.track("video").expect("track handle").subscribe(subscription), + ) + .await + .expect("subscribe"); + let mut group = within(&format!("group of {path}"), track.recv_group()) + .await + .expect("recv group") + .expect("track closed"); + within(&format!("frame of {path}"), group.read_frame()) + .await + .expect("read frame") + .expect("group closed"); +} + +/// Each path `origin` routes now, with its route's hop ids, oldest first. +/// +/// Read from a fresh cursor's initial set, which reflects the route table as it +/// stands, unlike a live cursor that holds a best-route change back briefly. +async fn routes(origin: &origin::Producer) -> BTreeMap> { + let mut announced = origin.consume().announced(); + let mut routes = BTreeMap::new(); + loop { + match announced.next().await.expect("origin closed") { + announce::Event::Start(announce) | announce::Event::Update(announce) => { + let hops = announce.route.hops.iter().map(|hop| hop.id()).collect(); + routes.insert(announce.prefix.as_str().to_string(), hops); + } + announce::Event::End(announce) => { + routes.remove(announce.prefix.as_str()); + } + announce::Event::Live => return routes, + } + } +} + +/// Wait until `origin`'s routes satisfy `ready`, and return them. +async fn routes_when( + step: &str, + origin: &origin::Producer, + ready: impl Fn(&BTreeMap>) -> bool, +) -> BTreeMap> { + within(step, async { + // Created first, so a change landing between two reads still wakes us. + let mut changes = origin.consume().announced(); + loop { + let routes = routes(origin).await; + if ready(&routes) { + return routes; + } + changes.next().await.expect("origin closed"); + } + }) + .await +} + +fn has(routes: &BTreeMap>, paths: &[&str]) -> bool { + paths.iter().all(|path| routes.contains_key(*path)) +} + +const R: u64 = 1; +const C1: u64 = 11; +const C2: u64 = 12; +const E1: u64 = 21; +const E2: u64 = 22; + +/// Two regions: two edges whose links to two cores are upstream, and one relay +/// that also serves clients. Both edges send a path to the same core, an edge's +/// broadcast reaches the far relay, and losing a core moves only its paths. +#[test] +fn tiered_regions() { + run_cluster_test(tiered_regions_inner()); +} + +async fn tiered_regions_inner() { + let r = node(R, vec![]).await; + let c1 = node(C1, vec![r.dial(false, false)]).await; + let c2 = node(C2, vec![r.dial(false, false)]).await; + let e1 = node(E1, vec![c1.dial(true, false), c2.dial(true, false)]).await; + let e2 = node(E2, vec![c1.dial(true, false), c2.dial(true, false)]).await; + + let paths: Vec = (0..8).map(|i| format!("live/{i}")).collect(); + let paths: Vec<&str> = paths.iter().map(String::as_str).collect(); + let _published: Vec<_> = paths.iter().map(|path| publish(&r.origin, path)).collect(); + routes_when("both cores learn the far region", &c1.origin, |routes| { + has(routes, &paths) + }) + .await; + routes_when("both cores learn the far region", &c2.origin, |routes| { + has(routes, &paths) + }) + .await; + + // A route is delivered before a later one: once an edge holds both cores' + // markers, it holds both cores' routes to the far region too. + let _c1_mark = publish(&c1.origin, "zz/c1"); + let _c2_mark = publish(&c2.origin, "zz/c2"); + let marks = ["zz/c1", "zz/c2"]; + let at_e1 = routes_when("edge 1 hears both cores", &e1.origin, |routes| has(routes, &marks)).await; + let at_e2 = routes_when("edge 2 hears both cores", &e2.origin, |routes| has(routes, &marks)).await; + + // Every edge sends a given path to the same core, so the far region serves + // one subscription per broadcast whichever edges watch it. + let core = |routes: &BTreeMap>, path: &str| *routes[path].last().expect("hops"); + for path in &paths { + assert_eq!(at_e1[*path].len(), 2, "{path} at edge 1: {:?}", at_e1[*path]); + assert_eq!(core(&at_e1, path), core(&at_e2, path), "{path} split across cores"); + } + read_frame(&e1.origin, paths[0]).await; + read_frame(&e2.origin, paths[0]).await; + + // A client on the far relay reaches a broadcast published on an edge. + let _edge = publish(&e1.origin, "edge/cam"); + read_frame(&r.origin, "edge/cam").await; + + // Losing a core moves only its paths. + let on = |core_id: u64| -> Vec<&str> { paths.iter().copied().filter(|p| core(&at_e1, p) == core_id).collect() }; + let (on_c1, on_c2) = (on(C1), on(C2)); + assert!( + !on_c1.is_empty() && !on_c2.is_empty(), + "paths all on one core: {at_e1:?}" + ); + drop(c1); + let after = routes_when("edge 1 drops the lost core", &e1.origin, |routes| { + has(routes, &paths) && routes.values().all(|hops| !hops.contains(&C1)) + }) + .await; + for path in &on_c2 { + assert_eq!(after[*path], at_e1[*path], "{path} moved off the surviving core"); + } + for path in &on_c1 { + assert_eq!(after[*path], [R, C2], "{path} did not fail over"); + } +} + +/// A route from one core never reaches the other through an edge, whether the +/// edge dialed the core or the core dialed the edge. +#[test] +fn edges_never_transit_between_cores() { + run_cluster_test(edges_never_transit_between_cores_inner()); +} + +async fn edges_never_transit_between_cores_inner() { + let c1 = node(C1, vec![]).await; + let e1 = node(E1, vec![c1.dial(true, false)]).await; + // The core dials this edge, which still treats it as upstream. + let c2 = node(C2, vec![e1.dial(false, true)]).await; + let e2 = node(E2, vec![c1.dial(true, false), c2.dial(true, false)]).await; + + let _core = publish(&c1.origin, "live/c1"); + routes_when("edge 1 learns core 1", &e1.origin, |routes| has(routes, &["live/c1"])).await; + routes_when("edge 2 learns core 1", &e2.origin, |routes| has(routes, &["live/c1"])).await; + + // Each edge's own broadcast reaches core 2 after anything it would offer from + // core 1, so once both arrive, core 2 has seen every offer it will get. + let _e1 = publish(&e1.origin, "zz/e1"); + let _e2 = publish(&e2.origin, "zz/e2"); + let at_c2 = routes_when("core 2 hears both edges", &c2.origin, |routes| { + has(routes, &["zz/e1", "zz/e2"]) + }) + .await; + assert!( + !at_c2.contains_key("live/c1"), + "core 1's broadcast reached core 2 through an edge: {at_c2:?}" + ); + read_frame(&c2.origin, "zz/e1").await; + read_frame(&c1.origin, "zz/e2").await; +} + +/// A drone mesh node, a drone with a CDN uplink marked upstream, and a CDN relay +/// in a line: the mesh reaches the CDN and the CDN reaches the mesh. +#[test] +fn uplink_transits_both_ways() { + run_cluster_test(uplink_transits_both_ways_inner()); +} + +async fn uplink_transits_both_ways_inner() { + let cdn = node(31, vec![]).await; + let uplink = node(32, vec![cdn.dial(true, false)]).await; + let mesh = node(33, vec![uplink.dial(false, false)]).await; + + let _mesh = publish(&mesh.origin, "mesh/cam"); + let _cdn = publish(&cdn.origin, "cdn/cam"); + read_frame(&cdn.origin, "mesh/cam").await; + read_frame(&mesh.origin, "cdn/cam").await; +} diff --git a/rs/moq-tokio/Cargo.toml b/rs/moq-tokio/Cargo.toml index bbe9c0defa..88b6f65498 100644 --- a/rs/moq-tokio/Cargo.toml +++ b/rs/moq-tokio/Cargo.toml @@ -39,7 +39,7 @@ iroh = ["_transport", "dep:web-transport-iroh", "dep:web-transport-proto", "dep: mdns = ["dep:mdns-sd", "dep:hmac", "dep:sha2", "dep:subtle"] jemalloc = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-ctl"] websocket = ["_transport", "dep:qmux", "watch", "dep:dns-lookup"] -# Plain-TCP qmux transport (`tcp://`), no TLS. Server-side and client-side. +# Qmux over TCP: plaintext `tcp://`, and `tls://` once a crypto provider is on. Server-side and client-side. tcp = ["_transport", "dep:qmux", "dep:dns-lookup"] # Unix-domain-socket qmux transport (`unix://`), unix-only. Adds peer-credential # inspection on accept. Pulls in `tcp` (shares qmux's stream transport). diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index d7c517d258..de3b156421 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -65,12 +65,16 @@ pub struct Client { /// Whether the TLS config pins a certificate fingerprint, which only verifies /// the configured host, so a GOAWAY may not redirect elsewhere. pub(crate) pinned: bool, - /// The resolved Happy Eyeballs timings, used by the `tcp://` dial here; the - /// QUIC backend captures its own copy from the config. + /// The resolved Happy Eyeballs timings, used by the `tcp://` and `tls://` + /// dials here; the QUIC backend captures its own copy from the config. #[cfg(feature = "tcp")] failover_delay: std::time::Duration, #[cfg(feature = "tcp")] resolution_delay: std::time::Duration, + /// The TLS settings a `tls://` dial builds from when it runs, so a client + /// that only dials plaintext `tcp://` never needs a crypto provider. + #[cfg(feature = "tcp")] + tcp_tls: crate::tls::Connect, #[cfg(feature = "websocket")] websocket: crate::websocket::Config, /// The TLS server name override used by the WebSocket fallback. @@ -130,6 +134,8 @@ impl Client { let failover_delay = resolved.race; #[cfg(feature = "tcp")] let resolution_delay = resolved.resolution_delay; + #[cfg(feature = "tcp")] + let tcp_tls = config.tls.clone(); #[cfg(feature = "websocket")] let tls_host_name = config.tls.host_name.clone(); let timeout = resolved.timeout; @@ -154,6 +160,8 @@ impl Client { failover_delay, #[cfg(feature = "tcp")] resolution_delay, + #[cfg(feature = "tcp")] + tcp_tls, #[cfg(feature = "websocket")] websocket: config.websocket, #[cfg(feature = "websocket")] @@ -388,6 +396,20 @@ impl Client { return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); } + // Qmux over TLS over TCP, for links that need neither QUIC nor a WebSocket. + #[cfg(feature = "tcp")] + if url.scheme() == "tls" { + let session = crate::tcp::connect_tls( + url, + &self.versions.alpns(), + &self.tcp_tls, + self.failover_delay, + self.resolution_delay, + ) + .await?; + return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?); + } + // Unix domain socket (qmux, no TLS). Same-host only; the server can // authenticate us by uid/gid via SO_PEERCRED. #[cfg(all(feature = "uds", unix))] @@ -551,7 +573,7 @@ fn setup_path(url: &Url) -> Option { .filter(|path| !path.is_empty()), // Raw QUIC and qmux over TCP negotiate an ALPN and nothing else, so the whole // request target travels in the SETUP. - "moqt" | "moql" | "tcp" => request_target(url), + "moqt" | "moql" | "tcp" | "tls" => request_target(url), _ => None, } } @@ -789,6 +811,8 @@ mod tests { ("tcp://localhost:4443/room", Some("/room")), ("tcp://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")), ("tcp://localhost:4443", None), + ("tls://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")), + ("tls://localhost:4443", None), // Raw QUIC: the URL is ours alone, so the path and query have to ride the // SETUP or the server never sees them. ("moqt://relay.example.com/anon", Some("/anon")), diff --git a/rs/moq-tokio/src/connect.rs b/rs/moq-tokio/src/connect.rs index 3dbbe6a889..6589087eb8 100644 --- a/rs/moq-tokio/src/connect.rs +++ b/rs/moq-tokio/src/connect.rs @@ -507,7 +507,7 @@ pub struct Config { /// When DNS returns multiple addresses, attempts alternate between IPv6 and /// IPv4, each starting this long after the previous one (or immediately when /// it fails), and the first connection to complete wins. `0s` dials every - /// address at once. Defaults to 250ms. Applies to the QUIC and `tcp://` dials. + /// address at once. Defaults to 250ms. Applies to the QUIC, `tcp://`, and `tls://` dials. /// /// This staggers the attempts within one [`crate::Client::connect`]; [`Self::timeout`] /// bounds that call as a whole. diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 1e9178c819..c3c6a727cb 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -272,14 +272,14 @@ impl Redirect { /// Whether this scheme's dial installs the Rustls verifier, so a configured /// fingerprint actually checked the peer. Plain and non-Rustls transports do not. fn fingerprint_pins(scheme: &str) -> bool { - matches!(scheme, "https" | "wss" | "moqt" | "moql") + matches!(scheme, "https" | "wss" | "moqt" | "moql" | "tls") } /// Rank a scheme so a peer-supplied redirect cannot silently drop encryption. /// Unknown schemes rank lowest, so a forgotten classification is refused. fn scheme_tier(scheme: &str) -> u8 { match scheme { - "https" | "moqt" | "moql" | "wss" | "iroh" => 2, + "https" | "moqt" | "moql" | "wss" | "iroh" | "tls" => 2, "tcp" | "ws" | "http" => 1, // `unix` lands here deliberately: local IPC is not an upgrade over a // network transport, it is a different reachability class (see `is_local`). @@ -1553,7 +1553,7 @@ mod tests { /// forgot to add is refused rather than trusted. #[test] fn scheme_tiers_rank_encrypted_above_plaintext() { - for scheme in ["https", "moqt", "moql", "wss", "iroh"] { + for scheme in ["https", "moqt", "moql", "wss", "iroh", "tls"] { assert_eq!(scheme_tier(scheme), 2, "{scheme} is encrypted"); } for scheme in ["tcp", "ws", "http"] { @@ -1703,7 +1703,7 @@ mod tests { /// must not inherit the pin. #[test] fn a_fingerprint_pin_only_covers_rustls_schemes() { - for scheme in ["https", "wss", "moqt", "moql"] { + for scheme in ["https", "wss", "moqt", "moql", "tls"] { assert!(fingerprint_pins(scheme), "{scheme}"); } for scheme in ["http", "ws", "tcp", "unix", "iroh"] { diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs index f520b86565..e5bea556af 100644 --- a/rs/moq-tokio/src/lib.rs +++ b/rs/moq-tokio/src/lib.rs @@ -4,7 +4,7 @@ //! - WebTransport (HTTP/3) //! - Raw QUIC (with ALPN negotiation) //! - WebSocket (fallback via [web-transport-ws](https://crates.io/crates/web-transport-ws)) -//! - Plain TCP via the `tcp://` scheme (qmux, no TLS; requires `tcp` feature) +//! - TCP via the `tcp://` (plaintext) and `tls://` schemes (qmux; requires `tcp` feature) //! - Unix domain socket via the `unix://` scheme (qmux, peer-credential aware; requires `uds` feature, unix-only) //! - Iroh P2P (requires `iroh` feature) //! diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index cbd88f64bd..03610d519b 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -423,6 +423,10 @@ impl Config { #[cfg(feature = "_transport")] pub(crate) fn validate(&self) -> crate::Result<()> { + #[cfg(feature = "tcp")] + if self.tcp.tls == Some(true) && self.tcp.bind.is_none() { + return Err(crate::Error::NoBackend("--listen-tcp-tls requires --listen-tcp-bind")); + } Ok(()) } diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 1a360703b8..0e4ab43fa2 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -329,7 +329,11 @@ impl Server { let mut stream_binds = Vec::new(); #[cfg(feature = "tcp")] if let Some(addr) = config.tcp.bind.filter(|_| build_streams) { - stream_binds.push(StreamBind::Tcp(addr)); + let tls = match config.tcp.tls == Some(true) { + true => Some(TcpTls::new(&config, &versions)?), + false => None, + }; + stream_binds.push(StreamBind::Tcp(addr, tls)); } #[cfg(all(feature = "uds", unix))] if let Some(path) = config.unix.bind.clone().filter(|_| build_streams) { @@ -798,7 +802,7 @@ impl Listener { self.server.websocket_local_addr() } - /// The address the plain TCP (qmux) listener bound to, if one was configured. + /// The address the TCP (qmux) listener bound to, if one was configured. #[cfg(feature = "tcp")] pub fn tcp_local_addr(&self) -> Option { self.server.streams.tcp_local_addr @@ -988,12 +992,67 @@ fn stream_versions(base: &moq_net::Versions) -> moq_net::Versions { /// A configured stream listener (`--listen-tcp-bind` / `--listen-unix-bind`). #[cfg(any(feature = "tcp", all(feature = "uds", unix)))] enum StreamBind { + /// With its TLS when the listener serves `tls://`. #[cfg(feature = "tcp")] - Tcp(net::SocketAddr), + Tcp(net::SocketAddr, Option), #[cfg(all(feature = "uds", unix))] Unix(PathBuf), } +/// The TLS a `--listen-tcp-tls` listener serves, reloading its certificate +/// while this lives. +#[cfg(feature = "tcp")] +struct TcpTls { + config: std::sync::Arc, + #[cfg(all(feature = "watch", feature = "_certs"))] + _reload: crate::tls::ReloadingServerConfig, +} + +#[cfg(all(feature = "tcp", feature = "_certs"))] +impl TcpTls { + /// Serve the listen certificate, accepting each stream version as a + /// `qmux-01.` TLS ALPN. + /// + /// Asks for no client certificate: qmux's TLS accept keeps no peer identity + /// for the auth server to read, so neither a `tls.root` nor pinned `peers` + /// meant for QUIC apply, and a peer on this listener authenticates with a token. + fn new(config: &crate::listen::Config, versions: &moq_net::Versions) -> crate::Result { + let mut listen = config.tls.clone(); + listen.root.clear(); + listen.peers = None; + let alpn = stream_versions(versions) + .alpns() + .iter() + .map(|alpn| format!("{}{alpn}", qmux::Version::QMux01.prefix()).into_bytes()) + .collect(); + Self::serve(&listen, alpn) + } + + #[cfg(feature = "watch")] + fn serve(listen: &crate::tls::Listen, alpn: Vec>) -> crate::Result { + let reload = listen.server_config_reloading(alpn)?; + Ok(Self { + config: reload.config(), + _reload: reload, + }) + } + + #[cfg(not(feature = "watch"))] + fn serve(listen: &crate::tls::Listen, alpn: Vec>) -> crate::Result { + Ok(Self { + config: listen.server_config(alpn)?, + }) + } +} + +#[cfg(all(feature = "tcp", not(feature = "_certs")))] +impl TcpTls { + /// Without a crypto provider there is no certificate to serve. + fn new(_config: &crate::listen::Config, _versions: &moq_net::Versions) -> crate::Result { + Err(Error::NoBackend("--listen-tcp-tls requires a crypto provider feature")) + } +} + /// A bound stream listener, before its accept loop is spawned. #[cfg(any(feature = "tcp", all(feature = "uds", unix)))] enum BoundListener { @@ -1009,7 +1068,7 @@ impl StreamBind { fn name(&self) -> &'static str { match self { #[cfg(feature = "tcp")] - Self::Tcp(_) => "tcp", + Self::Tcp(..) => "tcp", #[cfg(all(feature = "uds", unix))] Self::Unix(_) => "unix", } @@ -1025,6 +1084,9 @@ impl StreamBind { /// stopped when the [`Listener`] closes or drops, so no socket lingers. #[cfg(any(feature = "tcp", all(feature = "uds", unix)))] struct StreamListeners { + /// Keeps the TCP listener's TLS certificate reloading while it serves. + #[cfg(feature = "tcp")] + _tcp_tls: Option, binds: Vec, /// One per entry in `binds`, in the same order, and created up front rather than /// with the listener: an owner registering these with a metrics endpoint does so @@ -1057,6 +1119,8 @@ impl StreamListeners { .map(|bind| crate::accept::Health::new(bind.name())) .collect(); Self { + #[cfg(feature = "tcp")] + _tcp_tls: None, binds, health, versions, @@ -1086,16 +1150,21 @@ impl StreamListeners { let alpns = self.versions.alpns(); match bind { #[cfg(feature = "tcp")] - StreamBind::Tcp(addr) => { - if !addr.ip().is_loopback() { + StreamBind::Tcp(addr, tls) => { + if tls.is_none() && !addr.ip().is_loopback() { tracing::warn!(%addr, "tcp listener bound to a non-loopback address; qmux is UNENCRYPTED, ensure the network is trusted"); } - let listener = crate::tcp::Listener::bind(addr) + let mut listener = crate::tcp::Listener::bind(addr) .await? .with_protocols(alpns) .with_accept_health(health); + let encrypted = tls.is_some(); + if let Some(tls) = tls { + listener = listener.with_tls(tls.config.clone()); + self._tcp_tls = Some(tls); + } let local = listener.local_addr()?; - tracing::info!(addr = %local, "listening (tcp)"); + tracing::info!(addr = %local, tls = encrypted, "listening (tcp)"); self.tcp_local_addr = Some(local); pending.push(BoundListener::Tcp(listener)); } @@ -1338,7 +1407,7 @@ pub enum Transport { Iroh, /// A WebSocket connection using qmux framing. WebSocket, - /// A plaintext TCP connection using qmux framing. + /// A TCP connection using qmux framing, plaintext or TLS. Tcp, /// A Unix domain socket using qmux framing. Unix, diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs index bec2ac4d22..25a81911fc 100644 --- a/rs/moq-tokio/src/settings.rs +++ b/rs/moq-tokio/src/settings.rs @@ -72,6 +72,9 @@ struct ListenTls { struct ListenTcp { #[usage(env = "MOQ_LISTEN_TCP_BIND", cli("--listen-tcp-bind"))] bind: Option, + + #[usage(env = "MOQ_LISTEN_TCP_TLS", cli("--listen-tcp-tls"))] + tls: Option, } /// `[listen.unix]` / `--listen-unix-*`. diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs index b275388633..853dc883b8 100644 --- a/rs/moq-tokio/src/tcp.rs +++ b/rs/moq-tokio/src/tcp.rs @@ -1,11 +1,13 @@ -//! Plain-TCP qmux transport, reachable via the `tcp://` URL scheme. +//! Qmux over TCP: plaintext via the `tcp://` URL scheme, TLS via `tls://`. //! -//! Runs the QMux wire format directly over TCP with no TLS or WebSocket -//! framing. There is no transport encryption and no authentication, so only -//! use this on a trusted network (loopback, a private VPC interface, etc.). +//! Both run the QMux wire format directly over TCP with no WebSocket framing. +//! Plaintext `tcp://` has no transport encryption and no authentication, so only +//! use it on a trusted network (loopback, a private VPC interface, etc.). +//! `tls://` encrypts and verifies the server with the same TLS settings as the +//! other transports, for links that need neither QUIC nor a WebSocket. //! -//! TCP has no TLS handshake, so the application protocol (the moq ALPN) is -//! negotiated in-band: pass the offered/supported protocols and the resulting +//! Plaintext TCP has no TLS handshake, so the application protocol (the moq +//! ALPN) is negotiated in-band; TLS negotiates it in its handshake. Either way //! `qmux::Session::protocol()` is populated before connect/accept returns. use std::net; @@ -15,11 +17,12 @@ use url::Url; /// negotiated) since there's no TLS ALPN to carry it. const WIRE_VERSION: qmux::Version = qmux::Version::QMux01; -/// Plaintext-TCP qmux listener settings (no TLS, no UDP). +/// Qmux TCP listener settings, plaintext or TLS (no UDP). /// -/// Flattened onto [`crate::listen::Config::tcp`]. TCP carries no peer identity, so -/// the listener must only be reachable from trusted clients. Bind it to loopback -/// or a private interface; a non-loopback bind logs a warning but is allowed. +/// Flattened onto [`crate::listen::Config::tcp`]. Plaintext carries no peer +/// identity and no encryption, so bind it to loopback or a private interface; a +/// non-loopback plaintext bind logs a warning but is allowed. With +/// [`tls`](Self::tls) the listener serves the listen TLS certificate instead. // The derived arg group is named after the struct, so it needs an explicit id to // stay unique across the flattened sections. #[derive(usage::Args, Clone, Debug, Default, serde::Serialize, serde::Deserialize)] @@ -27,7 +30,7 @@ const WIRE_VERSION: qmux::Version = qmux::Version::QMux01; #[serde(deny_unknown_fields, default)] #[non_exhaustive] pub struct Config { - /// Bind a plaintext qmux TCP listener on this address. + /// Bind a qmux TCP listener on this address, plaintext unless [`Self::tls`]. #[usage( long = "listen-tcp-bind", name = "listen-tcp-bind", @@ -37,6 +40,21 @@ pub struct Config { #[serde(default, skip_serializing_if = "Option::is_none")] pub bind: Option, + /// Serve TLS on the TCP listener with the listen certificate, for `tls://` + /// dials. It asks for no client certificate, so a peer on it authenticates + /// with a token rather than mTLS. + #[usage( + long = "listen-tcp-tls", + name = "listen-tcp-tls", + env = "MOQ_LISTEN_TCP_TLS", + setting = "listen.tcp.tls", + default_missing = "true", + num_args = 0..=1, + require_equals = true, + )] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tls: Option, + /// The released `--server-tcp-bind` spelling and its env var, folded in by /// [`Config::resolved`]. #[usage(flatten)] @@ -76,7 +94,7 @@ impl Config { } } -/// Errors specific to the plain-TCP qmux transport. +/// Errors specific to the qmux TCP transport. #[derive(Debug, thiserror::Error)] #[non_exhaustive] pub enum Error { @@ -86,11 +104,11 @@ pub enum Error { #[error(transparent)] Io(#[from] std::io::Error), - /// The `tcp://` URL had no host. + /// The `tcp://` or `tls://` URL had no host. #[error("missing hostname")] MissingHostname, - /// The `tcp://` URL had no port. Unlike `https`, there is no default. + /// The `tcp://` or `tls://` URL had no port. Unlike `https`, there is no default. #[error("missing port")] MissingPort, @@ -152,6 +170,44 @@ pub(crate) async fn connect( connect_addrs(candidates, protocols, failover_delay).await } +/// Dial a `tls://host:port` URL: qmux over TLS over TCP, negotiating one of +/// `protocols` as the TLS ALPN and verifying the server with `tls`. +/// +/// Resolves and races addresses exactly like [`connect`], through the TLS and +/// qmux handshakes. The port is required. +pub(crate) async fn connect_tls( + url: Url, + protocols: &[&str], + tls: &crate::tls::Connect, + failover_delay: std::time::Duration, + resolution_delay: std::time::Duration, +) -> crate::Result { + let host = url.host().ok_or(Error::MissingHostname)?; + let port = url.port().ok_or(Error::MissingPort)?; + let name = tls.host_name.clone().unwrap_or_else(|| match &host { + url::Host::Domain(name) => name.to_string(), + url::Host::Ipv4(ip) => ip.to_string(), + url::Host::Ipv6(ip) => ip.to_string(), + }); + let client = qmux::tls::Client::new(std::sync::Arc::new(tls.build()?)) + .with_protocols(protocols.iter().map(|&alpn| (alpn, &[WIRE_VERSION][..]))) + .require_protocol(); + + tracing::debug!(peer = %crate::connect::Endpoint(&url), "connecting via TLS over TCP"); + let candidates = crate::resolve::Candidates::resolve(host, port, resolution_delay); + Ok(crate::failover::race(candidates, failover_delay, |addr| { + let client = client.clone(); + let name = name.clone(); + async move { + client + .connect(addr, &name) + .await + .map_err(|err| Error::Connect(crate::error::message(err))) + } + }) + .await?) +} + /// Dial `candidates` in Happy Eyeballs order, performing the qmux handshake on /// each attempt; the first session to complete wins. async fn connect_addrs( @@ -172,11 +228,13 @@ async fn connect_addrs( .await } -/// Listens for incoming plain-TCP qmux connections on a TCP port. +/// Listens for incoming qmux connections on a TCP port, plaintext or TLS. pub struct Listener { listener: tokio::net::TcpListener, protocols: Vec, health: crate::accept::Health, + /// Present when the listener serves TLS; it negotiates the ALPN itself. + tls: Option, } impl Listener { @@ -187,9 +245,17 @@ impl Listener { listener, protocols: Vec::new(), health: crate::accept::Health::new("tcp"), + tls: None, }) } + /// Serve TLS with `config`, whose ALPN list already names each + /// `qmux-01.` pair to accept. + pub(crate) fn with_tls(mut self, config: std::sync::Arc) -> Self { + self.tls = Some(qmux::tls::Server::new(config)); + self + } + /// A live handle to this listener's accept-loop health, for an embedder that /// publishes it (see [`crate::accept`]). pub fn accept_health(&self) -> crate::accept::Health { @@ -222,7 +288,8 @@ impl Listener { Ok(self.listener.local_addr()?) } - /// Accept the next connection, performing the qmux handshake over plain TCP. + /// Accept the next connection, performing the TLS (when configured) and qmux + /// handshakes. /// /// A failed `accept(2)` is handled here rather than yielded: it is classified, /// counted, logged, and paced by [`accept_health`](Self::accept_health), then @@ -243,11 +310,13 @@ impl Listener { let (stream, addr) = self.accept_socket().await; tracing::debug!(%addr, "accepted TCP connection"); let config = qmux::tcp::Config::new(WIRE_VERSION).protocols(self.protocols.iter().map(String::as_str)); + let tls = self.tls.clone(); async move { - let session = config - .accept(stream) - .await - .map_err(|err| Error::Accept(crate::error::message(err)))?; + let session = match tls { + Some(tls) => tls.accept(stream).await, + None => config.accept(stream).await, + } + .map_err(|err| Error::Accept(crate::error::message(err)))?; Ok((session, addr)) } } @@ -276,6 +345,79 @@ mod tests { use std::time::Duration; use web_transport_trait::Session as _; + /// TLS needs a TCP listener to serve on; asking for it alone is refused. + #[test] + fn tls_without_a_bind_is_refused() { + let mut listen = crate::listen::Config::default(); + listen.tcp.tls = Some(true); + assert!(matches!( + listen.init(Default::default()), + Err(crate::Error::NoBackend(_)) + )); + } + + /// A `tls://` dial carries qmux and the request target over TLS, and a client + /// that does not trust the certificate is refused. Client-certificate policy + /// meant for QUIC (CA roots or pinned peers) never makes the TCP listener ask + /// for one, so a token-only client still gets through. + #[cfg(feature = "aws-lc-rs")] + #[tokio::test] + async fn tls_carries_qmux_and_refuses_an_untrusted_certificate() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + for pinned in [false, true] { + let mut listen = crate::listen::Config::default(); + listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap()); + listen.tcp.tls = Some(true); + listen.tls.generate = vec!["localhost".into()]; + match pinned { + true => listen.tls.peers = Some(crate::tls::Peers::new()), + false => listen.tls.root = vec!["/nonexistent/client-ca.pem".into()], + } + tls_round_trip(listen).await; + } + } + + #[cfg(feature = "aws-lc-rs")] + async fn tls_round_trip(listen: crate::listen::Config) { + let mut server = listen.init(Default::default()).unwrap().listen().await.unwrap(); + let port = server.tcp_local_addr().unwrap().port(); + let url: Url = format!("tls://localhost:{port}/room?jwt=credential").parse().unwrap(); + let accept = tokio::spawn(async move { + let request = server.accept().await.unwrap(); + assert_eq!(request.transport(), crate::server::Transport::Tcp); + assert_eq!(request.path(), "/room"); + assert_eq!(request.query(), Some("jwt=credential")); + let session = request.ok().await.unwrap(); + (server, session) + }); + + let dial = |insecure: bool| { + let mut config = crate::connect::Config { + once: Some(true), + ..Default::default() + }; + config.tls.insecure = Some(insecure); + config + .init(Default::default()) + .unwrap() + .with_reconnect(false) + .connect(url.clone()) + }; + let untrusted = tokio::time::timeout(Duration::from_secs(5), dial(false).established()).await; + assert!( + matches!(untrusted, Ok(Err(_))), + "an untrusted certificate must be refused" + ); + let _session = tokio::time::timeout(Duration::from_secs(5), dial(true).established()) + .await + .expect("trusted dial timed out") + .expect("trusted dial failed"); + let (_server, _accepted) = tokio::time::timeout(Duration::from_secs(5), accept) + .await + .expect("accept timed out") + .expect("accept task panicked"); + } + /// End-to-end failover: the preferred candidate blackholes (TEST-NET-1 never /// answers, or is unroutable outright in a sandbox), so the race must fall /// through to the loopback listener within the stagger delay.