diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 1a4d3d2aaf..a1ef1633de 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -190,6 +190,10 @@ jobs: if: ${{ !cancelled() }} run: nix develop --command cargo bench --locked -p moq-relay --bench http_idle -- --test + - name: Relay auth refresh benchmark smoke + if: ${{ !cancelled() }} + run: nix develop --command just rs bench-relay-auth --test + # One relay's stats tick (drain plus plain and compressed encode), swept # over held paths, tiers, and how many of those paths changed, plus the # decode table. `--test` runs each tick routine once (decode ignores it diff --git a/doc/bin/relay/auth.md b/doc/bin/relay/auth.md index 04e0d2724d..6780bc2085 100644 --- a/doc/bin/relay/auth.md +++ b/doc/bin/relay/auth.md @@ -44,11 +44,12 @@ same `root` and `mounts` resizes the live session in place, narrower or wider, so a moderation decision lands on the session it targets and can be lifted the same way: what falls outside a narrower grant resets with `Unauthorized`, the broadcasts the session published outside it abort, and everything else keeps -flowing. A wider grant brings those paths back, up to what the session was -admitted with. A one-shot HTTP `/fetch` ends on a narrower grant instead. A -changed `root`, `mounts`, `peer`, or `upstream` closes the session. A changed -`tier` keeps the session and moves its stats from then on. A 401 or -403 closes it. Any other failure retries with backoff and the session lives +flowing. A wider grant brings those paths back within the admitted root and +SETUP role. A one-shot HTTP `/fetch` ends on a narrower grant instead. A +changed `root`, `mounts`, `peer`, or `upstream`, or a 401 or 403, ends that +credential's grant, which closes the session unless an +[in-band](#in-band-renewal) credential still holds it. A changed `tier` keeps +the session and moves its stats from then on. Any other failure retries with backoff and the session lives until `expires`, so an outage always has the bound the server chose. A grant with `revalidate` must set `expires` for that reason. @@ -113,6 +114,39 @@ TOKEN` option with Token Type 0. HMAC, RSA, ECDSA, and EdDSA keys all work. A key can itself be **scoped** at generation (`--root`, `--publish`, `--subscribe`), after which it can never sign a broader token. +### In-band renewal + +On moq-lite-07-wip or moq-transport draft-17 and later with MoQ Auth negotiated, +a client can present another JWT without reconnecting. Present its replacement +before the current grant expires and keep the returned token handle alive. +The native `Connection::auth()` handle re-presents retained tokens after a +reconnect; the JavaScript session's `auth.add()` presents one for that session. +Older versions refuse in-band tokens. + +Each accepted token adds its publish and subscribe patterns to the session's +union, within its admitted root and SETUP role. A publish-only connection cannot +become a subscriber. A replacement must retain the root, mounts, and routing +classification. Dropping a token handle, expiry, or refusal at revalidation +removes that token's grant: streams that lose permission reset, and other grants +keep the connection alive. A refused new token leaves existing grants alone. +The grant reports its real expiry so the client can renew in time. + +The relay verifies a presented JWT through the same admission path, replacing +`jwt` in the original query and preserving the connection's identity. The auth +server sees another `connect` for the same session id. Attached grants revalidate +independently but never report `end`; only the connection reports its end, even +if its first credential expired earlier, after its attached grants stop. Each +attached JWT counts against its own `--limit-token` cap until the connection +ends. The initial credential remains responsible for the connection's stats tier. +Certificate sessions cannot renew a certificate in band, and public rules +cannot verify a presented token; both refuse it as unsupported. A burst of +tokens past the session's pending-verification bound is refused. On moq-lite, an +auth server that cannot answer, by timeout or outage, refuses a token as a +transient error, and `Connection::auth()` presents it again on its reconnect +backoff. On moq-transport the refusal is final and the token ends. Retrying +never extends an accepted grant, so a session whose last grant expires during an +outage still closes. + ### Claims | Claim | Meaning | @@ -120,7 +154,7 @@ key can itself be **scoped** at generation (`--root`, `--publish`, | `root` | Base path. Optional. | | `publish` | Patterns the bearer may publish under `root`. `**` means everything; omitted means no publishing. | | `subscribe` | Patterns the bearer may subscribe to under `root`. Same rules. | -| `exp`, `iat`, `nbf` | Expiry, issue time, and not-before. `exp` is enforced for the whole session, not just at connect, and a token is refused from its `exp` on and before its `nbf`. | +| `exp`, `iat`, `nbf` | Expiry, issue time, and not-before. `exp` ends that credential's grant, including after connect; another accepted token can keep the session alive. A token is refused from its `exp` on and before its `nbf`. | | `iss`, `sub`, `jti` | Read and ignored. | Any other claim refuses the token with its name, `aud` included: an unknown @@ -296,7 +330,9 @@ admissions itself, without the HTTP: leave `[auth]` empty and take refusal. A fixed lease never changes; a `lease::Producer` the application keeps re-checks, updates, and revokes on its own clock. The relay still closes the session at the grant's `expires`, and an admission left unanswered for ten -seconds is refused. Gateway accept loops can call `Cluster::admit` for the same +seconds is refused. `Admission::attached()` marks a token presented +[in band](#in-band-renewal) for a session already admitted under the same id, +so count sessions only on the others. Gateway accept loops can call `Cluster::admit` for the same scoped session as a native connection. See [docs.rs/moq-relay](https://docs.rs/moq-relay) and [moq-auth](/lib/rs/moq-auth). diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index a0758d866a..11695bdbd1 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -168,7 +168,11 @@ publisher learns before anyone subscribes whether its broadcasts can reach the peer. More tokens can be added without reconnecting; the session's scope is the union of every open token's grant. A token the peer has no verifier for, or whose grant is too large to send, fails with `Unsupported` and ends only that -token; the session and its other grants stay up. +token; the session and its other grants stay up. A relay with an +[auth server](/bin/relay/auth#in-band-renewal) verifies these tokens through it, +so a fresh grant can keep the connection alive after its original credential +expires; under public rules or a client certificate it refuses them as +`Unsupported`. A subscription or fetch that loses access resets with the `UNAUTHORIZED` stream code and the session stays up. A client that publishes outside its grant closes diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 03b1eadd3a..049100b0dd 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -29,7 +29,7 @@ subset drops. What a peer actually observes against this implementation: - **History is one group.** A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed. - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed, as is every track on drafts 14 through 16. A Rust reader that never subscribed takes the units from `TRACK_STATUS` on draft 18 and later. - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. Message parameters follow the negotiated draft's lists: drafts 14 and 15 ignore an unknown or misplaced parameter, draft 16 ignores one defined only for another message but closes the session on an unknown one, and draft 17 and later close it on both. -- **One credential per session**, carried in `SETUP` and forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. A token attached to an individual request is otherwise decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` is refused unread, so its token is not decoded. +- **One `SETUP` credential**, forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second `SETUP` token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. A token attached to an individual request is otherwise decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` is refused unread, so its token is not decoded. On draft 17 and later, the negotiated [MoQ Auth extension](/concept/moq-lite#authorization) accepts additional session tokens in band. - **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, range filters, `TRACK_STATUS` to a JavaScript publisher, `SUBSCRIBE_TRACKS`, and the fetch forms above. On draft 19 and later, and in Rust on drafts 14 through 16, a subscription update may change only priority; any other update ends that subscription. - **Datagrams** are a single normal object at object 0, forwarded without renumbering. Anything else is dropped, and a malformed one closes the session. Rust and JavaScript both carry them on every draft. - **Priority.** Higher is served first. The IETF default of 128 is this stack's 127, and a track that never sets one is 127. diff --git a/js/net/src/auth.test.ts b/js/net/src/auth.test.ts index 4eae496d32..3ce563b47f 100644 --- a/js/net/src/auth.test.ts +++ b/js/net/src/auth.test.ts @@ -61,6 +61,46 @@ async function connect(opts: { publish?: OriginProducer; serverPublish?: OriginP // Every case runs on moq-lite-07-wip and on moq-transport with the MoQ Auth extension. describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", (protocol) => { + test("pending application tokens have a per-session bound", async () => { + const { client, server } = await connect({ publish: new OriginProducer(), protocol }); + try { + const requests = server.auth.requests(); + const setup = await requests.next(); + if (!setup) throw new Error("setup token missing"); + setup.accept(grant(["a"], [])); + let refused!: () => void; + const overflow = new Promise((resolve) => { + refused = resolve; + }); + const pending = Array.from({ length: 129 }, (_, index) => + client.auth.add(`token-${index}`).then( + (token) => ({ token }), + (err: unknown) => { + refused(); + return { err }; + }, + ), + ); + await withTimeout(overflow, 1000, "pending token bound did not refuse overflow"); + for (let index = 0; index < 128; index++) { + const request = await requests.next(); + if (!request) throw new Error("pending token missing"); + request.accept(grant(["a"], [])); + } + const results = await Promise.all(pending); + const errors = results.flatMap((result) => ("err" in result ? [result.err] : [])); + expect(errors).toHaveLength(1); + // moq-transport's AUTH_ERROR registry only distinguishes unsupported from unauthorized. + expect((errors[0] as SessionError).code).toBe( + protocol === Lite.ALPN_07_WIP ? SessionCode.TooManyRequests : SessionCode.Unauthorized, + ); + for (const result of results) if ("token" in result) result.token.close(); + } finally { + client.abort(); + server.abort(); + } + }); + test("both sides learn their default grant", async () => { const { client, server } = await connect({ publish: new OriginProducer(), protocol }); @@ -449,6 +489,33 @@ test("a revoked grant resets its subscriptions with UNAUTHORIZED", async () => { } }); +test("withdrawn queued tokens are pruned before the next application decision", async () => { + const { client, server, transport } = await connect({ publish: new OriginProducer(), protocol: Lite.ALPN_07_WIP }); + try { + const requests = server.auth.requests(); + const setup = await requests.next(); + if (!setup) throw new Error("setup token missing"); + setup.accept(grant(["a"], [])); + const wire = new LiteAuthWire(transport, Lite.Version.DRAFT_07); + for (let index = 0; index < 128; index++) { + const stream = await wire.present(new TextEncoder().encode("withdrawn")); + await stream.writer.close(); + await stream.reader.closed; + } + const live = client.auth.add("live"); + live.catch(() => void 0); + const request = await requests.next(); + if (!request) throw new Error("live token missing"); + expect(new TextDecoder().decode(request.token)).toBe("live"); + request.accept(grant(["a"], [])); + (await live).close(); + await request.closed; + } finally { + client.abort(); + server.abort(); + } +}); + // Capture the acceptor's actual AUTH writer after a valid setup grant. The following // bytes travel through the same connection dispatcher and presenter as a real peer. describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("AUTH violations %s", (protocol) => { diff --git a/js/net/src/auth.ts b/js/net/src/auth.ts index 7bce301da9..75d1124821 100644 --- a/js/net/src/auth.ts +++ b/js/net/src/auth.ts @@ -60,6 +60,8 @@ export interface Token { export interface Request { /** The token the peer presented. Empty means the credential its connection carried. */ readonly token: Uint8Array; + /** Settles when the peer withdraws this unanswered token or closes the session. */ + readonly closed: PromiseLike; /** * Grant the token until the returned {@link Issued} is revoked or closed. * diff --git a/js/net/src/auth_session.ts b/js/net/src/auth_session.ts index c790fbb90e..696bd01f58 100644 --- a/js/net/src/auth_session.ts +++ b/js/net/src/auth_session.ts @@ -128,6 +128,14 @@ class PeerRequest implements Request { this.#issued = issued; } + get closed(): Promise { + return this.#issued.closed; + } + + get ended(): boolean { + return this.#issued.ended; + } + accept(grant: Grant): Issued { if (this.#answered) throw new Error("already answered"); this.#answered = true; @@ -144,6 +152,7 @@ class PeerRequest implements Request { /** Our side of one of the peer's tokens: the grant we issued and its stream. */ class IssuedGrant implements Issued { + ended = false; readonly closed: Promise; #stream: Stream; #wire: AuthWire; @@ -162,7 +171,10 @@ class IssuedGrant implements Issued { .then( () => null, (err: unknown) => (err instanceof StreamError && err.code === StreamCode.Cancel ? null : error(err)), - ); + ) + .finally(() => { + this.ended = true; + }); } #write(write: () => Promise) { @@ -204,16 +216,25 @@ class RequestQueue implements Requests { #waiters: ((request: PeerRequest | undefined) => void)[] = []; #closed = false; - push(request: PeerRequest): boolean { - if (this.#closed) return false; + /** Queue the token for the application, or refuse it when closed or full. */ + push(request: PeerRequest): void { + if (this.#closed) { + request.reject(SessionCode.Unauthorized, "not accepting tokens"); + return; + } + if (this.#queue.length >= 128) this.#queue = this.#queue.filter((queued) => !queued.ended); + if (this.#queue.length >= 128) { + request.reject(SessionCode.TooManyRequests, "too many pending tokens"); + return; + } const waiter = this.#waiters.shift(); if (waiter) waiter(request); else this.#queue.push(request); - return true; } next(): Promise { - const next = this.#queue.shift(); + let next = this.#queue.shift(); + while (next?.ended) next = this.#queue.shift(); if (next || this.#closed) return Promise.resolve(next); return new Promise((resolve) => this.#waiters.push(resolve)); } @@ -323,7 +344,7 @@ export class AuthSession implements AuthApi { const issued = new IssuedGrant(stream, wire); if (this.#acceptor instanceof RequestQueue) { const request = new PeerRequest(token, issued); - if (!this.#acceptor.push(request)) request.reject(SessionCode.Unauthorized, "not accepting tokens"); + this.#acceptor.push(request); } else if (token.byteLength > 0) { // Only the connection's own credential has a default answer; a token needs // someone to verify it. diff --git a/quest/m1/auth/README.md b/quest/m1/auth/README.md index ad46dfac38..5b0e0b5836 100644 --- a/quest/m1/auth/README.md +++ b/quest/m1/auth/README.md @@ -111,8 +111,6 @@ because `main` assigned 0x3A to NOT_FETCHABLE first. ## Required - [WebSocket refusal](/quest/m1/auth/ws-unauthorized.md) - the relay serves WebSocket through moq-tokio, so a refused token closes the session as Unauthorized, as QUIC does -- [Relay tokens](/quest/m1/auth/relay-refresh.md) - the relay verifies tokens - sent in band, unions their grants, and cancels only work that loses access - [Request tokens](/quest/m1/auth/request-token.md) - an experimental Rust verifier hook confines a token's authority to its request and renews it - [Request gate](/quest/m1/auth/request-gate.md) - every incoming request is diff --git a/quest/m1/auth/bindings.md b/quest/m1/auth/bindings.md index 072d514512..8395c24771 100644 --- a/quest/m1/auth/bindings.md +++ b/quest/m1/auth/bindings.md @@ -22,8 +22,7 @@ export the experimental Rust request hook through the bindings here. Arc` whose object carries that token's own `grant()` and `closed()` and whose release withdraws it, raising the structured error moq-ffi already maps on refusal. `MoqClient` sessions reached through the - reconnecting connection use the accessor - [Relay tokens](/quest/m1/auth/relay-refresh.md) adds. + reconnecting connection use `moq_tokio::Connection::auth()`. - No new libmoq API: the hand-written C ABI gets no more feature work. The generated C and C++ bindings get this from moq-ffi, and OBS reaches it through the generated C++; update `cpp/obs/src` @@ -48,5 +47,3 @@ surface comes from moq-ffi rather than new `moq_session_auth_*` calls. ## Required - [FFI shape](/quest/m1/ffi-shape/README.md) - reshapes the moq-ffi session and wrappers this edits, so the wrappers break once -- [Relay tokens](/quest/m1/auth/relay-refresh.md) - supplies the connection - accessor and a relay that answers a real token, which the wrapper tests need diff --git a/quest/m1/auth/expired-error.md b/quest/m1/auth/expired-error.md index ae4ecae1c3..85519877fc 100644 --- a/quest/m1/auth/expired-error.md +++ b/quest/m1/auth/expired-error.md @@ -17,10 +17,12 @@ refusal as final. shared moq-transport session value if one means the same thing, otherwise 0x31. - Carry it through moq-ffi's error mapping and each wrapper. +- Switch the relay's in-band renewal (`rs/moq-relay/src/auth/refresh.rs`) to + refuse or revoke an expired credential with `Expired`. Today it uses + `UNAUTHORIZED`, with the cause only in the reason. -Moved from m2 into the auth line: relay tokens refuse an expired -token with `AUTH_ERROR { Expired }`, so without this moq-net cannot tell it -apart from `Unauthorized`. +Moved from m2 into the auth line: without this, a client renewing in band +cannot tell an expired token apart from a refused one. ## Required diff --git a/quest/m1/auth/relay-refresh.md b/quest/m1/auth/relay-refresh.md deleted file mode 100644 index c43a34163c..0000000000 --- a/quest/m1/auth/relay-refresh.md +++ /dev/null @@ -1,100 +0,0 @@ -# [M] Relay tokens - -## Goal - -A client presents tokens in band and moq-relay verifies each through the same -`Auth` path the URL token took: the session's scope is the union of every -accepted token, a refused token gets AUTH_ERROR and changes nothing, and a -session whose credential is about to lapse renews by presenting a fresh one -before the old expires. The grant a client receives carries the token's real -expiry, an expiry that leaves the union intact ends only that token, and -`doc/bin/relay/auth.md` documents the exchange. - -## Plan - -- Coordinate shared plumbing with [Request tokens](/quest/m1/auth/request-token.md) - and #5148. The 2026-10-12 request-token plan chooses one queue with explicit - session/request event variants and distinct acceptance handles. The relay - accepts session events here and explicitly refuses request verification - as `Unsupported`; it must never attach a request token as a session grant. - Share cancellation, pending-work bounds, and persistent connection handles - whichever implementation lands first. #5282 already implements this shared - plumbing; its separate open API and limit-policy decisions remain open. -- `Connection::run` in `rs/moq-relay/src/connection.rs` takes `requests()` - from `moq_net::server::Handshake::auth()` before `.ok()`, so the relay owns - the initial empty AUTH too and the driver's fallback never races it. - `moq_tokio::server::Request::ok` spawns the driver itself, so it needs the - same `auth()` passthrough first. An empty - token is answered from the origin handles as the default does, plus - the admitted grant's `expires`. A non-empty token is presented through - `Client::attach(&connection_lease, request)`, a method this quest adds: - the request carries the connection's `id`, path, and transport, the query - `jwt=` exactly as the URL would have, and no certificate facts, - since a certificate cannot be presented in band. An `attach` lease - revalidates like any other but never POSTs `end`, and the server treats a - `connect` for an id it already holds as one more grant on that session, - so lifecycle stays with the connection lease and a withdrawn token frees - no session-limit slot. `doc/bin/relay/auth.md` states that rule. mTLS - sessions reject a non-empty token as `Unsupported`. -- Union: the connection holds the set of accepted leases. Every one - must carry the admitted root, else `AUTH_ERROR { Unauthorized }` naming the - root. The peer is authorized for the union of the set's publish and subscribe - patterns, intersected with the role the client declared at SETUP so a - publish-only session never starts receiving announcements because a later - token happened to carry subscribe prefixes, re-authorized through - `auth::Handle::authorize` whenever the union changes, grown or shrunk; keep it - behind one function. A grant never reaches past the session's origin handles, - so scope those to the admitted root and role rather than to the first - token's patterns, and nothing is rebuilt or swapped in. -- Expiry: the deadline today is the admitted lease's `closed()` inside one - `tokio::select!` arm. Select on one lease per token in the set instead, and on any firing recompute the union without it: unchanged - means `AUTH_ERROR { Expired }` on that token's stream and the session - continues, shrunk re-authorizes the smaller union, which resets each stream that - lost access with `UNAUTHORIZED`. A token - withdrawn by the client (its stream closed or reset) recomputes the union - the same way with nothing written back, since the stream is gone: unchanged - means the session continues, shrunk re-authorizes the same way. Each lease revalidates on - its own cadence, and every re-check that changes the grant, a lower - `expires` or a different `publish` or `subscribe`, writes the complete - replacement grant as an update AUTH_OK on that token's stream and - recomputes the union from the latest grant of every token, so the client - can present a replacement in time and a narrowed token never keeps its old - scope; a changed union re-authorizes as above. -- Refusals map as connect-time ones do and never expose a status: every - refusal `Client::attach` returns for a new token, a `403`, an empty or - invalid grant, an unparseable body, or a `5xx`, is - `AUTH_ERROR { Unauthorized }`, and a timeout is `AUTH_ERROR { Timeout }`. - On re-check a refusal closes that token's stream with - `AUTH_ERROR { Unauthorized }` and recomputes the union, while a failed - re-check (timeout or `5xx`) keeps the token until its `expires`, never a - close on its own. -- The client side: `moq_tokio::Connection` gains `auth()` - returning a handle the connection owns, not the current session's. It - stores every token added through it, presents them on each new session as - it attaches, unions the live session's grant, and its `add` resolves against - the session that is up at the time; a token the app drops is withdrawn from - the live session and forgotten. Its request-token half belongs to - [Request tokens](/quest/m1/auth/request-token.md); whichever lands first - adds the handle. -- Bound pending tokens: `auth::Handle::requests()` hands out an unbounded - queue today, and a closed AUTH stream leaves its request queued, so a peer - that finishes AUTH streams faster than the relay verifies them grows it - without limit (Codex security review on #4039). Bound it per session and - refuse what overflows; share the bound and withdrawal support with the - request-token hook whichever lands first. Neither consumer may land with - unbounded pending verification. Dropping a `Request` already refuses its - token. -- Docs: `doc/bin/relay/auth.md` gains an "in-band tokens" section beside - revalidation stating that grants union, that a token needs the admitted - root, what an expiry does to the union today, and that the grant's expiry - is the token's `exp`. -- Tests: a second token with a later `exp` and the same scope keeps the - session past the first token's expiry, with only the first stream ending; - an expired or invalid token is refused and nothing changes; a token with a - different root is refused naming it; a token adding a prefix widens the - grant and the prefix is now solicited; withdrawing the only token - covering a prefix resets that prefix's streams with `UNAUTHORIZED` and - the session stays connected; an mTLS session refuses a token; a - proxy-mode outage on re-check keeps the token. - -Additive. diff --git a/quest/m1/auth/request-token.md b/quest/m1/auth/request-token.md index 99ca31f570..7ad6fa89b4 100644 --- a/quest/m1/auth/request-token.md +++ b/quest/m1/auth/request-token.md @@ -43,12 +43,10 @@ Continue [#5148](https://github.com/moq-dev/moq/pull/5148), which carries capacity before awaiting the application and observes request/session cancellation. Withdrawal must cancel pending work and make late answers ineffective; any verifier work still in flight remains counted until it - ends. Renewal has the same ownership discipline. Coordinate queue - bounds and cancellation with [Relay tokens](/quest/m1/auth/relay-refresh.md) - and #5282, which already adds shared pending-AUTH support; do not defer the - initial-admission fix to the later request-gate refactor. Reuse shared - primitives whichever PR lands first, without approving #5282's separate - API or limit-policy decisions here. + ends. Renewal has the same ownership discipline. Reuse the bounded + pending-AUTH queue and `Request::closed` withdrawal that session tokens + already use; do not defer the initial-admission fix to the later + request-gate refactor. - **Documentation.** Minimal Rust API comments only: experimental hook, application verification required, no per-request relay integration, and setter success is not peer authorization. No feature page, guide, or guide @@ -83,8 +81,9 @@ Retained decisions: for each request. Replacement renews live eligible requests, and success means local configuration accepted, not authentication acknowledged. `Connection::auth()` owns it across reconnects and - `connect::Config::with_request_token` seeds it. Share the persistent handle - and server passthrough with #5282; add no parallel `Client` methods. + `connect::Config::with_request_token` seeds it. Extend the existing + `Connection::auth()` handle and `server::Request::auth()` passthrough; add no + parallel `Client` methods. - **Update credit.** Advertise and enforce MAX_REQUEST_UPDATES on draft 19+, with TOO_MANY_REQUEST_UPDATES (0x1B) in the shared Rust/JS session registry and moq-lite's registry. Earlier drafts keep a local guard that ends only @@ -118,7 +117,5 @@ promised by these mechanisms. - [#5106](https://github.com/moq-dev/moq/pull/5106) - landed setup extension configuration; no remaining prerequisite for starting this hook -- [Relay tokens](/quest/m1/auth/relay-refresh.md) - shared pending-AUTH and - persistent connection plumbing, with session verification kept separate - [Request leases](/quest/m3/auth-request-lease.md) - deferred relay verification, not a prerequisite for this experimental hook diff --git a/quest/m1/auth/token-in-band.md b/quest/m1/auth/token-in-band.md index 10c3a26d23..c6ffe22c3c 100644 --- a/quest/m1/auth/token-in-band.md +++ b/quest/m1/auth/token-in-band.md @@ -82,7 +82,5 @@ Additive. ## Required - [FFI shape](/quest/m1/ffi-shape/README.md) - puts the `MoqClientConfig` record in place that `tokens` extends, so the wrappers break once -- [Relay tokens](/quest/m1/auth/relay-refresh.md) - supplies the verify and - widen path the configured tokens reuse - [Bindings](/quest/m1/auth/bindings.md) - supplies the client surface the new token setters sit beside diff --git a/quest/m3/auth-request-lease.md b/quest/m3/auth-request-lease.md index e5b75dda5e..da72b0a5f4 100644 --- a/quest/m3/auth-request-lease.md +++ b/quest/m3/auth-request-lease.md @@ -23,7 +23,7 @@ it is the relay's half of [Request tokens](/quest/m1/auth/request-token.md). - A per-request call on `moq_auth::Client`, not the `Client::attach` that - [Relay tokens](/quest/m1/auth/relay-refresh.md) builds. `attach` connects + session tokens use. `attach` connects with the connection's id, and the auth server treats that as one more grant on the session that never POSTs `end`, so a request token would widen the whole connection for its life. The per-request call carries the token in @@ -44,5 +44,3 @@ Public API: additive on `moq_auth::Client`. Wire: none. - [Request tokens](/quest/m1/auth/request-token.md) - the per-request grant this lease backs -- [Relay tokens](/quest/m1/auth/relay-refresh.md) - supplies the lease - revalidation this reuses diff --git a/quest/m3/p2p/peer-grant.md b/quest/m3/p2p/peer-grant.md index d6551c9ab0..95b659bc82 100644 --- a/quest/m3/p2p/peer-grant.md +++ b/quest/m3/p2p/peer-grant.md @@ -21,8 +21,7 @@ roster, so a copied JWT is not enough: the presenter proves possession of the matching private key on the direct AUTH stream. A roster peer that received someone else's grant cannot replay it. -Issuance: the relay owns AUTH via -[Relay tokens](/quest/m1/auth/relay-refresh.md). The client writes +Issuance: the relay owns AUTH (`rs/moq-relay/src/auth/refresh.rs`). The client writes `AUTH_CNF { Jwk (b) }` after AUTH on the empty-token stream; the relay copies that JWK into `cnf`. Then, after AUTH_OK: diff --git a/rs/justfile b/rs/justfile index b34502ea0c..64ea888b60 100644 --- a/rs/justfile +++ b/rs/justfile @@ -349,6 +349,11 @@ bench-playback: bench-rtmp *args: cargo bench --locked -p moq-rtmp --bench chunk_interleaving -- "$@" +# Refresh one credential across unrelated sessions and held token counts. +[positional-arguments] +bench-relay-auth *args: + cargo bench --locked -p moq-relay --bench auth -- "$@" + # One relay's stats tick, plus the decode table. `--test` runs each tick once; decode ignores it and prints its whole table. [positional-arguments] bench-stats *args: diff --git a/rs/moq-auth/src/client.rs b/rs/moq-auth/src/client.rs index 061be2e0cd..e15d0d08f4 100644 --- a/rs/moq-auth/src/client.rs +++ b/rs/moq-auth/src/client.rs @@ -1,5 +1,6 @@ use std::future::Future; use std::pin::Pin; +use std::sync::Arc; use std::time::{Duration, Instant}; use url::Url; @@ -18,6 +19,7 @@ const TIMEOUT: Duration = Duration::from_secs(10); /// `end` when the session closes. #[derive(Clone)] pub struct Client { + identity: Arc<()>, http: reqwest::Client, url: Url, } @@ -59,15 +61,154 @@ impl Client { }; Ok(Self { + identity: Arc::new(()), http: builder.build()?, url, }) } /// Admit a session: POST `connect`, validate the reply, and return the lease the - /// session holds. The session reports totals through [`lease::Consumer::close`]. + /// session holds. The session reports its end and totals through [`lease::Consumer::close`] + /// or drop; credential expiry alone never reports that the connection ended. pub async fn connect(&self, request: Request) -> crate::Result { - connect(self.clone(), request).await + let context = Context::new(request.clone(), self.identity.clone()); + admit(self.clone(), request, context).await + } + + /// Verify another credential for this connection without owning its end event. + pub async fn attach(&self, connection: &lease::Consumer, request: Request) -> crate::Result { + let context = connection.context().ok_or(Error::Refused)?; + let context = context.attach(&request, &self.identity)?; + admit(self.clone(), request, context).await + } +} + +/// A credential's ownership of the connection it belongs to. +#[derive(Clone)] +pub(crate) struct Context { + connection: Arc, + owner: bool, +} + +struct Connection { + request: Request, + issuer: Arc<()>, + lifecycle: kio::Shared, +} + +#[derive(Default)] +struct Lifecycle { + ended: Option<(Reason, Bytes, Instant)>, + pending: usize, +} + +/// End must follow every request an attached credential has in flight for this +/// connection id: its Connect, and each Revalidate until its driver stops. +struct Admission(Arc); + +impl Drop for Admission { + fn drop(&mut self) { + self.0.lifecycle.lock().pending -= 1; + } +} + +impl std::fmt::Debug for Context { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Context") + .field("owner", &self.owner) + .finish_non_exhaustive() + } +} + +impl Context { + fn new(mut request: Request, issuer: Arc<()>) -> Self { + request.event = Event::Connect; + Self { + connection: Arc::new(Connection { + request, + issuer, + lifecycle: Default::default(), + }), + owner: true, + } + } + + fn attach(&self, request: &Request, issuer: &Arc<()>) -> crate::Result { + if !self.owner || !Arc::ptr_eq(issuer, &self.connection.issuer) || self.connection.request.tls.is_some() { + return Err(Error::Refused); + } + let mut expected = self.connection.request.clone(); + expected.query = request.query.clone(); + expected.token = request.token.clone(); + if &expected != request { + return Err(Error::Refused); + } + if self.connection.lifecycle.read().ended.is_some() { + return Err(Error::Refused); + } + Ok(Self { + connection: self.connection.clone(), + owner: false, + }) + } + + fn admitting(&self) -> crate::Result> { + if self.owner { + return Ok(None); + } + let mut lifecycle = self.connection.lifecycle.lock(); + if lifecycle.ended.is_some() { + return Err(Error::Refused); + } + lifecycle.pending += 1; + Ok(Some(Admission(self.connection.clone()))) + } + + pub(crate) fn close(&self, reason: Reason, bytes: Bytes) { + if self.owner { + self.connection + .lifecycle + .lock() + .ended + .get_or_insert((reason, bytes, Instant::now())); + } + } + + /// Resolve with the connection's end reason, for an attached credential only. + async fn connection_ended(&self) -> Reason { + if self.owner { + return std::future::pending().await; + } + kio::wait(|waiter| { + let lifecycle = std::task::ready!(self.connection.lifecycle.poll(waiter, |lifecycle| { + if lifecycle.ended.is_some() { + std::task::Poll::Ready(()) + } else { + std::task::Poll::Pending + } + })); + std::task::Poll::Ready(lifecycle.ended.as_ref().expect("waited for connection end").0.clone()) + }) + .await + } + + async fn end(&self) -> Option<(Reason, Bytes, Instant)> { + if !self.owner { + return None; + } + Some( + kio::wait(|waiter| { + let lifecycle = std::task::ready!(self.connection.lifecycle.poll(waiter, |ended| { + if ended.ended.is_some() && ended.pending == 0 { + std::task::Poll::Ready(()) + } else { + std::task::Poll::Pending + } + })); + std::task::Poll::Ready(lifecycle.ended.clone().expect("waited for connection end")) + }) + .await, + ) } } @@ -110,13 +251,20 @@ impl Post for Client { } /// Admit a session through `server` and spawn the driver behind its lease. -async fn connect(server: S, mut request: Request) -> crate::Result { +async fn admit(server: S, mut request: Request, context: Context) -> crate::Result { request.event = Event::Connect; + let admission = context.admitting()?; let grant = ask(&server, &request).await?; - let (producer, consumer) = lease::Producer::new(grant.clone()); + if context.connection.lifecycle.read().ended.is_some() { + return Err(Error::Refused); + } + let (producer, mut consumer) = lease::Producer::new(grant.clone()); + consumer.set_context(context.clone()); let driver = Driver { + context, + _admission: admission, server, request, producer: Some(producer), @@ -151,6 +299,9 @@ fn redact(url: &Url) -> String { /// The task behind a lease: re-checks on cadence and reports the end. struct Driver { + context: Context, + /// Held by an attached credential's driver, so the connection's End waits for it. + _admission: Option, server: S, request: Request, producer: Option, @@ -159,12 +310,15 @@ struct Driver { impl Driver { async fn run(mut self) { - let (reason, bytes) = self.drive().await; + self.drive().await; + let Some((reason, bytes, ended)) = self.context.end().await else { + return; + }; let mut request = self.request.clone(); request.event = Event::End { reason, - duration: self.started.elapsed(), + duration: ended.saturating_duration_since(self.started), bytes, }; // The session is already gone; nothing to do with a failure but say so. @@ -183,7 +337,7 @@ impl Driver { let mut inflight: Option> + Send>>> = None; let mut pending = false; - loop { + let ended = loop { let reply = async { match inflight.as_mut() { Some(request) => request.await, @@ -192,9 +346,11 @@ impl Driver { }; tokio::select! { biased; - ended = producer.closed() => return ended, + ended = producer.closed() => break ended, + // An attached credential ends with its connection. + reason = self.context.connection_ended() => break producer.finish(reason, Bytes::default()), due = producer.due() => match due { - Due::Expired => return producer.finish(Reason::Expired, Bytes::default()), + Due::Expired => break producer.finish(Reason::Expired, Bytes::default()), Due::Revalidate if inflight.is_some() => pending = true, Due::Revalidate => inflight = Some(self.post_revalidate()), }, @@ -202,9 +358,9 @@ impl Driver { inflight = None; match result { Ok(fresh) => producer.update(fresh), - Err(Error::Refused) => return producer.finish(Reason::Refused, Bytes::default()), + Err(Error::Refused) => break producer.finish(Reason::Refused, Bytes::default()), Err(Error::GrantExpired | Error::UnboundedRevalidate | Error::ZeroRevalidate) => { - return producer.finish(Reason::Invalid, Bytes::default()); + break producer.finish(Reason::Invalid, Bytes::default()); }, Err(err) => { let delay = producer.failed(); @@ -218,7 +374,13 @@ impl Driver { } }, } + }; + // Land an in-flight Revalidate before releasing the connection's End, so + // End is the last request for this id however the lease ended. + if let Some(request) = inflight { + let _ = request.await; } + ended } fn post_revalidate(&self) -> Pin> + Send>> { @@ -302,6 +464,7 @@ mod tests { /// socket, the clock only advances once every answer has landed. #[derive(Clone)] struct Script { + identity: Arc<()>, log: Log, #[allow(clippy::type_complexity)] respond: std::sync::Arc Option> + Send + Sync>, @@ -310,13 +473,21 @@ mod tests { impl Script { fn new(log: Log, respond: impl Fn(&Request) -> Option> + Send + Sync + 'static) -> Self { Self { + identity: Arc::new(()), log, respond: std::sync::Arc::new(respond), } } async fn connect(&self) -> crate::Result { - connect(self.clone(), request()).await + admit(self.clone(), request(), Context::new(request(), self.identity.clone())).await + } + async fn attach(&self, owner: &lease::Consumer, request: Request) -> crate::Result { + let context = owner + .context() + .ok_or(Error::Refused)? + .attach(&request, &self.identity)?; + admit(self.clone(), request, context).await } } @@ -374,6 +545,205 @@ mod tests { } } + #[tokio::test] + async fn an_attached_lease_never_reports_the_connection_end() { + let log = Log::default(); + let script = Script::new(log.clone(), |_| Some(Ok(grant(None, None)))); + let owner = script.connect().await.unwrap(); + let retained = owner.clone(); + let mut request = request(); + request.query = Some("jwt=replacement".into()); + let attached = script.attach(&owner, request).await.unwrap(); + drop(attached); + tokio::task::yield_now().await; + assert_eq!(log.events(), [Event::Connect, Event::Connect]); + owner.close("done", Bytes { sent: 7, received: 11 }); + match log.end().await.event { + Event::End { reason, bytes, .. } => { + assert_eq!(reason, Reason::Session("done".into())); + assert_eq!(bytes, Bytes { sent: 7, received: 11 }); + } + other => panic!("expected connection end, got {other:?}"), + } + drop(retained); + } + + #[tokio::test] + async fn an_attached_lease_ends_with_its_connection() { + let log = Log::default(); + let script = Script::new(log.clone(), |_| Some(Ok(grant(None, None)))); + let owner = script.connect().await.unwrap(); + let mut request = request(); + request.query = Some("jwt=replacement".into()); + let attached = script.attach(&owner, request).await.unwrap(); + owner.close(Reason::Shutdown, Bytes::default()); + assert_eq!(attached.closed().await, Reason::Shutdown); + log.end().await; + attached.revalidate(); + tokio::task::yield_now().await; + assert!( + matches!( + log.events().as_slice(), + [Event::Connect, Event::Connect, Event::End { .. }] + ), + "an attached lease outlived its connection's End" + ); + } + + #[tokio::test] + async fn a_withdrawn_attachment_lands_its_revalidate_before_end() { + #[derive(Clone)] + struct HeldRevalidate { + script: Script, + arrived: Arc, + release: Arc, + } + impl Post for HeldRevalidate { + async fn post(&self, request: &Request) -> crate::Result { + if request.event == Event::Revalidate { + self.arrived.notify_one(); + self.release.notified().await; + } + self.script.post(request).await + } + async fn post_end(&self, request: &Request) -> crate::Result<()> { + self.script.post_end(request).await + } + } + let log = Log::default(); + let script = Script::new(log.clone(), |_| Some(Ok(grant(None, None)))); + let owner = script.connect().await.unwrap(); + let server = HeldRevalidate { + script: script.clone(), + arrived: Arc::default(), + release: Arc::default(), + }; + let mut request = request(); + request.query = Some("jwt=replacement".into()); + let context = owner.context().unwrap().attach(&request, &script.identity).unwrap(); + let attached = admit(server.clone(), request, context).await.unwrap(); + attached.revalidate(); + server.arrived.notified().await; + drop(attached); + tokio::task::yield_now().await; + owner.close(Reason::Shutdown, Bytes::default()); + tokio::task::yield_now().await; + assert_eq!( + log.events(), + [Event::Connect, Event::Connect], + "end overtook a withdrawn attachment's in-flight revalidate" + ); + server.release.notify_one(); + log.end().await; + assert!(matches!( + log.events().as_slice(), + [Event::Connect, Event::Connect, Event::Revalidate, Event::End { .. }] + )); + } + + #[tokio::test] + async fn initial_credential_expiry_does_not_end_a_renewed_connection() { + tokio::time::pause(); + let log = Log::default(); + let script = Script::new(log.clone(), |request| { + let seconds = if request.query.as_deref() == Some("jwt=replacement") { + 5 + } else { + 1 + }; + Some(Ok(grant(Some(Duration::from_secs(seconds)), None))) + }); + let owner = script.connect().await.unwrap(); + let mut request = request(); + request.query = Some("jwt=replacement".into()); + let attached = script.attach(&owner, request).await.unwrap(); + assert_eq!(owner.closed().await, Reason::Expired); + assert_eq!(log.events(), [Event::Connect, Event::Connect]); + assert!(matches!(attached.poll_closed(&kio::Waiter::noop()), Poll::Pending)); + owner.close(Reason::Shutdown, Bytes::default()); + assert!(matches!( + log.end().await.event, + Event::End { + reason: Reason::Shutdown, + .. + } + )); + drop(attached); + } + + #[tokio::test] + async fn connection_end_waits_for_an_attachment_already_being_verified() { + #[derive(Clone)] + struct HeldPost { + script: Script, + arrived: Arc, + release: Arc, + } + impl Post for HeldPost { + async fn post(&self, request: &Request) -> crate::Result { + self.arrived.notify_one(); + self.release.notified().await; + self.script.post(request).await + } + async fn post_end(&self, request: &Request) -> crate::Result<()> { + self.script.post_end(request).await + } + } + let log = Log::default(); + let script = Script::new(log.clone(), |_| Some(Ok(grant(None, None)))); + let owner = script.connect().await.unwrap(); + let server = HeldPost { + script: script.clone(), + arrived: Arc::default(), + release: Arc::default(), + }; + let context = owner.context().unwrap().attach(&request(), &script.identity).unwrap(); + let pending = tokio::spawn(admit(server.clone(), request(), context)); + server.arrived.notified().await; + owner.close(Reason::Shutdown, Bytes::default()); + tokio::task::yield_now().await; + assert_eq!( + log.events(), + [Event::Connect], + "end overtook an in-flight connect for the same id" + ); + server.release.notify_one(); + assert!(matches!(pending.await.unwrap(), Err(Error::Refused))); + assert!(matches!( + log.end().await.event, + Event::End { + reason: Reason::Shutdown, + .. + } + )); + assert!(matches!( + log.events().as_slice(), + [Event::Connect, Event::Connect, Event::End { .. }] + )); + } + + #[tokio::test] + async fn an_attachment_cannot_change_connection_identity() { + let log = Log::default(); + let script = Script::new(log.clone(), |_| Some(Ok(grant(None, None)))); + let owner = script.connect().await.unwrap(); + for change in [0, 1, 2, 3] { + let mut request = request(); + match change { + 0 => request.id = "another".into(), + 1 => request.path = "/another".into(), + 2 => request.role = Some(crate::Role::Publisher), + _ => request.node = "another".into(), + } + assert!(matches!(script.attach(&owner, request).await, Err(Error::Refused))); + } + let other = Script::new(log.clone(), |_| Some(Ok(grant(None, None)))); + assert!(matches!(other.attach(&owner, request()).await, Err(Error::Refused))); + assert_eq!(log.events(), [Event::Connect]); + owner.clone().close("done", Bytes::default()); + assert!(matches!(script.attach(&owner, request()).await, Err(Error::Refused))); + } + #[tokio::test] async fn a_bare_drop_ends_as_dropped() { let log = Log::default(); @@ -573,6 +943,7 @@ mod tests { .await .expect("closed at expires, not later"); assert_eq!(reason, Reason::Expired); + consumer.close(reason, Bytes::default()); assert!(matches!( log.end().await.event, Event::End { diff --git a/rs/moq-auth/src/lease.rs b/rs/moq-auth/src/lease.rs index 023ad7a0a5..b4d659023c 100644 --- a/rs/moq-auth/src/lease.rs +++ b/rs/moq-auth/src/lease.rs @@ -6,6 +6,7 @@ //! producer side is driven by [`Client`](crate::Client) when an auth server answers, or //! by any in-process logic when the embedder decides itself. No trait, no callbacks. +use std::sync::Arc; #[cfg(feature = "tokio")] use std::sync::Mutex; use std::task::Poll; @@ -170,7 +171,7 @@ impl Producer { #[cfg(feature = "tokio")] clock_changed: tokio::sync::Notify::new(), }, - Consumer { state, seen: 0 }, + Consumer::new(state), ) } @@ -312,14 +313,42 @@ impl Drop for Producer { /// The session's side of a lease: reads the grant and learns when it changes or ends. /// -/// Dropping it ends the lease with [`Reason::Dropped`]; [`close`](Self::close) says why. -#[derive(Debug)] +/// Dropping the last clone ends the lease with [`Reason::Dropped`]; [`close`](Self::close) ends every clone. +#[derive(Debug, Clone)] pub struct Consumer { - state: kio::Shared, + hold: Arc, seen: u64, } +#[derive(Debug)] +struct Held { + state: kio::Shared, + #[cfg(feature = "client")] + context: Option, +} + impl Consumer { + fn new(state: kio::Shared) -> Self { + Self { + hold: Arc::new(Held { + state, + #[cfg(feature = "client")] + context: None, + }), + seen: 0, + } + } + + #[cfg(feature = "client")] + pub(crate) fn context(&self) -> Option<&crate::client::Context> { + self.hold.context.as_ref() + } + + #[cfg(feature = "client")] + pub(crate) fn set_context(&mut self, context: crate::client::Context) { + Arc::get_mut(&mut self.hold).expect("new lease has one holder").context = Some(context); + } + /// A lease on a grant nobody drives: it never changes and is never revoked, /// so only the holder ends it. What a static or public grant admits under. pub fn fixed(grant: Grant) -> Self { @@ -329,19 +358,20 @@ impl Consumer { closed: None, revalidate: 0, }); - Self { state, seen: 0 } + Self::new(state) } /// The grant as it stands now. pub fn grant(&self) -> Grant { - self.state.read().grant.clone() + self.hold.state.read().grant.clone() } /// Poll for the next update: the new grant, or the reason the lease ended. pub fn poll_changed(&mut self, waiter: &kio::Waiter) -> Poll> { let seen = self.seen; let state = std::task::ready!( - self.state + self.hold + .state .poll(waiter, |state| ready_if(state.epoch > seen || state.closed.is_some())) ); if let Some((reason, _)) = &state.closed { @@ -358,7 +388,7 @@ impl Consumer { /// Poll for the lease ending. pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll { - let state = std::task::ready!(self.state.poll(waiter, |state| ready_if(state.closed.is_some()))); + let state = std::task::ready!(self.hold.state.poll(waiter, |state| ready_if(state.closed.is_some()))); Poll::Ready(state.closed.as_ref().expect("waited for a close").0.clone()) } @@ -370,19 +400,28 @@ impl Consumer { /// Ask the producer to re-check now. A no-op on a [`fixed`](Self::fixed) /// lease, which has nobody to ask. pub fn revalidate(&self) { - self.state.lock().revalidate += 1; + self.hold.state.lock().revalidate += 1; } /// End the lease with the session's own close classification and the totals it - /// moved, consuming the handle, and return the reason the lease ended with: + /// moved, and return the reason the lease ended with: /// `reason`, or the producer's if it revoked first. [`Drop`] reports zero bytes. pub fn close(self, reason: impl Into, bytes: Bytes) -> Reason { - self.state.lock().closed.get_or_insert((reason.into(), bytes)).0.clone() + let reason = reason.into(); + #[cfg(feature = "client")] + if let Some(context) = &self.hold.context { + context.close(reason.clone(), bytes); + } + self.hold.state.lock().closed.get_or_insert((reason, bytes)).0.clone() } } -impl Drop for Consumer { +impl Drop for Held { fn drop(&mut self) { + #[cfg(feature = "client")] + if let Some(context) = &self.context { + context.close(Reason::Dropped, Bytes::default()); + } self.state .lock() .closed @@ -442,6 +481,31 @@ mod tests { ); } + #[test] + fn consumer_clones_hold_the_lease_until_the_last_drop() { + let (producer, consumer) = Producer::new(grant("a/**")); + let retained = consumer.clone(); + drop(consumer); + assert!(poll(producer.closed()).is_pending()); + drop(retained); + assert_eq!( + poll(producer.closed()), + Poll::Ready((Reason::Dropped, Bytes::default())) + ); + } + + #[test] + fn closing_one_consumer_ends_every_clone() { + let (producer, consumer) = Producer::new(grant("a/**")); + let retained = consumer.clone(); + consumer.close(Reason::Shutdown, Bytes { sent: 3, received: 5 }); + assert_eq!(poll(retained.closed()), Poll::Ready(Reason::Shutdown)); + assert_eq!( + poll(producer.closed()), + Poll::Ready((Reason::Shutdown, Bytes { sent: 3, received: 5 })) + ); + } + #[test] fn dropping_the_producer_revokes() { let (producer, consumer) = Producer::new(grant("a/**")); diff --git a/rs/moq-auth/src/serve.rs b/rs/moq-auth/src/serve.rs index 9b2c379993..f7e6d2a5b3 100644 --- a/rs/moq-auth/src/serve.rs +++ b/rs/moq-auth/src/serve.rs @@ -256,7 +256,8 @@ fn remote(request: &Request) -> Option { /// One live session's share of the limits. struct Slot { - token: Option, + /// Every JWT presented for this session: its connect, plus each one attached in band. + tokens: Vec, remote: Option, seen: Instant, } @@ -275,20 +276,25 @@ impl Sessions { .retain(|_, slot| now.saturating_duration_since(slot.seen) < cadence.saturating_mul(2)); } - /// Admit a `connect`, refusing over the cap. A known id refreshes instead. + /// Admit a `connect`, refusing over the cap. A known id is the same session + /// attaching another credential: it holds no new remote slot, but a JWT new to + /// it counts against that JWT's cap until the session ends. fn connect(&mut self, request: &Request, limits: Limits) -> Result<(), Refusal> { - if let Some(slot) = self.slots.get_mut(&request.id) { - slot.seen = Instant::now(); - return Ok(()); - } // Decided before counting, so the credential is already known to be one JWT. let token = jwt(request).ok().flatten().map(hash); - let remote = remote(request); - if let (Some(cap), Some(token)) = (limits.token, token) - && self.slots.values().filter(|slot| slot.token == Some(token)).count() >= cap + let known = self.slots.get(&request.id); + let new_token = token.filter(|token| !known.is_some_and(|slot| slot.tokens.contains(token))); + if let (Some(cap), Some(token)) = (limits.token, new_token) + && self.slots.values().filter(|slot| slot.tokens.contains(&token)).count() >= cap { return Err(Refusal::TokenLimit); } + if let Some(slot) = self.slots.get_mut(&request.id) { + slot.tokens.extend(new_token); + slot.seen = Instant::now(); + return Ok(()); + } + let remote = remote(request); if let (Some(cap), Some(remote)) = (limits.remote, remote) && self.slots.values().filter(|slot| slot.remote == Some(remote)).count() >= cap { @@ -297,7 +303,7 @@ impl Sessions { self.slots.insert( request.id.clone(), Slot { - token, + tokens: token.into_iter().collect(), remote, seen: Instant::now(), }, @@ -309,11 +315,17 @@ impl Sessions { /// lost to a restart. The cap is not enforced here: the session was admitted, and /// which survivor to revoke would be an accident of arrival order. fn revalidate(&mut self, request: &Request) { + let token = jwt(request).ok().flatten().map(hash); let slot = self.slots.entry(request.id.clone()).or_insert_with(|| Slot { - token: jwt(request).ok().flatten().map(hash), + tokens: Vec::new(), remote: remote(request), seen: Instant::now(), }); + if let Some(token) = token + && !slot.tokens.contains(&token) + { + slot.tokens.push(token); + } slot.seen = Instant::now(); } @@ -1118,6 +1130,50 @@ mod tests { assert_eq!(server.answer(&same_host).await.unwrap_err(), Refusal::RemoteLimit); } + /// An in-band credential arrives as a `connect` for a live session id. It takes + /// no new remote slot, but counts against its own JWT's cap like a session would. + #[tokio::test] + async fn attached_tokens_count_against_their_own_cap() { + let (dir, key) = key_dir(); + let server = Server::new(Policy { + keys: Some(Keys::Dir(dir.path().into())), + limits: Limits { + token: Some(1), + remote: None, + }, + ..limited() + }) + .unwrap(); + let exp = Some(SystemTime::now() + Duration::from_secs(60)); + let a = sign(&key, "demo", &["**"], &[], None); + let b = sign(&key, "demo", &["**"], &[], exp); + let c = sign( + &key, + "demo", + &["**"], + &[], + Some(SystemTime::now() + Duration::from_secs(61)), + ); + + let mut first = with_token(request("/demo"), &a); + first.id = "first".into(); + let mut second = with_token(request("/demo"), &c); + second.id = "second".into(); + server.answer(&first).await.unwrap(); + server.answer(&second).await.unwrap(); + + let attach = |session: &Request| { + let mut attach = with_token(session.clone(), &b); + attach.id = session.id.clone(); + attach + }; + server.answer(&attach(&first)).await.unwrap(); + // Re-presenting a JWT the session already holds is not another use of it. + server.answer(&attach(&first)).await.unwrap(); + assert_eq!(server.answer(&attach(&second)).await.unwrap_err(), Refusal::TokenLimit); + assert_eq!(server.sessions.lock().unwrap().slots.len(), 2); + } + #[tokio::test] async fn the_client_admits_and_reads_the_refusal() { let (client, ended) = serve(Policy { diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 49ebf27a42..13c72b870c 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -136,3 +136,7 @@ required-features = ["fuzz"] name = "retirement" harness = false required-features = ["fuzz"] + +[[bench]] +name = "auth" +harness = false diff --git a/rs/moq-net/benches/auth.rs b/rs/moq-net/benches/auth.rs new file mode 100644 index 0000000000..044c88254a --- /dev/null +++ b/rs/moq-net/benches/auth.rs @@ -0,0 +1,102 @@ +//! AUTH admission bursts across sessions and tokens per session, without sockets. +//! +//! Measures the pending queue and complete grant/withdrawal lifecycle. Run the same +//! matrix before and after queue changes so a per-session scan appears on its own axis. + +#[path = "../tests/support/mod.rs"] +mod support; + +use bytes::Bytes; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use futures::future::join_all; +use moq_net::{Session, Version, auth}; +use std::time::Duration; +use support::{harness, mock::create_mock_session_pair}; + +struct Pair { + client: Session, + _server: Session, + requests: auth::Requests, + _initial: auth::Issued, +} + +async fn pair(version: Version) -> Pair { + let (client, server) = create_mock_session_pair(Some(version.alpn())); + let client = async { + let (session, driver) = moq_net::Client::new() + .with_versions(version.into()) + .connect(harness::now(), client) + .await + .unwrap(); + harness::spawn(driver); + session + }; + let server = async { + let request = moq_net::Server::new() + .with_versions(version.into()) + .accept_request(harness::now(), server) + .await + .unwrap(); + let requests = request.auth().requests().unwrap(); + let (session, driver) = request.ok().await.unwrap(); + harness::spawn(driver); + (session, requests) + }; + let (client, (server, mut requests)) = futures::join!(client, server); + let initial = requests.next().await.unwrap().accept(auth::Grant::all()); + Pair { + client, + _server: server, + requests, + _initial: initial, + } +} + +impl Pair { + async fn round(&mut self, count: usize) { + let handle = self.client.auth(); + let present = join_all((0..count).map(|_| handle.add(Bytes::from_static(b"token")))); + let accept = async { + let mut issued = Vec::with_capacity(count); + for _ in 0..count { + issued.push(self.requests.next().await.unwrap().accept(auth::Grant::all())); + } + issued + }; + let (tokens, issued) = futures::join!(present, accept); + let tokens: Vec<_> = tokens.into_iter().map(Result::unwrap).collect(); + drop(tokens); + join_all(issued.iter().map(auth::Issued::closed)).await; + } +} + +fn admission(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + for version in ["moq-lite-07-wip", "moq-transport-22"] { + let mut group = c.benchmark_group(format!("auth/{version}")); + group.sample_size(10); + group.warm_up_time(Duration::from_millis(100)); + group.measurement_time(Duration::from_millis(300)); + for sessions in [1, 16, 64] { + for tokens in [1, 16, 128] { + group.throughput(Throughput::Elements((sessions * tokens) as u64)); + group.bench_with_input( + BenchmarkId::new(sessions.to_string(), tokens), + &(sessions, tokens), + |b, &(sessions, tokens)| { + let mut pairs = + runtime.block_on(join_all((0..sessions).map(|_| pair(version.parse().unwrap())))); + b.iter(|| runtime.block_on(join_all(pairs.iter_mut().map(|pair| pair.round(tokens))))); + }, + ); + } + } + group.finish(); + } +} + +criterion_group!(benches, admission); +criterion_main!(benches); diff --git a/rs/moq-net/src/auth.rs b/rs/moq-net/src/auth.rs index 2ae193a0c4..cf036f7d98 100644 --- a/rs/moq-net/src/auth.rs +++ b/rs/moq-net/src/auth.rs @@ -159,7 +159,7 @@ enum Acceptor { #[default] Undecided, /// The app answers every token. - App(kio::Queue), + App(Pending), /// The session answers from its own origin handles. Default, } @@ -335,6 +335,10 @@ impl Handle { /// origin handles allow for the connection's credential, and refuses any other /// token as unsupported. /// + /// At most 128 unanswered tokens queue, not counting withdrawn ones; more are + /// refused with [`SessionError::TooManyRequests`] (`UNAUTHORIZED` on moq-transport, + /// whose AUTH_ERROR registry has no such code). + /// /// # Errors /// /// [`Error::Duplicate`] if the requests were already taken, or the driver @@ -344,7 +348,7 @@ impl Handle { if !matches!(state.acceptor, Acceptor::Undecided) { return Err(Error::Duplicate); } - let queue = kio::Queue::new(); + let queue = Pending::default(); // A version without AUTH never receives a token, so its requests end with // the session like any other. if !state.supported { @@ -356,7 +360,7 @@ impl Handle { /// Decide who answers the peer's tokens: the app if it took the requests, /// otherwise the session itself (`None`). Idempotent. - pub(crate) fn acceptor(&self) -> Option> { + pub(crate) fn acceptor(&self) -> Option { let mut state = self.state.lock(); match &state.acceptor { Acceptor::App(queue) => Some(queue.clone()), @@ -716,27 +720,83 @@ impl Watch { /// /// Returned by [`Handle::requests`]. Dropping it refuses every later token. pub struct Requests { - queue: kio::Queue, + queue: Pending, } impl Requests { /// Poll for the next token, or `None` once the session ends. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - self.queue.poll_pop(waiter).map(|res| res.ok()) + self.queue.poll_next(waiter) } /// Wait for the next token the peer presents, or `None` once the session ends. pub async fn next(&mut self) -> Option { - self.queue.pop().await.ok() + kio::wait(|waiter| self.poll_next(waiter)).await } } impl Drop for Requests { fn drop(&mut self) { self.queue.close(); - // The session keeps a handle to the queue, so drop what is already queued here: - // each request refuses its token as it drops. - while let Ok(Some(_)) = self.queue.try_pop() {} + } +} + +/// Pending application decisions, bounded independently of transport stream credit. +#[derive(Clone, Default)] +pub(crate) struct Pending { + state: kio::Shared, +} + +#[derive(Default)] +struct PendingState { + requests: VecDeque, + closed: bool, +} + +impl Pending { + fn close(&self) { + let mut state = self.state.lock(); + state.closed = true; + state.requests.clear(); + } + + /// Queue the token for the application, or refuse it when closed or full. + pub(crate) fn push(&self, request: Request) { + let mut state = self.state.lock(); + // Dropping the request refuses it as unanswered. + if state.closed { + return; + } + // A withdrawn stream must not occupy a slot while the verifier is busy. + if state.requests.len() >= 128 { + state.requests.retain(|request| !request.ended()); + } + if state.requests.len() >= 128 { + drop(state); + request.reject(SessionError::TooManyRequests, "too many pending tokens"); + return; + } + state.requests.push_back(request); + } + + fn poll_next(&self, waiter: &kio::Waiter) -> Poll> { + loop { + let mut state = std::task::ready!(self.state.poll(waiter, |state| { + if state.closed || !state.requests.is_empty() { + Poll::Ready(()) + } else { + Poll::Pending + } + })); + while let Some(request) = state.requests.pop_front() { + if !request.ended() { + return Poll::Ready(Some(request)); + } + } + if state.closed { + return Poll::Ready(None); + } + } } } @@ -810,6 +870,15 @@ impl Request { &self.token } + fn ended(&self) -> bool { + self.issue.as_ref().expect("unanswered request").read().peer.is_some() + } + + /// Wait for the peer to withdraw this unanswered token or close the session. + pub async fn closed(&self) -> Error { + closed(self.issue.as_ref().expect("unanswered request")).await + } + /// Grant the token. The grant holds until the returned [`Issued`] is revoked /// or dropped, or the peer withdraws the token. /// @@ -867,17 +936,24 @@ impl Issued { /// Wait for the peer to withdraw the token ([`Error::Cancel`]), or for its /// stream or the session to end. pub async fn closed(&self) -> Error { - kio::wait(|waiter| { - let issue = ready_or!(self.issue.poll(waiter, |issue| match issue.peer.is_some() { - true => Poll::Ready(()), - false => Poll::Pending, - })); - Poll::Ready(issue.peer.clone().expect("checked above")) - }) - .await + closed(&self.issue).await } } +async fn closed(issue: &kio::Shared) -> Error { + kio::wait(|waiter| { + let issue = ready_or!(issue.poll(waiter, |issue| { + if issue.peer.is_some() { + Poll::Ready(()) + } else { + Poll::Pending + } + })); + Poll::Ready(issue.peer.clone().expect("checked above")) + }) + .await +} + impl Drop for Issued { fn drop(&mut self) { self.issue.lock().done = true; @@ -1040,6 +1116,45 @@ impl Enforce { mod tests { use super::*; + #[test] + fn pending_tokens_are_bounded() { + let handle = Handle::new(true); + let _requests = handle.requests().unwrap(); + let queue = handle.acceptor().unwrap(); + for _ in 0..128 { + queue.push(Request::new(Bytes::new(), Default::default())); + } + let overflow = kio::Shared::::default(); + queue.push(Request::new(Bytes::new(), overflow.clone())); + let overflow = overflow.lock(); + assert!(overflow.done); + assert!(matches!( + overflow.outbox.front(), + Some(Reply::Refuse { + code: SessionError::TooManyRequests, + .. + }) + )); + } + + #[test] + fn withdrawn_tokens_do_not_occupy_pending_slots() { + let handle = Handle::new(true); + let mut requests = handle.requests().unwrap(); + let queue = handle.acceptor().unwrap(); + let expired = kio::Shared::::default(); + for _ in 0..128 { + queue.push(Request::new(Bytes::from_static(b"withdrawn"), expired.clone())); + } + expired.lock().peer = Some(Error::Cancel); + queue.push(Request::new(Bytes::from_static(b"live"), Default::default())); + let Poll::Ready(Some(request)) = requests.poll_next(&kio::Waiter::noop()) else { + panic!("live request missing") + }; + assert_eq!(request.token().as_ref(), b"live"); + assert!(requests.poll_next(&kio::Waiter::noop()).is_pending()); + } + fn patterns(prefixes: &[&str]) -> Patterns { prefixes .iter() diff --git a/rs/moq-net/src/ietf/auth.rs b/rs/moq-net/src/ietf/auth.rs index f7f41c2b8f..f24452f7c3 100644 --- a/rs/moq-net/src/ietf/auth.rs +++ b/rs/moq-net/src/ietf/auth.rs @@ -433,8 +433,7 @@ impl Serve { let mut default = None; match self.handle.acceptor() { Some(requests) => { - // A closed queue hands the request back, and dropping it refuses the token. - let _ = requests.try_push(Request::new(msg.token, issue.clone())); + requests.push(Request::new(msg.token, issue.clone())); } // Only the connection's own credential has a default answer; a token needs // someone to verify it. diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 1aaa4f236c..0d8feb0d3a 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -462,9 +462,7 @@ impl AuthServe { let issue = serving.issue.clone(); match self.shared.auth.acceptor() { Some(requests) => { - // A closed queue hands the request back, and dropping it refuses - // the token on this stream. - let _ = requests.try_push(crate::auth::Request::new(msg.token, issue.clone())); + requests.push(crate::auth::Request::new(msg.token, issue.clone())); } // Only the connection's own credential has a default answer; a token // needs someone to verify it. Tell the presenter explicitly. diff --git a/rs/moq-relay/Cargo.toml b/rs/moq-relay/Cargo.toml index 27e906eb1c..602d70ae1a 100644 --- a/rs/moq-relay/Cargo.toml +++ b/rs/moq-relay/Cargo.toml @@ -113,3 +113,7 @@ libc = "0.2" [[bench]] name = "http_idle" harness = false + +[[bench]] +name = "auth" +harness = false diff --git a/rs/moq-relay/benches/auth.rs b/rs/moq-relay/benches/auth.rs new file mode 100644 index 0000000000..138dc1d675 --- /dev/null +++ b/rs/moq-relay/benches/auth.rs @@ -0,0 +1,147 @@ +//! Revalidate one token while sweeping unrelated sessions and held credentials. +//! The real relay/controller runs over loopback TCP; timings include one AUTH_OK. + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use moq_auth::{Grant, lease}; +use moq_relay::{Connection, auth, cluster}; +use std::time::Duration; + +fn grant(prefix: &str) -> Grant { + Grant::new( + [format!("{prefix}/**").parse().unwrap()].into_iter().collect(), + Default::default(), + ) +} + +struct Fixture { + _connections: Vec, + _tokens: Vec, + producers: Vec, + selected: usize, + watch: moq_tokio::connection::auth::Watch, + tasks: Vec>, + alternate: bool, +} + +impl Drop for Fixture { + fn drop(&mut self) { + for task in &self.tasks { + task.abort(); + } + } +} + +impl Fixture { + async fn new(sessions: usize, tokens: usize) -> Self { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let (auth, mut admissions) = auth::Auth::embedded("bench"); + let (send, mut receive) = tokio::sync::mpsc::unbounded_channel(); + let decider = tokio::spawn(async move { + while let Some(admission) = admissions.next().await { + let prefix = admission + .request + .query + .as_deref() + .unwrap_or("initial") + .strip_prefix("jwt=") + .unwrap_or("initial") + .to_owned(); + let (producer, consumer) = lease::Producer::new(grant(&prefix)); + admission.grant(consumer); + if send.send((prefix, producer)).is_err() { + break; + } + } + }); + let mut config = moq_tokio::listen::Config::default(); + config.tcp.bind = Some("127.0.0.1:0".parse().unwrap()); + let mut server = config.init(Default::default()).unwrap().listen().await.unwrap(); + let port = server.tcp_local_addr().unwrap().port(); + let cluster = cluster::Cluster::new(Default::default()).unwrap(); + let listener = tokio::spawn(async move { + let mut serving = tokio::task::JoinSet::new(); + while let Some(request) = server.accept().await { + let connection = Connection::new(request, cluster.clone(), auth.clone()); + serving.spawn(async move { + let _ = connection.run().await; + }); + } + }); + let mut config = moq_tokio::connect::Config::default(); + config.once = Some(true); + config.version = vec!["moq-transport-22".parse().unwrap()]; + let client = config.init(Default::default()).unwrap(); + let url: url::Url = format!("tcp://127.0.0.1:{port}/").parse().unwrap(); + let mut connections = Vec::new(); + let mut retained = Vec::new(); + let mut producers = Vec::new(); + let mut selected = 0; + let mut watch = None; + for session in 0..sessions { + let connection = client.clone().connect(url.clone()).established().await.unwrap(); + producers.push(receive.recv().await.unwrap().1); + let auth = connection.auth(); + let added = futures::future::join_all((0..tokens).map(|index| auth.add(format!("token{index}")))).await; + let added: Vec<_> = added.into_iter().map(Result::unwrap).collect(); + if session == 0 { + let mut observed = added.last().unwrap().grant(); + observed.changed().await.unwrap(); + watch = Some(observed); + } + for _ in 0..tokens { + let (prefix, producer) = receive.recv().await.unwrap(); + if session == 0 && prefix == format!("token{}", tokens - 1) { + selected = producers.len(); + } + producers.push(producer); + } + retained.extend(added); + connections.push(connection); + } + Self { + _connections: connections, + _tokens: retained, + producers, + selected, + watch: watch.unwrap(), + tasks: vec![listener, decider], + alternate: false, + } + } + + async fn round(&mut self) { + self.alternate = !self.alternate; + let expected = grant(if self.alternate { "changed" } else { "restored" }); + self.producers[self.selected].update(expected.clone()); + let actual = self.watch.changed().await.unwrap().unwrap(); + assert_eq!(actual.publish, expected.publish); + } +} + +fn refresh(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let mut group = c.benchmark_group("relay_auth_refresh"); + group.sample_size(10); + group.warm_up_time(Duration::from_millis(100)); + group.measurement_time(Duration::from_millis(300)); + for sessions in [1, 16, 64] { + // QMux grants 100 bidirectional streams, including persistent control streams. + for tokens in [1, 16, 64] { + let mut fixture = runtime.block_on(Fixture::new(sessions, tokens)); + group.bench_with_input( + BenchmarkId::new(sessions.to_string(), tokens), + &(sessions, tokens), + |b, _| { + b.iter(|| runtime.block_on(fixture.round())); + }, + ); + } + } + group.finish(); +} + +criterion_group!(benches, refresh); +criterion_main!(benches); diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 9d84805be8..8861999b5f 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -3,6 +3,8 @@ //! process answering [`Admissions`]. Nothing else admits anyone; a verified client //! certificate is reported to whoever decides as a fact. +mod refresh; + use std::sync::Arc; use axum::http; @@ -301,7 +303,8 @@ impl Config { ), (None, None) => unreachable!("validated above"), }; - let (auth, admissions) = Auth::embedded(node); + let (mut auth, admissions) = Auth::embedded(node); + auth.in_band = matches!(decider, Decider::Server(_)); decider.spawn(admissions); Ok(auth) } @@ -436,6 +439,8 @@ impl Token { /// a fixed grant. [`ended`](Self::ended) resolves when that word no longer covers /// the session; the holder then closes the session, and the lease with the reason. pub struct Lease { + refresh: Option, + source: Option, consumer: lease::Consumer, token: Token, /// The live session a re-check re-authorizes in place, once attached. @@ -449,12 +454,19 @@ pub struct Lease { stats: moq_net::stats::Session, } +struct Source { + auth: Auth, + request: Request, +} + impl Lease { /// Hold `consumer` for a session that dialed `path`, reducing its grant to /// what the origin scopes by. pub fn new(path: &str, consumer: lease::Consumer) -> Self { let grant = consumer.grant(); Self { + refresh: None, + source: None, token: Token::new(path, &grant), expires: grant.deadline(), consumer, @@ -463,6 +475,16 @@ impl Lease { } } + /// Take AUTH decisions before the protocol driver chooses its default acceptor. + pub(crate) fn preparing( + mut self, + handle: moq_net::auth::Handle, + role: Option, + ) -> Result { + self.refresh = Some(refresh::Refresh::new(&mut self, handle, role)?); + Ok(self) + } + /// Re-authorize `session` in place when a re-check changes what the grant allows, /// narrower or wider. Without it a narrower grant ends the lease. pub(crate) fn authorizing(mut self, session: &moq_net::Session) -> Self { @@ -484,7 +506,11 @@ impl Lease { /// Ask the lease's producer to re-check now. A no-op on a fixed lease. pub fn revalidate(&self) { - self.consumer.revalidate(); + if let Some(refresh) = &self.refresh { + refresh.revalidate(); + } else { + self.consumer.revalidate(); + } } /// Wait for the lease to stop covering the session: the grant expired, was @@ -498,7 +524,13 @@ impl Lease { /// 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. + /// + /// Once the relay answers the session's AUTH requests itself, each of + /// these ends only that credential's grant; the lease ends when no credential remains. pub async fn ended(&mut self) -> lease::Reason { + if let Some(refresh) = &mut self.refresh { + return refresh.ended().await; + } loop { let expire = async { match self.expires { @@ -562,6 +594,14 @@ impl Lease { /// moved, and learn what it ended with: that, or the decider's reason if it /// revoked first. Dropping the consumer reports zero bytes. pub fn close(self, reason: impl Into, bytes: Bytes) -> lease::Reason { + let mut reason = reason.into(); + // A gateway has one credential, so its revocation caused the transfer to + // end. A refreshed session instead reports the result of its whole union. + if self.refresh.is_none() + && let std::task::Poll::Ready(ended) = self.consumer.poll_closed(&moq_net::kio::Waiter::noop()) + { + reason = ended; + } self.consumer.close(reason, bytes) } } @@ -609,7 +649,11 @@ impl Decider { Self::Server(client) => { let client = client.clone(); tokio::spawn(async move { - match client.connect(admission.request.clone()).await { + let lease = match &admission.connection { + Some(connection) => client.attach(connection, admission.request.clone()).await, + None => client.connect(admission.request.clone()).await, + }; + match lease { Ok(lease) => admission.grant(lease), Err(err) => admission.refuse(err.into()), } @@ -656,6 +700,8 @@ fn presents_token(request: &Request) -> bool { pub struct Auth { admissions: mpsc::UnboundedSender, node: Arc, + /// Whether the decider can verify a token presented in band; public rules cannot. + in_band: bool, } impl Auth { @@ -667,6 +713,7 @@ impl Auth { let auth = Self { admissions: sender, node: Arc::from(node.into()), + in_band: true, }; (auth, Admissions(receiver)) } @@ -692,19 +739,34 @@ impl Auth { /// Admit a session: the lease it holds, carrying the scope the origin applies. pub async fn admit(&self, request: Request) -> Result { - let path = request.path.clone(); + let consumer = self.decide(request.clone()).await?; + let mut lease = Lease::new(&request.path, consumer); + lease.source = Some(Source { + auth: self.clone(), + request, + }); + Ok(lease) + } + + async fn decide(&self, request: Request) -> Result { + let answer = self.submit(request, None)?; + let answered = tokio::time::timeout(ADMIT_TIMEOUT, answer) + .await + .map_err(|_| timed_out())?; + answered_with(answered) + } + + /// Queue one admission for the decider, returning its pending answer. + fn submit(&self, request: Request, connection: Option) -> Result { let (reply, answer) = oneshot::channel(); self.admissions - .send(Admission { request, reply }) + .send(Admission { + request, + connection, + reply, + }) .map_err(|_| Error::Unavailable("nobody is answering admissions".into()))?; - let consumer = tokio::time::timeout(ADMIT_TIMEOUT, answer) - .await - .map_err(|_| Error::Unavailable("the admission timed out".into()))? - .map_err(|_| Error::Unavailable("the admission went unanswered".into()))??; - // Every decider is held to the same answer contract: a grant that admits - // nothing or asks for a re-check without a bound is a bug, not a refusal. - consumer.grant().validate()?; - Ok(Lease::new(&path, consumer)) + Ok(answer) } /// Admit a session on a grant decided locally, bypassing the server: the LAN @@ -714,11 +776,31 @@ impl Auth { } } +/// A decider's pending answer to one [`Admission`]. +type Answer = oneshot::Receiver>; + +fn timed_out() -> Error { + Error::Unavailable("the admission timed out".into()) +} + +/// Check a decider's answer against the contract every decider is held to: a +/// grant that admits nothing or asks for a re-check without a bound is a bug, not +/// a refusal. +fn answered_with( + answered: Result, oneshot::error::RecvError>, +) -> Result { + let consumer = answered.map_err(|_| Error::Unavailable("the admission went unanswered".into()))??; + consumer.grant().validate()?; + Ok(consumer) +} + /// The sessions an embedded [`Auth`] is waiting to admit, in arrival order. /// /// The transport has already accepted each one; it waits for its answer, so a /// slow decider holds connects the way a slow auth server would, and one that -/// takes longer than a server may (ten seconds) is refused as unavailable. +/// takes longer than a server may (ten seconds) is refused as unavailable. An +/// [attached](Admission::attached) token keeps its place in the session's +/// verification bound until the decider answers or drops it, even past that. /// Answer in place or hand each [`Admission`] to its own task; nothing here /// serializes them. pub struct Admissions(mpsc::UnboundedReceiver); @@ -736,6 +818,7 @@ impl Admissions { pub struct Admission { /// The `connect` request the relay built: every fact the transport knows. pub request: Request, + connection: Option, reply: oneshot::Sender>, } @@ -748,6 +831,12 @@ impl Admission { } impl Admission { + /// Whether this verifies another credential presented in band for the session + /// already admitted as `request.id`, rather than admitting a new session. + pub fn attached(&self) -> bool { + self.connection.is_some() + } + /// Admit the session on `lease`: [`lease::Consumer::fixed`] for a grant that /// never changes, or the consumer of a [`lease::Producer`] the decider keeps to /// re-check, update, revoke, and learn when the session ends. diff --git a/rs/moq-relay/src/auth/refresh.rs b/rs/moq-relay/src/auth/refresh.rs new file mode 100644 index 0000000000..0c50dde4a9 --- /dev/null +++ b/rs/moq-relay/src/auth/refresh.rs @@ -0,0 +1,708 @@ +use std::collections::BTreeMap; + +use futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered}; +use moq_net::{Role, SessionError, auth}; + +use super::{Grant, Source, Token, lease}; + +/// A connection owns its lifecycle while each credential owns only its grant. +pub(super) struct Refresh { + handle: auth::Handle, + requests: auth::Requests, + requests_open: bool, + owner: lease::Consumer, + source: Option, + admitted: Token, + stats: moq_net::stats::Session, + role: Option, + credentials: BTreeMap, + /// At most 128 tokens being verified, counting `settling`; more are refused with + /// `TooManyRequests`. A withdrawn token keeps its slot until the decider answers. + pending: FuturesUnordered, Option)>>, + /// Decisions that outlived the admission timeout: the token was already refused, + /// but the decider is still working, so its slot stays held until it answers. + settling: FuturesUnordered, + next: u64, +} + +type Verified = (auth::Request, Result); +type Settling = BoxFuture<'static, ()>; + +struct Credential { + consumer: lease::Consumer, + grant: Grant, + expires: Option, + issued: Option, + initial: bool, +} + +enum Change { + Grant(Grant), + Ended(lease::Reason), + Withdrawn, +} + +impl Credential { + fn new(consumer: lease::Consumer) -> Self { + let grant = consumer.grant(); + Self { + expires: grant.deadline(), + consumer, + grant, + issued: None, + initial: false, + } + } + + async fn changed(&mut self) -> Change { + let expire = async { + match self.expires { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending().await, + } + }; + let withdrawn = async { + match &self.issued { + Some(issued) => { + issued.closed().await; + } + None => std::future::pending().await, + } + }; + tokio::select! { + biased; + () = withdrawn => Change::Withdrawn, + () = expire => Change::Ended(lease::Reason::Expired), + changed = self.consumer.changed() => match changed { + Ok(grant) => Change::Grant(grant), + Err(reason) => Change::Ended(reason), + }, + } + } + + fn wire(&self, role: Option) -> auth::Grant { + auth::Grant { + publish: if role == Some(Role::Subscriber) { + Default::default() + } else { + self.grant.publish.clone() + }, + subscribe: if role == Some(Role::Publisher) { + Default::default() + } else { + self.grant.subscribe.clone() + }, + expires: self.expires.map(tokio::time::Instant::into_std), + } + } +} + +impl Refresh { + pub(super) fn new( + lease: &mut super::Lease, + handle: auth::Handle, + role: Option, + ) -> Result { + let requests = handle.requests()?; + let mut initial = Credential::new(lease.consumer.clone()); + initial.initial = true; + let this = Self { + handle, + requests, + requests_open: true, + owner: lease.consumer.clone(), + source: lease.source.take(), + admitted: lease.token.clone(), + stats: lease.stats.clone(), + role, + credentials: [(0, initial)].into(), + pending: FuturesUnordered::new(), + settling: FuturesUnordered::new(), + next: 1, + }; + this.authorize(); + Ok(this) + } + + fn valid(&self, grant: &Grant) -> bool { + let fresh = self.admitted.recheck(grant); + fresh.root == self.admitted.root + && fresh.mounts == self.admitted.mounts + && fresh.peer == self.admitted.peer + && fresh.upstream == self.admitted.upstream + && match self.role { + Some(Role::Publisher) => !grant.publish.is_empty(), + Some(Role::Subscriber) => !grant.subscribe.is_empty(), + Some(_) => false, + None => !grant.publish.is_empty() || !grant.subscribe.is_empty(), + } + } + + fn authorize(&self) { + let mut union = auth::Grant::default(); + for credential in self.credentials.values() { + let grant = credential.wire(self.role); + union.publish.extend(grant.publish.iter().cloned()); + union.subscribe.extend(grant.subscribe.iter().cloned()); + } + // Credential timers remove one grant at a time; no single deadline may + // expire a union that another credential still covers. + self.handle.authorize(&union); + } + + fn request(&mut self, request: auth::Request) { + if request.token().is_empty() { + // The connection's own lease answers an empty token while it lives, even + // after an earlier empty-token stream was withdrawn. + if self.owner.closed().now_or_never().is_some() { + request.reject(SessionError::Unauthorized, "connection credential ended"); + return; + } + let unissued = self + .credentials + .values_mut() + .find(|credential| credential.initial && credential.issued.is_none()); + if let Some(initial) = unissued { + initial.issued = Some(request.accept(initial.wire(self.role))); + return; + } + // The lease may have revalidated to a grant this session can no longer + // use while another credential keeps the session alive. Validate the + // exact snapshot that would be issued. + let mut credential = Credential::new(self.owner.clone()); + if !self.valid(&credential.grant) { + request.reject(SessionError::Unauthorized, "connection credential no longer valid"); + return; + } + credential.initial = true; + credential.issued = Some(request.accept(credential.wire(self.role))); + self.credentials.insert(self.next, credential); + self.next += 1; + self.authorize(); + return; + } + let Some(source) = self.source.as_ref().filter(|source| source.auth.in_band) else { + request.reject(SessionError::Unsupported, "tokens are not verified in band"); + return; + }; + if source.request.tls.is_some() { + request.reject(SessionError::Unsupported, "a certificate cannot be renewed in band"); + return; + } + if self.pending.len() + self.settling.len() >= 128 { + request.reject(SessionError::TooManyRequests, "too many tokens being verified"); + return; + } + let Ok(token) = std::str::from_utf8(request.token()) else { + request.reject(SessionError::Unauthorized, "token is not UTF-8"); + return; + }; + let mut facts = source.request.clone(); + facts.query = Some(with_jwt(facts.query.as_deref(), token)); + facts.token = None; + let mut answer = match source.auth.submit(facts, Some(self.owner.clone())) { + Ok(answer) => answer, + Err(err) => return refuse(request, err), + }; + self.pending.push(Box::pin(async move { + // Withdrawing the stream cannot cancel the decider's work. Keep its slot + // until the decider answers so churn cannot bypass the bound. + let (result, settling) = match tokio::time::timeout(super::ADMIT_TIMEOUT, &mut answer).await { + Ok(answered) => (super::answered_with(answered), None), + Err(_) => { + let settling: Settling = Box::pin(async move { + let _ = answer.await; + }); + (Err(super::timed_out()), Some(settling)) + } + }; + let verdict = request.closed().now_or_never().is_none().then_some((request, result)); + (verdict, settling) + })); + } + + pub(super) fn revalidate(&self) { + for credential in self.credentials.values() { + credential.consumer.revalidate(); + } + } + + pub(super) async fn ended(&mut self) -> lease::Reason { + loop { + enum Event { + Request(Option), + Verified(Option, Option), + Settled, + Changed(u64, Change), + } + let event = { + let mut changes: FuturesUnordered<_> = self + .credentials + .iter_mut() + .map(|(id, credential)| async move { (*id, credential.changed().await) }) + .collect(); + tokio::select! { + biased; + Some((id, change)) = changes.next(), if !changes.is_empty() => Event::Changed(id, change), + Some((verdict, settling)) = self.pending.next(), if !self.pending.is_empty() => Event::Verified(verdict, settling), + Some(()) = self.settling.next(), if !self.settling.is_empty() => Event::Settled, + request = self.requests.next(), if self.requests_open => Event::Request(request), + } + }; + match event { + Event::Request(Some(request)) => self.request(request), + Event::Request(None) => self.requests_open = false, + Event::Settled => {} + Event::Verified(verdict, settling) => { + self.settling.extend(settling); + let Some((request, result)) = verdict else { + continue; + }; + // Validate the exact snapshot that is issued, not a later read. + let credential = result.map(Credential::new); + match credential { + Ok(mut credential) if self.valid(&credential.grant) => { + credential.issued = Some(request.accept(credential.wire(self.role))); + self.credentials.insert(self.next, credential); + self.next += 1; + self.authorize(); + } + Ok(_) => request.reject( + SessionError::Unauthorized, + format!( + "token must retain root {}, connection routing, and usable SETUP role permissions", + self.admitted.root + ), + ), + Err(err) => refuse(request, err), + } + } + Event::Changed(id, Change::Grant(grant)) if self.valid(&grant) => { + let credential = self.credentials.get_mut(&id).expect("observed credential"); + credential.expires = grant.deadline(); + if credential.initial { + self.stats.set_tier(crate::configured_tier(grant.tier.clone())); + } + credential.grant = grant; + if let Some(issued) = &credential.issued { + issued.update(credential.wire(self.role)); + } + self.authorize(); + } + Event::Changed(id, change) => { + let credential = self.credentials.remove(&id).expect("observed credential"); + let withdrawn = matches!(change, Change::Withdrawn); + let reason = match change { + Change::Ended(reason) => reason, + Change::Withdrawn => lease::Reason::Dropped, + Change::Grant(_) => lease::Reason::Narrowed, + }; + if !withdrawn && let Some(issued) = credential.issued { + issued.revoke(SessionError::Unauthorized, reason.to_string()); + } + self.authorize(); + if self.credentials.is_empty() { + return reason; + } + } + } + } + } +} + +/// Refuse a token the decider did not admit. An auth server that could not decide +/// is no verdict on the token, so the presenter may present it again. +fn refuse(request: auth::Request, err: super::Error) { + match err { + super::Error::Unavailable(reason) => { + tracing::debug!(%reason, "in-band token not verified; auth server unavailable"); + request.reject(SessionError::Internal, "auth server unavailable"); + } + _ => request.reject(SessionError::Unauthorized, "token refused"), + } +} + +/// Replace the `jwt` pair, leaving every other byte of the dialed query as it was, +/// since the auth server may interpret or authenticate its raw form. +fn with_jwt(query: Option<&str>, token: &str) -> String { + let jwt = format!( + "jwt={}", + url::form_urlencoded::byte_serialize(token.as_bytes()).collect::() + ); + let mut pairs = Vec::new(); + let mut replaced = false; + let query = query.unwrap_or_default(); + let segments = query.split('&').filter(|_| !query.is_empty()); + for pair in segments { + let is_jwt = url::form_urlencoded::parse(pair.as_bytes()) + .next() + .is_some_and(|(key, _)| key == "jwt"); + if !is_jwt { + pairs.push(pair); + } else if !replaced { + pairs.push(&jwt); + replaced = true; + } + } + if !replaced { + pairs.push(&jwt); + } + pairs.join("&") +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::FutureExt; + use std::time::{Duration, SystemTime}; + + fn grant(prefix: &str, seconds: u64) -> Grant { + let patterns = [prefix.parse().unwrap()].into_iter().collect(); + let mut grant = Grant::new(patterns, Default::default()); + grant.expires = Some(SystemTime::now() + Duration::from_secs(seconds)); + grant + } + + async fn sessions(lease: super::super::Lease) -> (moq_net::Session, tokio::task::JoinHandle) { + let config = || { + let mut config = qmux::Config::new(qmux::Version::QMux01); + config.protocol = qmux::Protocol::Negotiate( + moq_net::Versions::from("moq-lite-07-wip".parse::().unwrap()) + .alpns() + .iter() + .map(|alpn| alpn.to_string()) + .collect(), + ); + config + }; + let stream = |io| qmux::transport::Stream::new(io, qmux::Version::QMux01, config().max_record_size); + let (client, server) = tokio::io::duplex(64 * 1024); + let (client, server) = tokio::try_join!( + qmux::Session::connect(stream(client), config()), + qmux::Session::accept(stream(server), config()) + ) + .unwrap(); + let now = tokio::time::Instant::now().into_std(); + let client = async { + let origin = moq_tokio::origin::spawn(); + let (session, driver) = moq_net::Client::new() + .with_versions("moq-lite-07-wip".parse::().unwrap().into()) + .with_publisher(origin.consume()) + .connect(now, moq_tokio::transport::Session::new(client)) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + }; + let server = async { + let origin = moq_tokio::origin::spawn(); + let request = moq_net::Server::new() + .with_versions("moq-lite-07-wip".parse::().unwrap().into()) + .with_subscriber(origin) + .accept_request(now, moq_tokio::transport::Session::new(server)) + .await + .unwrap(); + let mut lease = lease.preparing(request.auth(), request.role()).unwrap(); + let (session, driver) = request.ok().await.unwrap(); + tokio::spawn(moq_net::time::run(driver)); + (tokio::spawn(async move { + let reason = lease.ended().await; + session.abort(moq_net::Error::Unauthorized); + lease.close(reason.clone(), Default::default()); + reason + }),) + }; + let (client, (server,)) = tokio::join!(client, server); + (client, server) + } + + async fn admission(initial: Grant) -> (super::super::Lease, super::super::Admissions) { + let (auth, mut admissions) = super::super::Auth::embedded("test"); + let request = auth.request(moq_auth::Transport::Quic, "/room"); + let (lease, ()) = tokio::join!(auth.admit(request), async { + let admission = admissions.next().await.unwrap(); + assert!(!admission.attached(), "a new session arrived as an attachment"); + admission.grant(lease::Consumer::fixed(initial)); + }); + (lease.unwrap(), admissions) + } + + #[tokio::test(start_paused = true)] + async fn an_in_band_token_arrives_as_an_attachment() { + let (lease, mut admissions) = admission(grant("a/**", 3000)).await; + let (client, supervising) = sessions(lease).await; + let auth = client.auth(); + let (token, ()) = tokio::join!(auth.add("replacement"), async { + let admission = admissions.next().await.unwrap(); + assert!(admission.attached(), "an in-band token arrived as a new session"); + admission.grant(lease::Consumer::fixed(grant("a/**", 3000))); + }); + token.unwrap(); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + + #[tokio::test(start_paused = true)] + async fn an_outage_refuses_transiently_and_never_extends_the_expiry() { + let (lease, mut admissions) = admission(grant("a/**", 30)).await; + let (client, supervising) = sessions(lease).await; + let start = tokio::time::Instant::now(); + for _ in 0..3 { + let auth = client.auth(); + let (result, ()) = tokio::join!(auth.add("replacement"), async { + admissions + .next() + .await + .unwrap() + .refuse(super::super::Error::Unavailable("auth server down".into())); + }); + assert!(matches!(result, Err(moq_net::Error::Session(SessionError::Internal)))); + tokio::time::advance(Duration::from_secs(5)).await; + } + assert_eq!(supervising.await.unwrap(), lease::Reason::Expired); + let lived = start.elapsed(); + assert!( + lived >= Duration::from_secs(29) && lived <= Duration::from_secs(31), + "the session outlived its only accepted grant: {lived:?}" + ); + client.abort(moq_net::Error::Cancel); + } + + #[tokio::test(start_paused = true)] + async fn a_decider_past_the_admission_timeout_keeps_its_slot() { + let (lease, mut admissions) = admission(grant("a/**", 3000)).await; + let (client, supervising) = sessions(lease).await; + let mut held = Vec::new(); + for _ in 0..128 { + let auth = client.auth(); + let (result, admission) = tokio::join!(auth.add("slow"), async { admissions.next().await.unwrap() }); + // Refused at the timeout, while the decider still holds the admission. + assert!(matches!(result, Err(moq_net::Error::Session(SessionError::Internal)))); + held.push(admission); + } + let auth = client.auth(); + tokio::select! { + biased; + _ = admissions.next() => panic!("a timed-out decision released its slot"), + result = auth.add("overflow") => assert!(matches!( + result, + Err(moq_net::Error::Session(SessionError::TooManyRequests)) + )), + } + // The decider giving up frees the slot. + drop(held.pop()); + let auth = client.auth(); + let (_, admission) = tokio::join!(auth.add("next"), async { admissions.next().await.unwrap() }); + assert_eq!(admission.request.query.as_deref(), Some("jwt=next")); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + + #[tokio::test(start_paused = true)] + async fn renewal_outlives_the_initial_credential_and_withdrawal_ends_only_its_grant() { + let (lease, mut admissions) = admission(grant("a/**", 3)).await; + let (client, mut supervising) = sessions(lease).await; + let auth = client.auth(); + let (token, ()) = tokio::join!(auth.add("replacement"), async { + let admission = admissions.next().await.unwrap(); + assert_eq!(admission.request.query.as_deref(), Some("jwt=replacement")); + admission.grant(lease::Consumer::fixed(grant("a/**", 20))); + }); + let token = token.unwrap(); + tokio::time::advance(Duration::from_secs(4)).await; + tokio::task::yield_now().await; + assert!( + (&mut supervising).now_or_never().is_none(), + "renewed connection ended with its first credential" + ); + assert!(token.closed().now_or_never().is_none()); + drop(token); + assert_eq!(supervising.await.unwrap(), lease::Reason::Dropped); + } + + #[tokio::test(start_paused = true)] + async fn refused_and_different_root_tokens_leave_the_session_grant_unchanged() { + let (lease, mut admissions) = admission(grant("a/**", 30)).await; + let (client, supervising) = sessions(lease).await; + let auth = client.auth(); + for refusal in [0, 1, 2] { + let (result, ()) = tokio::join!(auth.add("bad"), async { + let admission = admissions.next().await.unwrap(); + if refusal == 1 { + let mut other = grant("b/**", 30); + other.root = Some("other".into()); + admission.grant(lease::Consumer::fixed(other)); + } else if refusal == 2 { + let mut other = grant("b/**", 30); + other.subscribe = other.publish.clone(); + other.publish = Default::default(); + admission.grant(lease::Consumer::fixed(other)); + } else { + admission.refuse(super::super::Error::Refused); + } + }); + assert!(result.is_err()); + } + assert!(!supervising.is_finished()); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + #[tokio::test(start_paused = true)] + async fn attached_grants_update_union_and_withdraw_without_widening_setup_role() { + let (lease, mut admissions) = admission(grant("a/**", 30)).await; + let (client, supervising) = sessions(lease).await; + let auth = client.auth(); + let mut added = grant("b/**", 30); + added.subscribe = ["**".parse().unwrap()].into_iter().collect(); + let (producer, consumer) = lease::Producer::new(added); + let (token, ()) = tokio::join!(auth.add("replacement"), async { + admissions.next().await.unwrap().grant(consumer); + }); + let token = token.unwrap(); + assert!( + token.grant().peek().unwrap().subscribe.is_empty(), + "publisher SETUP role widened" + ); + let mut union = auth.grant(); + let current = union.peek().unwrap(); + assert!(current.publish.matches("a/cam")); + assert!(current.publish.matches("b/cam")); + producer.update(grant("c/**", 30)); + let updated = union.changed().await.unwrap().unwrap(); + assert!(updated.publish.matches("c/cam")); + assert!(!updated.publish.matches("b/cam")); + drop(token); + let narrowed = union.changed().await.unwrap().unwrap(); + assert!(narrowed.publish.matches("a/cam")); + assert!(!narrowed.publish.matches("c/cam")); + assert!(!supervising.is_finished()); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + + #[tokio::test(start_paused = true)] + async fn an_empty_token_is_refused_once_the_connection_credential_changes_root() { + let (auth, mut admissions) = super::super::Auth::embedded("test"); + let request = auth.request(moq_auth::Transport::Quic, "/room"); + let (owner, consumer) = lease::Producer::new(grant("a/**", 3000)); + let (lease, ()) = tokio::join!(auth.admit(request), async { + admissions.next().await.unwrap().grant(consumer); + }); + let (client, supervising) = sessions(lease.unwrap()).await; + let auth = client.auth(); + let (token, ()) = tokio::join!(auth.add("replacement"), async { + admissions + .next() + .await + .unwrap() + .grant(lease::Consumer::fixed(grant("a/**", 3000))); + }); + let _token = token.unwrap(); + let mut moved = grant("a/**", 3000); + moved.root = Some("other".into()); + owner.update(moved); + tokio::task::yield_now().await; + let result = auth.add("").await; + assert!( + matches!(result, Err(moq_net::Error::Session(SessionError::Unauthorized))), + "an empty token reissued a credential the relay removed: {:?}", + result.map(|token| token.grant().peek()) + ); + assert!(!supervising.is_finished()); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + + #[tokio::test(start_paused = true)] + async fn withdrawn_tokens_keep_their_pending_verification_slots() { + let (lease, mut admissions) = admission(grant("a/**", 3000)).await; + let (client, supervising) = sessions(lease).await; + let mut held = Vec::new(); + for _ in 0..128 { + let auth = client.auth(); + let adding = tokio::spawn(async move { auth.add("withdrawn").await }); + held.push(admissions.next().await.unwrap()); + adding.abort(); + let _ = adding.await; + } + let auth = client.auth(); + tokio::select! { + biased; + _ = admissions.next() => panic!("withdrawal released an in-flight verification slot"), + result = auth.add("overflow") => assert!(matches!( + result, + Err(moq_net::Error::Session(SessionError::TooManyRequests)) + )), + } + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + + #[test] + fn renewal_replaces_only_the_jwt_pair() { + assert_eq!( + with_jwt(Some("flag&a=%2f&jwt=old&b=x+y&j%77t=dup"), "new token"), + "flag&a=%2f&jwt=new+token&b=x+y" + ); + assert_eq!(with_jwt(Some("flag&&jwt=old&"), "t"), "flag&&jwt=t&"); + assert_eq!(with_jwt(Some("a=1"), "t"), "a=1&jwt=t"); + assert_eq!(with_jwt(Some(""), "t"), "jwt=t"); + assert_eq!(with_jwt(None, "t"), "jwt=t"); + } + + #[tokio::test(start_paused = true)] + async fn public_rule_sessions_refuse_tokens_as_unsupported() { + let auth = super::super::Config { + public_publish: vec!["**".parse().unwrap()], + ..Default::default() + } + .init("test", &moq_tokio::tls::Connect::default(), false) + .unwrap(); + let lease = auth + .admit(auth.request(moq_auth::Transport::Quic, "/room")) + .await + .unwrap(); + let (client, supervising) = sessions(lease).await; + assert!(matches!( + client.auth().add("token").await, + Err(moq_net::Error::Unsupported) + )); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } + + #[tokio::test(start_paused = true)] + async fn certificate_sessions_refuse_tokens_as_unsupported_before_admission() { + let (auth, mut admissions) = super::super::Auth::embedded("test"); + let mut request = auth.request(moq_auth::Transport::Quic, "/room"); + request.tls = Some(moq_auth::Peer { + name: "edge".into(), + fingerprint: "ab".repeat(32), + expires: None, + issuer: "test".into(), + }); + let (lease, ()) = tokio::join!(auth.admit(request), async { + admissions + .next() + .await + .unwrap() + .grant(lease::Consumer::fixed(grant("a/**", 30))); + }); + let (client, supervising) = sessions(lease.unwrap()).await; + assert!(matches!( + client.auth().add("replacement").await, + Err(moq_net::Error::Unsupported) + )); + assert!(admissions.next().now_or_never().is_none()); + client.abort(moq_net::Error::Cancel); + supervising.abort(); + let _ = supervising.await; + } +} diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index be53bf84d2..45193a9739 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -1059,6 +1059,41 @@ impl Cluster { self.scope(lease, &request) } + /// Network sessions enforce changing grants in AUTH, within a fixed root and SETUP role. + pub(crate) async fn admit_session( + &self, + auth: &auth::Auth, + request: moq_auth::Request, + ) -> Result { + let lease = auth.admit(request.clone()).await?; + self.scope_session(lease, &request) + } + + pub(crate) fn scope_session( + &self, + lease: auth::Lease, + request: &moq_auth::Request, + ) -> Result { + let mut admitted = self.scope(lease, request)?; + let mut bound = admitted.lease.token().clone(); + bound.publish = moq_net::Pattern::all().into(); + bound.subscribe = moq_net::Pattern::all().into(); + if request.role != Some(moq_auth::Role::Subscriber) { + admitted.publisher = self + .publisher(&bound) + .map(|origin| origin.with_stats(admitted.stats.clone())); + } + if request.role != Some(moq_auth::Role::Publisher) { + admitted.subscriber = self.subscriber(&bound).map(|origin| { + origin + .consume() + .with_hidden(request.tls.is_some() || Self::is_lan_path(&request.path)) + .with_stats(admitted.stats.clone()) + }); + } + Ok(admitted) + } + /// Resolve origin handles for a lease already admitted by a local relay rule. pub(crate) fn scope(&self, lease: auth::Lease, request: &moq_auth::Request) -> Result { let token = lease.token(); diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index 6019fefa1e..a65666eaa5 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -120,6 +120,7 @@ impl Connection { if let Some(publisher) = publisher { request = request.with_subscriber(publisher); } + let lease = lease.preparing(request.auth(), request.role())?; // Bounded too: a peer that stalls the SETUP reply must not outlive the relay. let session = self .shutdown @@ -151,7 +152,7 @@ impl Connection { if self.request.peer_identity().is_some() { tracing::debug!("client certificate verified; reported to the auth server"); } - let admitted = self.cluster.admit(&self.auth, request.clone()).await?; + let admitted = self.cluster.admit_session(&self.auth, request.clone()).await?; let registration = self.sessions.as_ref().map(|sessions| sessions.register(request)); Ok((admitted, registration)) } diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index b36ff97f75..70bd201219 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -721,7 +721,9 @@ async fn serve_connection( .tokio .spawn(async move { // Abandoned at the drain deadline: worker teardown only detaches this task. - let admitted = shutdown.admit(cluster.admit(&auth, auth_request.clone())).await?; + let admitted = shutdown + .admit(cluster.admit_session(&auth, auth_request.clone())) + .await?; Ok::<_, crate::auth::Error>((admitted, sessions.register(auth_request))) }) .await @@ -763,6 +765,7 @@ async fn serve_connection( if let Some(publisher) = publisher { request = request.with_subscriber(publisher); } + let lease = lease.preparing(request.auth(), request.role())?; let (session, driver) = request.ok().await?; let driver_handle = handle.clone(); handle.spawn(async move { diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index fe9ceb93ec..654830d1e6 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -15,7 +15,7 @@ use axum::{ use moq_net::origin; use moq_net::stats::Session; -use crate::{auth, refusals::Refusal, web::MtlsPeer, web::WebState, web::landing_response}; +use crate::{auth, web::MtlsPeer, web::WebState, web::landing_response}; // One axum extractor per fact the upgrade needs; there is no struct to fold them into. #[allow(clippy::too_many_arguments)] @@ -69,24 +69,15 @@ pub(crate) async fn serve_ws( .await .inspect_err(|err| state.cluster.refusals.record(err.into()))?; let limits = state.cluster.limits(&lease); - let token = lease.token(); - let publish = state.cluster.publisher(token); - // A verified client certificate marks a cluster peer, which discovers hidden - // routes; see `Cluster::scope`. - let subscribe = state + let crate::cluster::Admitted { + lease, + publisher: publish, + subscriber: subscribe, + stats, + } = state .cluster - .subscriber(token) - .map(|subscribe| subscribe.consume().with_hidden(request.tls.is_some())); - - if publish.is_none() && subscribe.is_none() { - // Bad token, we can't publish or subscribe. - state.cluster.refusals.record(Refusal::Forbidden); - return Err(StatusCode::UNAUTHORIZED.into()); - } - - // Only an admitted session counts as present, as on the native path. - let stats = state.cluster.stats.tier(token.tier.clone()).session(&token.root); - let lease = lease.with_stats(stats.clone()); + .scope_session(lease, &request) + .inspect_err(|err| state.cluster.refusals.record(err.into()))?; Ok(ws.on_upgrade(async move |socket| { let _active = active; @@ -215,24 +206,33 @@ where // Keep the driver in this task so cancellation tears down the transport. let ws = moq_tokio::transport::Session::new(ws); let mut refused = ws.clone(); - let accept = server.accept(tokio::time::Instant::now().into_std(), ws); - let accept = async move { + let handshake = async move { + let request = server + .accept_request(tokio::time::Instant::now().into_std(), ws) + .await?; + let lease = lease.preparing(request.auth(), request.role())?; + let (session, driver) = request.ok().await?; + Ok::<_, moq_net::Error>((lease, session, driver)) + }; + // The handshake timeout covers the SETUP reply too, so a peer that sends SETUP + // and then withholds flow-control credit is dropped like one that never sends it. + let handshake = async move { match timeout { - None => Ok(accept.await), - Some(timeout) => tokio::time::timeout(timeout, accept) + None => handshake.await, + Some(timeout) => tokio::time::timeout(timeout, handshake) .await - .map_err(|_| moq_net::Error::Timeout), + .unwrap_or(Err(moq_net::Error::Timeout)), } }; - // The drain deadline bounds SETUP too, so a peer that never sends it cannot - // outlive the relay. - let accepted = shutdown - .until_deadline(accept) + // The drain deadline bounds the handshake too, so a stalled peer cannot outlive + // the relay. + let handshake = shutdown + .until_deadline(handshake) .await .unwrap_or(Err(moq_net::Error::GoawayTimeout)); - let (session, driver) = match accepted { - Ok(accepted) => accepted?, - // A peer that upgraded and never sent SETUP in time; tell it why it is dropped. + let (lease, session, driver) = match handshake { + Ok(handshake) => handshake, + // Tell the peer why it is dropped. Err(err) => { use web_transport_trait::poll::Session as _; refused.close(moq_net::SessionError::from(&err).to_code(), &err.to_string()); diff --git a/rs/moq-relay/tests/auth_lifetime.rs b/rs/moq-relay/tests/auth_lifetime.rs index 8d72f17012..c98836b49c 100644 --- a/rs/moq-relay/tests/auth_lifetime.rs +++ b/rs/moq-relay/tests/auth_lifetime.rs @@ -1381,6 +1381,116 @@ async fn quic_and_webtransport_report_distinct_auth_transports() { let _ = relay.await; } +/// A token can reach a prefix outside the initial credential, and withdrawing it +/// revokes only the streams that depended on it on both native and upgraded WS. +#[tokio::test] +async fn in_band_tokens_widen_and_withdraw_live_prefixes() { + for scheme in ["tcp", "ws"] { + let mut initial = Grant::new(all(), ["a/**".parse().unwrap()].into_iter().collect()); + initial.expires = Some(SystemTime::now() + Duration::from_secs(3600)); + let script = Script::new(initial); + let auth = build_auth(script.spawn().await); + let (port, relay) = if scheme == "tcp" { + spawn_relay(auth).await + } else { + spawn_ws_relay(auth).await + }; + let client = || { + let mut config = moq_tokio::connect::Config::default(); + config.once = Some(true); + config.version = vec!["moq-transport-22".parse().unwrap()]; + config.websocket.delay = Duration::ZERO; + config.init(Default::default()).unwrap() + }; + let origin = moq_tokio::origin::spawn(); + let a = origin.create_broadcast("a/cam").unwrap(); + let a_track = a.create_track("video", timed()).unwrap(); + a.announce(Default::default()).unwrap(); + let b = origin.create_broadcast("b/cam").unwrap(); + let b_track = b.create_track("video", timed()).unwrap(); + b.announce(Default::default()).unwrap(); + let publisher = tokio::time::timeout( + TIMEOUT, + client() + .with_publisher(origin.consume()) + .connect(room_url(scheme, port)) + .established(), + ) + .await + .unwrap() + .unwrap(); + let received = moq_tokio::origin::spawn(); + let read = received.consume(); + let subscriber = tokio::time::timeout( + TIMEOUT, + client() + .with_subscriber(received) + .connect(room_url(scheme, port)) + .established(), + ) + .await + .unwrap() + .unwrap(); + let a_read = tokio::time::timeout(TIMEOUT, read.routed_broadcast("a/cam")) + .await + .unwrap() + .unwrap(); + let mut a_sub = a_read.track("video").unwrap().subscribe(None).await.unwrap(); + let mut replacement = Grant::new(all(), ["b/**".parse().unwrap()].into_iter().collect()); + replacement.expires = Some(SystemTime::now() + Duration::from_secs(3600)); + script.on_connect(Answer::Grant(replacement)); + let token = tokio::time::timeout(TIMEOUT, subscriber.auth().add("replacement")) + .await + .unwrap() + .unwrap(); + let b_read = tokio::time::timeout(TIMEOUT, read.routed_broadcast("b/cam")) + .await + .expect("new prefix not announced") + .unwrap(); + let mut b_sub = b_read.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = b_track.append_group().unwrap(); + group + .write_frame(moq_net::Timestamp::ZERO, b"new prefix".as_ref()) + .unwrap(); + group.finish().unwrap(); + assert!( + tokio::time::timeout(TIMEOUT, b_sub.recv_group()) + .await + .unwrap() + .unwrap() + .is_some() + ); + drop(token); + let ended = tokio::time::timeout(TIMEOUT, b_sub.recv_group()) + .await + .expect("withdrawn prefix stayed authorized"); + assert!( + matches!( + ended, + Err(moq_net::Error::Unauthorized | moq_net::Error::Stream(moq_net::StreamError::Unauthorized)) + ), + "{scheme}: {:?}", + ended.as_ref().err() + ); + let mut group = a_track.append_group().unwrap(); + group + .write_frame(moq_net::Timestamp::ZERO, b"original prefix".as_ref()) + .unwrap(); + group.finish().unwrap(); + assert!( + tokio::time::timeout(TIMEOUT, a_sub.recv_group()) + .await + .unwrap() + .unwrap() + .is_some() + ); + subscriber.abort(moq_net::Error::Cancel); + publisher.abort(moq_net::Error::Cancel); + relay.abort(); + let _ = relay.await; + } +} + /// Both admission paths apply the selected cap before starting the session driver. #[tokio::test] async fn relay_request_limits_distinguish_clients_and_peers() { diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs index 68cef08755..63f6c043f7 100644 --- a/rs/moq-tokio/src/connection.rs +++ b/rs/moq-tokio/src/connection.rs @@ -1,5 +1,8 @@ //! The reconnect loop around a session, and the [`Monitor`] that observes it. +pub mod auth; +pub use auth::Auth; + use std::task::{Poll, ready}; use std::time::Duration; @@ -385,6 +388,7 @@ const MIN_BACKOFF: Duration = Duration::from_millis(50); /// grows it, and `max` caps it, so a zero anywhere collapses the whole schedule. /// Resolving them once, here, keeps that from having to be re-argued at each of /// the three places the loop reaches for them. +#[derive(Clone, Copy)] struct Pacing { /// The first delay, and the bar a session must clear to count as healthy. At /// zero every session looks healthy, which resets the give-up window forever. @@ -676,6 +680,8 @@ pub struct Snapshot { pub struct Connection { task: std::sync::Arc, state: kio::Consumer, + /// The reconnect pacing, which retained tokens reuse to retry a transient refusal. + pacing: Pacing, /// Persistent send-bitrate estimate, fed by the loop from each live session. send_bandwidth: BandwidthConsumer, /// Persistent recv-bitrate estimate, fed by the loop from each live session. @@ -706,6 +712,7 @@ impl Connection { pub(crate) fn new(client: Client, addrs: Addrs) -> Self { let producer = kio::Producer::::default(); let state = producer.consume(); + let pacing = Pacing::new(&client.backoff); // The loop feeds these across every reconnect, so a consumer's handle survives session churn // (unlike a session's own bandwidth consumer, which dies with the session). @@ -742,6 +749,7 @@ impl Connection { closed, }), state, + pacing, send_bandwidth, recv_bandwidth, last_reported: None, @@ -1272,6 +1280,14 @@ impl Connection { kio::wait(|waiter| self.poll_closed(waiter)).await } + /// Present tokens that survive session replacement and observe the live grant union. + pub fn auth(&self) -> Auth { + Auth { + connection: self.state.clone(), + pacing: self.pacing, + } + } + /// Observe the connection's statistics and presence across reconnects. /// /// The handle keeps working across reconnects without keeping the connection loop alive. diff --git a/rs/moq-tokio/src/connection/auth.rs b/rs/moq-tokio/src/connection/auth.rs new file mode 100644 index 0000000000..b447952da6 --- /dev/null +++ b/rs/moq-tokio/src/connection/auth.rs @@ -0,0 +1,481 @@ +//! Tokens and grant watches that follow a reconnecting connection. + +use std::task::{Poll, ready}; + +use bytes::Bytes; +use futures::FutureExt; +use moq_net::{auth, kio}; + +use super::{Pacing, State, terminal}; +use crate::{Error, abort::AbortOnDrop}; + +/// Present session tokens across every session of a connection. +#[derive(Clone)] +pub struct Auth { + pub(super) connection: kio::Consumer, + pub(super) pacing: Pacing, +} + +impl Auth { + /// Watch the current session's union, clearing it while disconnected. + pub fn grant(&self) -> Watch { + Watch { + source: Source::Connection(self.connection.clone()), + last: None, + } + } + + /// Present a token now and on later sessions until its handle is dropped or refused. + pub async fn add(&self, token: impl Into) -> crate::Result { + let producer = kio::Producer::::default(); + let state = producer.consume(); + let connection = self.connection.clone(); + let pacing = self.pacing; + let token = token.into(); + let (answered, answer) = tokio::sync::oneshot::channel(); + let task = AbortOnDrop::new(tokio::spawn(async move { + let mut answered = Some(answered); + if let Err(err) = present(connection, pacing, token, &producer, &mut answered).await { + if let Some(answered) = answered.take() { + let _ = answered.send(Err(err.clone())); + } + if let Ok(mut state) = producer.write() { + state.grant = None; + state.error = Some(err); + } + } + })); + answer.await.map_err(|_| Error::from(moq_net::Error::Cancel))??; + Ok(Token { _task: task, state }) + } +} + +/// A presented token; dropping it withdraws it from live and draining sessions. +pub struct Token { + _task: AbortOnDrop, + state: kio::Consumer, +} + +impl Token { + /// Watch this token's grant on the current session. + pub fn grant(&self) -> Watch { + Watch { + source: Source::Token(self.state.clone()), + last: None, + } + } + + /// Wait until the token is refused or its connection ends. + pub async fn closed(&self) -> crate::Result<()> { + kio::wait(|waiter| self.state.poll_closed(waiter)).await; + match &self.state.read().error { + Some(err) => Err(err.clone()), + None => Ok(()), + } + } +} + +#[derive(Default)] +struct Granted { + grant: Option, + error: Option, +} + +#[derive(Clone)] +enum Source { + Connection(kio::Consumer), + Token(kio::Consumer), +} + +/// Observe a grant as tokens change and the connection replaces its session. +#[derive(Clone)] +pub struct Watch { + source: Source, + last: Option, +} + +impl Watch { + /// The current grant, or none while disconnected or after the token ends. + pub fn peek(&self) -> Option { + match &self.source { + Source::Connection(connection) => { + if connection.poll_closed(&kio::Waiter::noop()).is_ready() { + return None; + } + connection + .read() + .session + .as_ref() + .and_then(|session| session.auth().grant().peek()) + } + Source::Token(state) => { + if state.poll_closed(&kio::Waiter::noop()).is_ready() { + return None; + } + state.read().grant.clone() + } + } + } + + /// Poll for a different grant or the end of its owner. + pub fn poll_changed(&mut self, waiter: &kio::Waiter) -> Poll>> { + let result = match &self.source { + Source::Connection(connection) => match ready!(connection.poll(waiter, |state| { + let mut watch = state.session.as_ref().map(|session| session.auth().grant()); + let current = watch.as_ref().and_then(auth::Watch::peek); + if current != self.last { + return Poll::Ready(current); + } + if let Some(watch) = &mut watch + && let Poll::Ready(Ok(grant)) = watch.poll_changed(waiter) + { + return Poll::Ready(grant); + } + Poll::Pending + })) { + Ok(grant) => Ok(grant), + Err(state) => Err(terminal(&state)), + }, + Source::Token(state) => match ready!(state.poll(waiter, |state| { + if state.grant != self.last { + Poll::Ready(state.grant.clone()) + } else { + Poll::Pending + } + })) { + Ok(grant) => Ok(grant), + Err(state) => Err(state.error.clone().unwrap_or_else(|| moq_net::Error::Cancel.into())), + }, + }; + if let Ok(grant) = &result { + self.last = grant.clone(); + } + Poll::Ready(result) + } + + /// Wait for a different grant or the end of its owner. + pub async fn changed(&mut self) -> crate::Result> { + kio::wait(|waiter| self.poll_changed(waiter)).await + } +} + +async fn next( + connection: &kio::Consumer, + previous: Option<(u64, bool)>, +) -> crate::Result<(u64, Option)> { + kio::wait(|waiter| { + match ready!(connection.poll(waiter, |state| { + if Some((state.connects, state.session.is_some())) != previous { + Poll::Ready((state.connects, state.session.clone())) + } else { + Poll::Pending + } + })) { + Ok(session) => Poll::Ready(Ok(session)), + Err(state) => Poll::Ready(Err(terminal(&state))), + } + }) + .await +} + +/// Whether a refusal is the acceptor failing to decide rather than its verdict on +/// the token, such as an unreachable auth server or a full verification queue. +/// moq-transport's AUTH_ERROR carries neither, so there every refusal is final. +fn transient(err: &moq_net::Error) -> bool { + matches!( + err, + moq_net::Error::Session( + moq_net::SessionError::Internal | moq_net::SessionError::TooManyRequests | moq_net::SessionError::Timeout + ) + ) +} + +async fn present( + connection: kio::Consumer, + pacing: Pacing, + bytes: Bytes, + producer: &kio::Producer, + answered: &mut Option>>, +) -> crate::Result<()> { + let mut snapshot = next(&connection, None).await?; + let mut draining = Vec::new(); + 'session: loop { + let (connects, session) = snapshot; + let Some(session) = session else { + if let Ok(mut state) = producer.write() { + state.grant = None; + } + snapshot = next(&connection, Some((connects, false))).await?; + continue; + }; + if let Ok(mut state) = producer.write() { + state.grant = None; + } + let handle = session.auth(); + let mut delay = None; + let token = loop { + let attempt = async { + if let Some(delay) = delay { + tokio::time::sleep(delay).await; + } + handle.add(bytes.clone()).await + }; + let result = tokio::select! { + biased; + changed = next(&connection, Some((connects, true))) => { snapshot = changed?; continue 'session; } + result = attempt => result, + }; + match result { + Ok(token) => break token, + // The session ending is no verdict on the token: replay it on the next one. + Err(_) if session.closed().now_or_never().is_some() => { + snapshot = next(&connection, Some((connects, true))).await?; + continue 'session; + } + // Nor is an acceptor that could not decide: present it again, paced like a + // reconnect. Only an accepted grant counts, so a retry never extends one. + Err(err) if transient(&err) => { + let wait = delay.map_or(pacing.initial, |delay| pacing.next(delay)); + tracing::debug!(%err, ?wait, "token refused transiently; presenting it again"); + delay = Some(wait); + } + Err(err) => return Err(err.into()), + } + }; + if let Some(answered) = answered.take() { + let _ = answered.send(Ok(())); + } + let mut watch = token.grant(); + if let Ok(mut state) = producer.write() { + state.grant = watch.peek(); + } + loop { + tokio::select! { + biased; + changed = next(&connection, Some((connects, true))) => { snapshot = changed?; break; } + _ = session.closed() => { + if let Ok(mut state) = producer.write() { state.grant = None; } + snapshot = next(&connection, Some((connects, true))).await?; + break; + } + grant = watch.changed() => match grant { + Ok(grant) => { + if let Ok(mut state) = producer.write() { state.grant = grant; } + } + // The stream ended with its session, not by the peer's verdict. + Err(_) if session.closed().now_or_never().is_some() => { + if let Ok(mut state) = producer.write() { state.grant = None; } + snapshot = next(&connection, Some((connects, true))).await?; + break; + } + Err(err) => return Err(err.into()), + } + } + } + // A migrated session may still be serving its streams during GOAWAY drain. + // Keep its grant until that session ends or the application drops this token. + draining.retain(|(session, _): &(moq_net::Session, auth::Token)| session.closed().now_or_never().is_none()); + draining.push((session, token)); + } +} + +#[cfg(all(test, feature = "tcp"))] +mod tests { + use super::*; + + struct Pair { + client: moq_net::Session, + requests: tokio::sync::mpsc::UnboundedReceiver, + _tasks: Vec>, + _acceptor: AbortOnDrop, + } + + async fn pair() -> Pair { + let versions = moq_net::Versions::from("moq-lite-07-wip".parse::().unwrap()); + let config = || { + let mut config = qmux::Config::new(qmux::Version::QMux01); + config.protocol = qmux::Protocol::Negotiate(versions.alpns().iter().map(|alpn| alpn.to_string()).collect()); + config + }; + let stream = |io| qmux::transport::Stream::new(io, qmux::Version::QMux01, config().max_record_size); + let (client, server) = tokio::io::duplex(64 * 1024); + let (client, server) = tokio::try_join!( + qmux::Session::connect(stream(client), config()), + qmux::Session::accept(stream(server), config()) + ) + .unwrap(); + let now = tokio::time::Instant::now().into_std(); + let client = async { + let (session, driver) = moq_net::Client::new() + .with_versions(versions.clone()) + .connect(now, crate::transport::Session::new(client)) + .await + .unwrap(); + (session, AbortOnDrop::new(tokio::spawn(moq_net::time::run(driver)))) + }; + let server = async { + let request = moq_net::Server::new() + .with_versions(versions.clone()) + .accept_request(now, crate::transport::Session::new(server)) + .await + .unwrap(); + let mut requests = request.auth().requests().unwrap(); + let (session, driver) = request.ok().await.unwrap(); + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + let acceptor = tokio::spawn(async move { + let _session = session; + let mut initial = Vec::new(); + while let Some(request) = requests.next().await { + if request.token().is_empty() { + initial.push(request.accept(auth::Grant::default())); + } else if sender.send(request).is_err() { + break; + } + } + }); + ( + receiver, + AbortOnDrop::new(tokio::spawn(moq_net::time::run(driver))), + AbortOnDrop::new(acceptor), + ) + }; + let ((client, client_task), (requests, server_task, acceptor)) = tokio::join!(client, server); + Pair { + client, + requests, + _tasks: vec![client_task, server_task], + _acceptor: acceptor, + } + } + + fn attach(connection: &kio::Producer, session: &moq_net::Session) { + let mut state = connection.write().ok().expect("connection observed"); + state.connects += 1; + state.session = Some(session.clone()); + } + + #[tokio::test(start_paused = true)] + async fn a_token_survives_replacement_and_withdraws_from_both_sessions() { + let connection = kio::Producer::::default(); + let auth = retained(&connection); + let mut first = pair().await; + attach(&connection, &first.client); + let (token, first_grant) = tokio::join!(auth.add("renew"), async { + let request = first.requests.recv().await.unwrap(); + assert_eq!(request.token().as_ref(), b"renew"); + request.accept(auth::Grant::all()) + }); + let token = token.unwrap(); + let mut grant = token.grant(); + assert!(grant.changed().await.unwrap().is_some()); + let mut second = pair().await; + attach(&connection, &second.client); + let request = second.requests.recv().await.unwrap(); + assert_eq!(request.token().as_ref(), b"renew"); + assert!( + grant.changed().await.unwrap().is_none(), + "replacement reused its predecessor grant" + ); + let second_grant = request.accept(auth::Grant::all()); + assert!(grant.changed().await.unwrap().is_some()); + assert!( + first_grant.closed().now_or_never().is_none(), + "migration withdrew a draining grant" + ); + drop(token); + let _ = tokio::join!(first_grant.closed(), second_grant.closed()); + } + + #[tokio::test(start_paused = true)] + async fn a_session_closing_mid_replay_keeps_the_token() { + let connection = kio::Producer::::default(); + let auth = retained(&connection); + let mut first = pair().await; + attach(&connection, &first.client); + let (token, _first_grant) = tokio::join!(auth.add("renew"), async { + first.requests.recv().await.unwrap().accept(auth::Grant::all()) + }); + let token = token.unwrap(); + let mut second = pair().await; + attach(&connection, &second.client); + // Close the replacement while its replayed token is still unanswered, before + // the connection notices. + let _unanswered = second.requests.recv().await.unwrap(); + second.client.abort(moq_net::Error::Cancel); + second.client.closed().await; + let mut third = pair().await; + attach(&connection, &third.client); + let request = tokio::time::timeout(std::time::Duration::from_secs(5), third.requests.recv()) + .await + .expect("the token was not replayed after its session closed") + .unwrap(); + assert_eq!(request.token().as_ref(), b"renew"); + drop(token); + } + + fn retained(connection: &kio::Producer) -> Auth { + Auth { + connection: connection.consume(), + pacing: Pacing::new(&super::super::Backoff::default()), + } + } + + #[tokio::test(start_paused = true)] + async fn a_transient_refusal_is_presented_again_after_backoff() { + let connection = kio::Producer::::default(); + let auth = retained(&connection); + let mut pair = pair().await; + attach(&connection, &pair.client); + let start = tokio::time::Instant::now(); + let (token, issued) = tokio::join!(auth.add("renew"), async { + for _ in 0..2 { + let outage = pair.requests.recv().await.unwrap(); + outage.reject(moq_net::SessionError::Internal, "auth server unavailable"); + } + let retried = pair.requests.recv().await.unwrap(); + assert_eq!(retried.token().as_ref(), b"renew"); + retried.accept(auth::Grant::all()) + }); + let token = token.unwrap(); + // The default pacing waits 1s, then 2s, before the third presentation. + assert!( + start.elapsed() >= std::time::Duration::from_secs(3), + "retried without backing off: {:?}", + start.elapsed() + ); + assert!(token.grant().changed().await.unwrap().is_some()); + drop(token); + issued.closed().await; + } + + #[tokio::test(start_paused = true)] + async fn an_unauthorized_refusal_is_final() { + let connection = kio::Producer::::default(); + let auth = retained(&connection); + let mut pair = pair().await; + attach(&connection, &pair.client); + let (token, ()) = tokio::join!(auth.add("renew"), async { + let refused = pair.requests.recv().await.unwrap(); + refused.reject(moq_net::SessionError::Unauthorized, "token refused"); + }); + assert!(token.is_err()); + tokio::time::sleep(std::time::Duration::from_secs(60)).await; + assert!(pair.requests.try_recv().is_err(), "a refused token was presented again"); + } + + #[tokio::test(start_paused = true)] + async fn ending_the_connection_withdraws_a_retained_token_handle() { + let connection = kio::Producer::::default(); + let auth = retained(&connection); + let mut pair = pair().await; + attach(&connection, &pair.client); + let (token, issued) = tokio::join!(auth.add("renew"), async { + pair.requests.recv().await.unwrap().accept(auth::Grant::all()) + }); + let token = token.unwrap(); + drop(connection); + assert!(token.closed().await.is_err()); + assert!(token.grant().peek().is_none()); + issued.closed().await; + } +} diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs index 9bf366fd5a..67f27e5db5 100644 --- a/rs/moq-tokio/src/lib.rs +++ b/rs/moq-tokio/src/lib.rs @@ -70,7 +70,7 @@ pub mod websocket; // new `pub` item in these modules doesn't silently join it. pub use client::Client; pub use connect::{Addrs, ConnectError}; -pub use connection::{Backoff, Connection, Redirect, Status}; +pub use connection::{Auth, Backoff, Connection, Redirect, Status}; pub use error::{Error, Result}; pub use log::{Log, RedactedUrl}; #[cfg(feature = "_transport")] diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 3d42bb5b50..1d1bbe5607 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1733,6 +1733,11 @@ impl Request { request_ref!(self, r => r.role()) } + /// Answer the peer's tokens before accepting and starting its protocol driver. + pub fn auth(&self) -> moq_net::auth::Handle { + request_ref!(self, r => r.auth()) + } + /// The Hop ID the peer declared in its SETUP (moq-lite-05+). /// /// A peer declares this when it attaches a publish or subscribe origin.