From 0c9306c3fcd40a16181d2991521ae73c70296409 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 07:20:53 -0700 Subject: [PATCH 1/2] fix(net): gate incoming requests at dispatch Dispatch builds each incoming request's gate and hands it to the handler, in Rust and JS. Locally issued JS grants are enforced, including renewal, revocation, and expiry. Joining FETCH ends with its parent subscription. Co-Authored-By: Claude Opus 5.5 --- js/net/src/auth_gate.test.ts | 33 ++++ js/net/src/auth_gate.ts | 30 ++++ js/net/src/ietf/publisher.ts | 35 ++-- js/net/src/lite/publisher.ts | 29 +--- quest/m1/auth/README.md | 2 - quest/m1/auth/request-gate.md | 54 ------ quest/m3/auth-token-fetch.md | 6 +- rs/moq-net/src/auth.rs | 47 ++++- rs/moq-net/src/ietf/publisher.rs | 275 +++++++++++++++++++++--------- rs/moq-net/src/ietf/subscriber.rs | 56 +++++- rs/moq-net/src/lite/publisher.rs | 17 +- 11 files changed, 384 insertions(+), 200 deletions(-) create mode 100644 js/net/src/auth_gate.test.ts create mode 100644 js/net/src/auth_gate.ts delete mode 100644 quest/m1/auth/request-gate.md diff --git a/js/net/src/auth_gate.test.ts b/js/net/src/auth_gate.test.ts new file mode 100644 index 00000000000..440ba17be39 --- /dev/null +++ b/js/net/src/auth_gate.test.ts @@ -0,0 +1,33 @@ +import { expect, test } from "bun:test"; +import { Signal } from "@moq/signals"; +import type { Grant } from "./auth.ts"; +import { Gate } from "./auth_gate.ts"; +import * as Path from "./path.ts"; + +const grant = (prefix: string): Grant => ({ + publish: new Path.Patterns([Path.Pattern.subtree(prefix)]), + subscribe: new Path.Patterns([]), +}); + +test("a gate allows everything until the peer answers", () => { + const gate = new Gate(new Signal(undefined)); + expect(gate.allows(Path.from("room/alice"))).toBe(true); + expect(new Gate().allows(Path.from("room/alice"))).toBe(true); +}); + +test("a watch revokes once when the grant stops covering its path", async () => { + const peer = new Signal(grant("room")); + const gate = new Gate(peer); + let revoked = 0; + const watch = gate.watch(Path.from("room/alice"), () => revoked++); + peer.set(grant("room/alice")); + await Promise.resolve(); + expect(revoked).toBe(0); + peer.set(grant("other")); + await Promise.resolve(); + peer.set(grant("elsewhere")); + await Promise.resolve(); + expect(revoked).toBe(1); + expect(watch.revoked()).toBe(true); + watch.dispose(); +}); diff --git a/js/net/src/auth_gate.ts b/js/net/src/auth_gate.ts new file mode 100644 index 00000000000..299e5d9cf28 --- /dev/null +++ b/js/net/src/auth_gate.ts @@ -0,0 +1,30 @@ +import type { Dispose, Getter } from "@moq/signals"; +import type { Grant } from "./auth.ts"; +import type * as Path from "./path.ts"; + +/** Whether our grant still lets us serve one incoming request, built when it is dispatched. @internal */ +export class Gate { + #grant?: Getter; + + constructor(grant?: Getter) { + this.#grant = grant; + } + + /** Whether our grant lets us publish `path` right now. No grant yet allows it. */ + allows(path: Path.Valid): boolean { + const grant = this.#grant?.peek(); + return grant === undefined || grant.publish.matches(path); + } + + /** Call `onRevoke` once our grant stops covering `path`. Arm it before the request's first check. */ + watch(path: Path.Valid, onRevoke: () => void): { revoked: () => boolean; dispose: Dispose } { + let revoked = false; + const dispose = + this.#grant?.subscribe(() => { + if (revoked || this.allows(path)) return; + revoked = true; + onRevoke(); + }) ?? (() => {}); + return { revoked: () => revoked || !this.allows(path), dispose }; + } +} diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 7e4075d0328..8e8d98ecf4d 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -1,6 +1,7 @@ import { Instant, Milli, Nano, now, sleep } from "@moq/clock"; import { type Dispose, type Getter, race, Signal } from "@moq/signals"; import type { Auth, Grant } from "../auth.ts"; +import { Gate } from "../auth_gate.ts"; import { enforceGrant } from "../auth_session.ts"; import type * as broadcast from "../broadcast.ts"; import { Withdrawal } from "../connection/withdrawal.ts"; @@ -302,6 +303,8 @@ export class Publisher { // shrink withdraws what it no longer covers. Undefined until the peer answers, which // allows everything. #grant: Getter; + // Holds every incoming request to that grant from dispatch to its end. + #gate: Gate; // Resolves once the tokens this session presented at setup are answered, so nothing is // advertised before the grant it would be checked against. @@ -352,6 +355,7 @@ export class Publisher { this.#subscriptions = new Slots(limits.subscriptions, { quic, reason: "too many subscriptions" }); this.#fetches = new Slots(limits.fetches, { quic, reason: "too many fetches" }); this.#grant = grant; + this.#gate = new Gate(grant); this.#auth = auth; this.#ready = ready; this.#session = session; @@ -388,7 +392,7 @@ export class Publisher { } // Only a subscription we would serve holds a slot, as in Rust: a refused one ends at // once. A peer past the cap loses the session. - const charged = msg.forward && !msg.rangeFilters && !this.#denied(msg.trackNamespace); + const charged = msg.forward && !msg.rangeFilters && this.#gate.allows(msg.trackNamespace); if (charged) this.#subscriptions.acquire(); try { await this.#serveSubscribe(msg, stream); @@ -428,7 +432,7 @@ export class Publisher { ), reasonPhrase: unsupported, }; - } else if (this.#denied(name)) { + } else if (!this.#gate.allows(name)) { // Serve only what our grant lets us publish. Checked before resolving, so a denied // request never reaches the origin. refusal = { errorCode: toRequestCode("unauthorized", "subscribe", version), reasonPhrase: "not granted" }; @@ -726,11 +730,9 @@ export class Publisher { const revoked = new Promise<"revoked">((resolve) => { revoke = () => resolve("revoked"); }); - const disposeGrant = this.#grant.subscribe(() => { - if (this.#denied(name)) revoke(); - }); + const watch = this.#gate.watch(name, revoke); // The grant may have shrunk during setup, before this watcher existed. - if (this.#denied(name)) revoke(); + if (watch.revoked()) revoke(); let publishError: Error | undefined; let unauthorized = false; @@ -747,7 +749,7 @@ export class Publisher { } catch (err: unknown) { publishError = error(err); } finally { - disposeGrant(); + watch.dispose(); } // An update that broke the protocol, such as one whose AUTHORIZATION TOKEN names an // alias, closes the session: the connection answers a thrown violation with its code. @@ -1137,12 +1139,6 @@ export class Publisher { } } - // Whether our grant excludes publishing this broadcast. No grant yet allows it. - #denied(broadcast: Path.Valid): boolean { - const grant = this.#grant.peek(); - return grant !== undefined && !grant.publish.matches(broadcast); - } - /** * Close the session when our origin publishes a broadcast our grant does not cover; * see {@link enforceGrant}. @@ -1217,7 +1213,7 @@ export class Publisher { const carries = (covered: Path.Valid) => visible(covered) && (legacy ? this.#requiresSolicitation || (this.#hidden && hiddenBelow(Path.empty(), covered)) : true) && - !this.#denied(covered); + this.#gate.allows(covered); // Nothing is advertised before the grant it would be checked against. await this.#ready; @@ -1436,7 +1432,7 @@ export class Publisher { // nothing our grant does not cover reaches the peer at all (MoQ Auth). if ( (this.#hidden && hiddenBelow(Path.empty(), covered)) || - this.#denied(covered) || + !this.#gate.allows(covered) || candidates.length === 0 ) continue; @@ -1796,7 +1792,7 @@ export class Publisher { // Answer only for what our grant lets us publish. Checked before the track resolves, // so a denied request never reaches the origin. - if (this.#denied(name)) { + if (!this.#gate.allows(name)) { await this.#refuse(stream, msg.requestId, TRACK_STATUS_ERROR_14, denied); return; } @@ -1810,8 +1806,7 @@ export class Publisher { // A grant that narrows abandons the lookup, and resets an answer the transport has // not taken yet rather than delivering it. let answering = false; - const disposeGrant = this.#grant.subscribe(() => { - if (!this.#denied(name)) return; + const watch = this.#gate.watch(name, () => { hold.abort(); if (answering) stream.abort(unauthorized(name)); }); @@ -1842,7 +1837,7 @@ export class Publisher { stream.abort(cancel); return; } - const refusal = this.#denied(name) ? denied : this.#failed(err, "track_status"); + const refusal = !this.#gate.allows(name) ? denied : this.#failed(err, "track_status"); await this.#refuse(stream, msg.requestId, TRACK_STATUS_ERROR_14, refusal); return; } @@ -1854,7 +1849,7 @@ export class Publisher { stream.close(); await stream.writer.closed.catch(() => undefined); } finally { - disposeGrant(); + watch.dispose(); hold.abort(); } } diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 555160ebfb2..ffdeb0b030e 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -1,6 +1,7 @@ import { Instant, Milli, Nano, now, sleep } from "@moq/clock"; import { type Dispose, type Getter, Once, race, Signal } from "@moq/signals"; import type { Grant } from "../auth.ts"; +import { Gate } from "../auth_gate.ts"; import { enforceGrant } from "../auth_session.ts"; import type * as broadcast from "../broadcast.ts"; import { Withdrawal } from "../connection/withdrawal.ts"; @@ -398,6 +399,8 @@ export class Publisher { // withdraws what it no longer covers. Undefined until the peer answers, and forever on a // version without AUTH, which allows everything. #grant?: Getter; + // Holds every incoming request to that grant from dispatch to its end. + #gate: Gate; // Resolves once the tokens this session presented at setup are answered, so nothing is // announced before the grant it would be checked against. @@ -435,6 +438,7 @@ export class Publisher { this.#interests = new Interests(new Slots(limits.subscriptions, { quic, reason: "too many subscriptions" })); this.#fetches = new Slots(limits.fetches, { quic, reason: "too many fetches" }); this.#grant = auth?.grant; + this.#gate = new Gate(auth?.grant); this.#ready = auth?.ready ?? Promise.resolve(); this.version = version; this.hop = hop; @@ -681,7 +685,7 @@ export class Publisher { // broadcast resolves is never missed. const revoked = unauthorized(msg.broadcast); let serving: track.Subscriber | undefined; - const watch = this.#watch(msg.broadcast, () => { + const watch = this.#gate.watch(msg.broadcast, () => { console.debug(`publish revoked: broadcast=${msg.broadcast} track=${msg.track}`); serving?.close(revoked); stream.abort(revoked); @@ -696,19 +700,6 @@ export class Publisher { } } - // Watch whether our grant still lets us publish `broadcast`, calling `onRevoke` once - // it does not. Arm it before the request's first check. - #watch(broadcast: Path.Valid, onRevoke: () => void): { revoked: () => boolean; dispose: () => void } { - let revoked = false; - const dispose = - this.#grant?.subscribe(() => { - if (revoked || !this.#denied(broadcast)) return; - revoked = true; - onRevoke(); - }) ?? (() => {}); - return { revoked: () => revoked || this.#denied(broadcast), dispose }; - } - async #serveSubscribe( msg: Subscribe, stream: Stream, @@ -869,7 +860,7 @@ export class Publisher { // open as long as its track, so a check at accept alone would keep serving after a shrink. const revoked = unauthorized(msg.broadcast); let fetched: group.Consumer | undefined; - const watch = this.#watch(msg.broadcast, () => { + const watch = this.#gate.watch(msg.broadcast, () => { console.debug(`fetch revoked: broadcast=${msg.broadcast} track=${msg.track} group=${msg.group}`); fetched?.close(revoked); stream.abort(revoked); @@ -1280,7 +1271,7 @@ export class Publisher { hold.signal.addEventListener("abort", () => interest.release(), { once: true }); // Armed before the first check and held until the answer is acknowledged, so a grant // that shrinks while the answer is blocked on flow control resets it, never sends it. - const watch = this.#watch(msg.broadcast, () => { + const watch = this.#gate.watch(msg.broadcast, () => { console.debug(`track info revoked: broadcast=${msg.broadcast} track=${msg.track}`); hold.abort(); stream.writer.reset(unauthorized(msg.broadcast)); @@ -1323,12 +1314,6 @@ export class Publisher { } } - // Whether our grant excludes publishing this broadcast. No grant yet allows it. - #denied(broadcast: Path.Valid): boolean { - const grant = this.#grant?.peek(); - return grant !== undefined && !grant.publish.matches(broadcast); - } - // Resolve (and cache) a track's immutable TRACK_INFO by asking the application. // `resolveTrackInfo` triggers a TrackRequest the app answers with accept(TrackInfo); // only the immutable properties are needed (not the groups). Cached because they're diff --git a/quest/m1/auth/README.md b/quest/m1/auth/README.md index 8ebea95db45..270bef2c170 100644 --- a/quest/m1/auth/README.md +++ b/quest/m1/auth/README.md @@ -133,8 +133,6 @@ because `main` assigned 0x3A to NOT_FETCHABLE first. - [Live requests follow their auth](/quest/m1/auth/request-update.md) - a route swaps its auth and the session sends the update, keeping the request up - [Auth outlives its session](/quest/m1/auth/issuer.md) - one `auth::Handle` spans a connection's sessions, so a route's auth survives a reconnect - [Requests name their auth](/quest/m1/auth/request-auth-id.md) - lite-07 AUTH carries an ID and each request names the auth that admits it -- [Request gate](/quest/m1/auth/request-gate.md) - every incoming request is - refused up front and held to the grant through a gate its dispatcher owns - [Bindings](/quest/m1/auth/bindings.md) - grants and tokens reach every binding through moq-ffi - [Token in band](/quest/m1/auth/token-in-band.md) - the credential can leave diff --git a/quest/m1/auth/request-gate.md b/quest/m1/auth/request-gate.md deleted file mode 100644 index b5204651846..00000000000 --- a/quest/m1/auth/request-gate.md +++ /dev/null @@ -1,54 +0,0 @@ -# [M] Every incoming request is held to the grant by construction - -## Goal - -Every request a peer sends, on lite and IETF, in Rust and `@moq/net`, is -checked against the grant when it is dispatched and held to it for its whole -life: resolution, reads, and writes. A new request type cannot be served -without a gate. Per-handler checks already missed standalone FETCH and -TRACK_STATUS in Rust and the TRACK re-check and setup announces in JS, all -fixed in #4039, as were the joining FETCH and the TRACK_STATUS and TRACK_INFO -answers while they are written. Behavior stays the same: no wire or public API -change, and the existing auth tests are the regression net. - -## Plan - -Decisions settled while planning (2026-10-09): - -- **The dispatcher owns the gate.** It refuses an out-of-grant request up - front, the same way for every type, and hands the live gate to the handler, - whose signature requires one. The handler still picks its graceful end on a - narrowed grant (PUBLISH_DONE `Unauthorized`, a REQUEST_ERROR, or a stream - reset once FETCH_OK is out). Racing the whole handler future and dropping it - was rejected because it loses those graceful refusals. -- **Rust and JS together**, so the shape and names stay mirrored. -- ✅ Decided 2026-10-15 (maintainer, #5148): **effective authorization.** - An unnamed request is held to the union of session-lifetime auths; a - request named by its own auth (a moq-transport request token today, - `auth=id` on lite later) is held to that auth's issued grant alone, and - ends with it. Peer-issued AUTH_OK grants never authorize incoming - requests. Preserve renewal (a new auth replacing the old), revocation, - and expiry. Joining FETCH follows the named - subscription's live authorization and ends with that subscription; - standalone FETCH gains no request-token authority. JS mirrors the gate - structure. -- **Structure only.** Finding new gaps is not the goal, but any request type - the refactor shows was ungated gets a regression test in the same PR. - Initial verification already consumes request capacity and observes - cancellation (#5148); those fixes must not wait here. - -Today `auth::Gate` is built inside each handler: the IETF publisher's -SUBSCRIBE, standalone FETCH, and TRACK_STATUS; the IETF subscriber; and the -lite publisher and subscriber. JS checks `#denied` and `#watch` per handler. -A joining FETCH carries its subscription's namespace and observes that -subscription's effective authorization. A surviving cache is not authority -to continue a dependent join after the subscription ends. - -Test named admission outside the session union, session-union changes -without revoking a still-valid named grant, revoke/renewal/expiry of a named -auth, and dependent joining FETCH. Keep graceful terminal messages. - -## Related - -- [JS fetch grant watch](https://github.com/moq-dev/moq/pull/5260) - the outgoing - side: a JS `fetchGroup` this session issued ends when its path leaves the grant diff --git a/quest/m3/auth-token-fetch.md b/quest/m3/auth-token-fetch.md index 16208c93030..02382870403 100644 --- a/quest/m3/auth-token-fetch.md +++ b/quest/m3/auth-token-fetch.md @@ -19,7 +19,7 @@ peer can subscribe to a track but cannot look it up or fetch its history. Open question for the maintainer. This reverses a settled decision: "standalone FETCH gains no request-token authority" (2026-10-12, in [#5148](https://github.com/moq-dev/moq/pull/5148)'s quest and -[Request gate](/quest/m1/auth/request-gate.md)). +[the request gate refactor](https://github.com/moq-dev/moq/pull/5782)). 1. (recommended) Do it in m1, after the request gate. Move this file under [In-band auth](/quest/m1/auth/README.md) and list it in that epic's @@ -49,7 +49,3 @@ token. Run cross-language interop for the wire change. Public API: none expected beyond widening what `auth::RequestKind` names. Wire: the existing AUTHORIZATION TOKEN parameter (0x03), now sent and honored on TRACK_STATUS and FETCH. - -## Required - -- [Request gate](/quest/m1/auth/request-gate.md) - the dispatcher-owned gate this extends to two more request types diff --git a/rs/moq-net/src/auth.rs b/rs/moq-net/src/auth.rs index 5e14cd63fa7..1235203cd6b 100644 --- a/rs/moq-net/src/auth.rs +++ b/rs/moq-net/src/auth.rs @@ -1646,13 +1646,11 @@ impl Gate { } } - /// A gate on the peer's grant alone, for a peer's request its own token authorized. + /// Hold the request to the peer's grant alone, once its own token authorized it. #[cfg_attr(not(feature = "async"), allow(dead_code))] - pub(crate) fn granted(handle: Handle, path: crate::PathOwned, direction: Direction) -> Self { - Self { - part: Part::Granted, - ..Self::new(handle, path, direction) - } + pub(crate) fn named(mut self) -> Self { + self.part = Part::Granted; + self } /// A gate for a request under a namespace the peer announced, while `live` stays open: @@ -1682,6 +1680,43 @@ impl Gate { } } + /// Whether the peer's request token must be verified separately from session scope. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn token_decides(&self) -> bool { + self.handle.token_decides(self.direction) + } + + /// Verify an incoming request's token against the local acceptor. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn verify(&self, token: crate::setup::Token, hold: Option) -> RequestVerdict { + self.handle.verify_request(token, hold) + } + + /// What the session allows in this gate's direction, once it may have changed since `epoch`. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn poll_permit(&self, epoch: &mut u64, waiter: &kio::Waiter) -> Poll { + self.handle.poll_permit(self.direction, epoch, waiter) + } + + /// Preserve setup ordering before the first discovery snapshot. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn poll_setup_answered(&self, waiter: &kio::Waiter) -> Poll<()> { + self.handle.poll_setup_answered(waiter) + } + + /// Resolve one operation under this gate without dropping the handler's refusal path. + #[cfg(feature = "async")] + pub(crate) async fn guard(&mut self, operation: impl std::future::Future>) -> Result { + let mut operation = std::pin::pin!(operation); + kio::wait(|waiter| { + if self.poll_denied(waiter).is_ready() { + return Poll::Ready(Err(Error::Unauthorized)); + } + waiter.poll_future(operation.as_mut()) + }) + .await + } + /// Resolve once the session no longer allows the path. A union that is still /// `None` (no answer yet, or a version without AUTH) and a session never limited /// allow everything. diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index f86cfcb4e6e..145d0035596 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -449,6 +449,8 @@ struct Namespaces { woken: Arc, /// What we may advertise: our grant (MoQ Auth) and the ceiling on the peer. permit: crate::auth::Permit, + /// The dispatcher-owned gate for a solicited namespace request. + gate: Option, /// The auth epoch last applied to `permit`. epoch: u64, /// Whether this discovery cursor includes hidden namespaces. @@ -466,6 +468,7 @@ impl Namespaces { inflight: VecDeque::new(), woken: Default::default(), permit: Default::default(), + gate: None, epoch: 0, hidden: false, } @@ -618,6 +621,12 @@ pub(super) struct Publisher { priority: PriorityQueue, } +/// Authorization selected by dispatch: a standalone path or a live parent subscription. +enum FetchGate { + Standalone(crate::auth::Gate), + Joining { id: RequestId, auth: crate::auth::Handle }, +} + struct Serve(Arc); impl Drop for Serve { @@ -801,7 +810,12 @@ where return Err(Error::WrongSize); } tracing::debug!(message = ?msg, "received subscribe"); - let task = this.run_subscribe_stream(stream, msg); + let gate = crate::auth::Gate::new( + this.auth.clone(), + msg.track_namespace.to_owned(), + crate::auth::Direction::Publish, + ); + let task = this.serve_subscribe_stream(stream, msg, gate); async move { let _serve = serve; if let Err(err) = task.await { @@ -825,9 +839,10 @@ where return Err(Error::WrongSize); } tracing::debug!(message = ?msg, "received fetch"); + let gate = this.fetch_gate(&msg); async move { let _serve = serve; - if let Err(err) = this.run_fetch_stream(stream, msg).await { + if let Err(err) = this.serve_fetch_stream(stream, msg, gate).await { tracing::debug!(%err, "fetch stream error"); } } @@ -875,9 +890,14 @@ where .maybe_boxed()); } tracing::debug!(message = ?msg, "received subscribe_namespace"); + let gate = crate::auth::Gate::new( + this.auth.clone(), + msg.namespace.to_owned(), + crate::auth::Direction::Publish, + ); async move { let mut session = this.session.clone(); - match this.run_subscribe_namespace_stream(stream, msg).await { + match this.serve_subscribe_namespace_stream(stream, msg, gate).await { // Such as a malformed answer to an advertisement this request solicited. Err(err) if super::subscriber::is_protocol_violation(&err) => session.close( super::error::to_session_code(&crate::SessionError::from(&err)), @@ -931,8 +951,13 @@ where return Err(Error::WrongSize); } tracing::debug!(message = ?msg, "received track_status"); + let gate = crate::auth::Gate::new( + this.auth.clone(), + msg.track_namespace.to_owned(), + crate::auth::Direction::Publish, + ); async move { - if let Err(err) = this.run_track_status_stream(stream, msg).await { + if let Err(err) = this.serve_track_status_stream(stream, msg, gate).await { tracing::debug!(%err, "track_status stream error"); } } @@ -967,10 +992,25 @@ where } /// Handle a SUBSCRIBE on its bidi stream. + #[cfg(test)] fn run_subscribe_stream( + self, + stream: Stream, + msg: ietf::Subscribe<'_>, + ) -> impl std::future::Future> { + let gate = crate::auth::Gate::new( + self.auth.clone(), + msg.track_namespace.to_owned(), + crate::auth::Direction::Publish, + ); + self.serve_subscribe_stream(stream, msg, gate) + } + + fn serve_subscribe_stream( self, mut stream: Stream, msg: ietf::Subscribe<'_>, + gate: crate::auth::Gate, ) -> impl std::future::Future> { // Register during dispatch, before either request task can be polled. let join = (!Filter::is_draft20(self.version)).then(|| self.register_join(msg.request_id)); @@ -992,17 +1032,18 @@ where // this request alone (MoQ request-token), even where the limit covers it. What the // peer granted us bounds what we send either way, and never stands in for our // authorization of it. - let path = msg.track_namespace.as_str(); let token = match &msg.authorization_token { Some(token) - if self.auth.within_grant(crate::auth::Direction::Publish, path) - && self.auth.token_decides(crate::auth::Direction::Publish) => + if self + .auth + .within_grant(crate::auth::Direction::Publish, msg.track_namespace.as_str()) + && gate.token_decides() => { Some(token.0.clone()) } _ => None, }; - if token.is_none() && !self.auth.allows(crate::auth::Direction::Publish, path) { + if token.is_none() && !gate.allows() { let err = Error::Unauthorized; return self.reject_subscribe(stream, request_id, &err, "not granted").await; } @@ -1047,7 +1088,7 @@ where // once the limit does too. let (mut gate, request_grant) = match token { Some(token) => { - let verdict = self.auth.verify_request(token, Some(slot.clone())); + let verdict = gate.verify(token, Some(slot.clone())); // The subscriber cancelling or the session closing withdraws the token. let mut session = self.session.clone(); let verified = kio::wait(|waiter| { @@ -1072,14 +1113,7 @@ where Some(slot.clone()), ); match admitted { - Ok(grant) => ( - crate::auth::Gate::granted( - self.auth.clone(), - msg.track_namespace.to_owned(), - crate::auth::Direction::Publish, - ), - Some(grant), - ), + Ok(grant) => (gate.named(), Some(grant)), Err(refusal) => { return self .reject_subscribe(stream, request_id, &refusal.error, &refusal.reason) @@ -1087,14 +1121,7 @@ where } } } - None => ( - crate::auth::Gate::new( - self.auth.clone(), - msg.track_namespace.to_owned(), - crate::auth::Direction::Publish, - ), - None, - ), + None => (gate, None), }; let _slot = slot; @@ -2072,7 +2099,35 @@ where /// The answer is buffered before replying: FETCH_OK names where the response ends, /// which a range running past the track only learns by reading it, and a refusal can /// still replace it until then. - async fn run_fetch_stream(mut self, mut stream: Stream, msg: ietf::Fetch<'_>) -> Result<(), Error> { + fn fetch_gate(&self, msg: &ietf::Fetch<'_>) -> FetchGate { + match &msg.fetch_type { + FetchType::Standalone { namespace, .. } | FetchType::Filtered { namespace, .. } => FetchGate::Standalone( + crate::auth::Gate::new(self.auth.clone(), namespace.to_owned(), crate::auth::Direction::Publish), + ), + FetchType::RelativeJoining { + subscriber_request_id, .. + } + | FetchType::AbsoluteJoining { + subscriber_request_id, .. + } => FetchGate::Joining { + id: *subscriber_request_id, + auth: self.auth.clone(), + }, + } + } + + #[cfg(test)] + async fn run_fetch_stream(self, stream: Stream, msg: ietf::Fetch<'_>) -> Result<(), Error> { + let gate = self.fetch_gate(&msg); + self.serve_fetch_stream(stream, msg, gate).await + } + + async fn serve_fetch_stream( + mut self, + mut stream: Stream, + msg: ietf::Fetch<'_>, + admission: FetchGate, + ) -> Result<(), Error> { let _slot = match self.fetches.acquire() { Ok(slot) => slot, Err(err) => { @@ -2173,13 +2228,14 @@ where start, end, } => { - if !self.auth.allows(crate::auth::Direction::Publish, namespace.as_str()) { + let FetchGate::Standalone(mut gate) = admission else { + return Err(Error::Unauthorized); + }; + if !gate.allows() { return self .reject_fetch(stream, msg.request_id, &Error::Unauthorized, "not granted") .await; } - let gate = - crate::auth::Gate::new(self.auth.clone(), namespace.to_owned(), crate::auth::Direction::Publish); // An End Object of 0 asks for the whole End Group. let end = match end.object { @@ -2194,7 +2250,10 @@ where // The peer must have seen the announcement to name this namespace, so this // resolves like a SUBSCRIBE does. - let broadcast = match self.serving_origin().await.request_broadcast(&namespace, None).await { + let broadcast = match gate + .guard(async { self.serving_origin().await.request_broadcast(&namespace, None).await }) + .await + { Ok(broadcast) => broadcast, Err(err) => return self.reject_fetch(stream, msg.request_id, &err, &err.to_string()).await, }; @@ -2221,19 +2280,21 @@ where | FetchType::AbsoluteJoining { subscriber_request_id, .. } => { + let FetchGate::Joining { id, auth } = admission else { + return Err(Error::Unauthorized); + }; + debug_assert_eq!(id, subscriber_request_id); let (end, cache, namespace, ranking, token) = match self.joined(&mut stream, subscriber_request_id).await? { Ok(joined) => joined, Err((err, reason)) => return self.reject_fetch(stream, msg.request_id, &err, reason).await, }; - // The cache outlives the subscription, so the fetch holds its own gate rather - // than trusting the subscription's to stop it. One the subscription's request - // token authorized is held to the peer's grant alone, as the subscription is. - let gate = match token { - true => crate::auth::Gate::granted(self.auth.clone(), namespace, crate::auth::Direction::Publish), - false => crate::auth::Gate::new(self.auth.clone(), namespace, crate::auth::Direction::Publish), - }; - lapses_with = token.then_some(subscriber_request_id); + // The cache outlives the subscription, so the fetch holds its own gate and ends + // with the subscription rather than trusting the cache. One the subscription's + // request token authorized is held to the peer's grant alone, as the subscription is. + let gate = crate::auth::Gate::new(auth, namespace, crate::auth::Direction::Publish); + let gate = if token { gate.named() } else { gate }; + lapses_with = Some(subscriber_request_id); let start = match fetch_type { FetchType::RelativeJoining { group_offset, .. } => end.group.saturating_sub(group_offset), FetchType::AbsoluteJoining { group_id, .. } if group_id <= end.group => group_id, @@ -2598,39 +2659,47 @@ where /// The track resolves as it would for a SUBSCRIBE, so a relay asks its upstream for a /// track it does not hold yet and the answer waits on that. The Largest Location is /// whatever the cache holds now: nothing waits for a fresher one. - async fn run_track_status_stream( + #[cfg(test)] + fn run_track_status_stream( + self, + stream: Stream, + msg: ietf::TrackStatus<'_>, + ) -> impl std::future::Future> { + let gate = crate::auth::Gate::new( + self.auth.clone(), + msg.track_namespace.to_owned(), + crate::auth::Direction::Publish, + ); + self.serve_track_status_stream(stream, msg, gate) + } + + async fn serve_track_status_stream( self, mut stream: Stream, msg: ietf::TrackStatus<'_>, + mut gate: crate::auth::Gate, ) -> Result<(), Error> { if let Some(error) = super::namespace::refusal(&msg.track_namespace, self.version) { return super::request_stream::reject(stream, error).await; } let request_id = msg.request_id; - // Answer only for what our grant lets us publish (MoQ Auth): checked before the - // track resolves, and held until the answer. - if !self - .auth - .allows(crate::auth::Direction::Publish, msg.track_namespace.as_str()) - { + if !gate.allows() { return self.reject_track_status(stream, request_id, &Error::Unauthorized).await; } - let mut gate = crate::auth::Gate::new( - self.auth.clone(), - msg.track_namespace.to_owned(), - crate::auth::Direction::Publish, - ); - - let broadcast = match self - .serving_origin() - .await - .request_broadcast(&msg.track_namespace, None) + let broadcast = match gate + .guard(async { + self.serving_origin() + .await + .request_broadcast(&msg.track_namespace, None) + .await + }) .await { Ok(broadcast) => broadcast, Err(err) => return self.reject_track_status(stream, request_id, &err).await, }; + let track = match broadcast.track(&msg.track_name) { Ok(track) => track, Err(err) => return self.reject_track_status(stream, request_id, &err).await, @@ -3432,7 +3501,7 @@ where /// peer that would rather ask says so with the MoQ Solicit extension /// ([`solicit`](super::solicit)), /// and then this loop does nothing. - /// On draft-16 and later [`Self::run_subscribe_namespace_stream`] also carries every + /// On draft-16 and later [`Self::serve_subscribe_namespace_stream`] also carries every /// match, so a peer that did not ask hears each namespace both ways. A NAMESPACE is /// discovery only, not a second route. pub async fn run_publish_namespaces(self) -> Result<(), Error> { @@ -3463,10 +3532,25 @@ where /// this stream, whatever the peer's SETUP said. Draft-14 and 15 predate that /// message and still answer with PUBLISH_NAMESPACE only for what /// [`Self::run_publish_namespaces`] does not already say. + #[cfg(test)] async fn run_subscribe_namespace_stream( + self, + stream: Stream, + msg: ietf::SubscribeNamespace<'_>, + ) -> Result<(), Error> { + let gate = crate::auth::Gate::new( + self.auth.clone(), + msg.namespace.to_owned(), + crate::auth::Direction::Publish, + ); + self.serve_subscribe_namespace_stream(stream, msg, gate).await + } + + async fn serve_subscribe_namespace_stream( self, mut stream: Stream, msg: ietf::SubscribeNamespace<'_>, + gate: crate::auth::Gate, ) -> Result<(), Error> { if let Some(error) = super::namespace::refusal(&msg.namespace, self.version) { return super::request_stream::reject(stream, error).await; @@ -3528,9 +3612,9 @@ where let (initial, active) = match declared.active_count { true => { if declared.auth { - kio::wait(|waiter| self.auth.poll_setup_answered(waiter)).await; + kio::wait(|waiter| gate.poll_setup_answered(waiter)).await; } - let (permit, _) = self.permit_now(); + let (permit, _) = self.permit_now(Some(&gate)); let initial = Self::snapshot(&mut announced); let count = initial .iter() @@ -3583,17 +3667,21 @@ where }; let mut ns = Namespaces::new(peer, target); + ns.gate = Some(gate); ns.hidden = !declared.hidden || msg.hidden || self.origin.includes_hidden(); self.run_namespaces(announced, prefix, ns, initial).await } /// Our publish permit as it stands now, with the auth epoch it was read at. - fn permit_now(&self) -> (crate::auth::Permit, u64) { + fn permit_now(&self, gate: Option<&crate::auth::Gate>) -> (crate::auth::Permit, u64) { let mut epoch = 0; - match self - .auth - .poll_permit(crate::auth::Direction::Publish, &mut epoch, &kio::Waiter::noop()) - { + let permit = match gate { + Some(gate) => gate.poll_permit(&mut epoch, &kio::Waiter::noop()), + None => self + .auth + .poll_permit(crate::auth::Direction::Publish, &mut epoch, &kio::Waiter::noop()), + }; + match permit { Poll::Ready(permit) => (permit, epoch), Poll::Pending => (Default::default(), 0), } @@ -3644,7 +3732,7 @@ where if self.peer_setup.get().await.auth { kio::wait(|waiter| self.auth.poll_setup_answered(waiter)).await; } - (ns.permit, ns.epoch) = self.permit_now(); + (ns.permit, ns.epoch) = self.permit_now(ns.gate.as_ref()); for update in initial { self.apply_update(&mut ns, &prefix, update, true).await?; } @@ -3681,6 +3769,7 @@ where inflight, woken, updates, + gate, .. } = &mut ns; kio::wait(|waiter| { @@ -3689,7 +3778,11 @@ where } // A grant change applies before the next update, so a namespace it no // longer covers is withdrawn rather than re-sent. - if let Poll::Ready(permit) = self.auth.poll_permit(crate::auth::Direction::Publish, epoch, waiter) { + let permit = match gate { + Some(gate) => gate.poll_permit(epoch, waiter), + None => self.auth.poll_permit(crate::auth::Direction::Publish, epoch, waiter), + }; + if let Poll::Ready(permit) = permit { return Poll::Ready(NamespaceEvent::Regrant(permit)); } let mut cx = waiter.context(); @@ -9685,12 +9778,14 @@ mod serve_tests { } } - /// A joining FETCH on a subscription its own request token authorized holds no authority - /// of its own: revoking the token ends the subscription, and the fetch with it, even one - /// parked on stream credit after FETCH_OK. + /// A joining FETCH holds no authority of its own: revocation or dropping its parent + /// ends it even while stream credit blocks the response after FETCH_OK. #[moq_clock::test] - async fn a_joining_fetch_ends_with_its_token_subscription() { - for version in JOINING_DRAFTS { + async fn a_joining_fetch_ends_with_its_subscription() { + for (version, token) in JOINING_DRAFTS + .into_iter() + .flat_map(|version| [false, true].map(|token| (version, token))) + { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let broadcast = origin.publish("room", crate::origin::Route::default()).unwrap(); let track = broadcast.create_track("video", crate::track::Info::timed()).unwrap(); @@ -9706,7 +9801,10 @@ mod serve_tests { let auth = crate::auth::Handle::new(false); let mut requests = auth.requests().unwrap(); // A limit of nothing, so only the token admits the subscription. - auth.authorize(&crate::auth::Grant::default()); + auth.authorize(&match token { + true => crate::auth::Grant::default(), + false => crate::auth::Grant::all(), + }); let publisher = Publisher::new( crate::time::sim(), session.clone(), @@ -9719,13 +9817,22 @@ mod serve_tests { .with_auth(auth.clone()); let stream = Stream::open(&mut session.clone(), version).await.unwrap(); - let mut subscription = std::pin::pin!(publisher.clone().run_subscribe_stream(stream, token_subscribe())); - let accepted = async { requests.next().await.expect("a request") }; - let request = match futures::future::select(std::pin::pin!(accepted), subscription.as_mut()).await { - futures::future::Either::Left((request, _)) => request, - futures::future::Either::Right((res, _)) => panic!("{version}: subscription ended: {res:?}"), + let mut msg = token_subscribe(); + if !token { + msg.authorization_token = None; + } + let mut subscription = Box::pin(publisher.clone().run_subscribe_stream(stream, msg)); + let enforced = if token { + let accepted = async { requests.next().await.expect("a request") }; + let request = match futures::future::select(std::pin::pin!(accepted), subscription.as_mut()).await { + futures::future::Either::Left((request, _)) => request, + futures::future::Either::Right((res, _)) => panic!("{version}: subscription ended: {res:?}"), + }; + Some(request.accept(crate::auth::Grant::all())) + } else { + None }; - let enforced = request.accept(crate::auth::Grant::all()); + let joined = publisher.joins.wait(|joins| match joins.get(&RequestId(REQUEST_ID)) { Some(Some(_)) => Poll::Ready(()), _ => Poll::Pending, @@ -9767,10 +9874,16 @@ mod serve_tests { "{version}: FETCH_OK never went out" ); - enforced.revoke(crate::SessionError::Unauthorized, "revoked"); - let _ = moq_clock::timeout(std::time::Duration::from_secs(1), subscription.as_mut()) - .await - .unwrap_or_else(|_| panic!("{version}: the revoked subscription kept serving")); + if let Some(enforced) = enforced { + enforced.revoke(crate::SessionError::Unauthorized, "revoked"); + let _ = moq_clock::timeout(std::time::Duration::from_secs(1), subscription.as_mut()) + .await + .unwrap_or_else(|_| panic!("{version}: the revoked subscription kept serving")); + } else { + // Session authorization stays valid, and the cache stays alive. + drop(subscription); + } + let res = moq_clock::timeout(std::time::Duration::from_secs(1), fetch) .await .unwrap_or_else(|_| panic!("{version}: the fetch outlived its subscription")); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index d237a236743..1f454765a0e 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -584,6 +584,14 @@ struct BroadcastState { _slot: crate::auth::Hold, } +/// A namespace request and the authorization selected at dispatch. +struct IncomingNamespace<'a> { + msg: ietf::PublishNamespace<'a>, + peer: cluster::Peer, + declared: Option, + gate: crate::auth::Gate, +} + /// What authorized one advertisement. enum Admit { /// The session, so the limit holds it back while it does not cover the namespace. @@ -1253,8 +1261,19 @@ where return Err(Error::WrongSize); } tracing::debug!(message = ?msg, "received publish_namespace"); + let gate = crate::auth::Gate::new( + this.auth.clone(), + msg.track_namespace.to_owned(), + crate::auth::Direction::Subscribe, + ); + let incoming = IncomingNamespace { + msg, + peer, + declared, + gate, + }; async move { - if let Err(err) = this.run_publish_namespace_stream(stream, msg, peer, declared).await { + if let Err(err) = this.serve_publish_namespace_stream(stream, incoming).await { // An advertisement update is decoded here rather than in the // dispatcher, so nothing else would surface a malformed one. The // cluster draft requires closing the session on those; a stream @@ -1328,13 +1347,42 @@ where } /// Handle an incoming PUBLISH_NAMESPACE on its bidi stream. + #[cfg(test)] async fn run_publish_namespace_stream( &mut self, - mut stream: Stream, + stream: Stream, msg: ietf::PublishNamespace<'_>, peer: cluster::Peer, declared: Option, ) -> Result<(), Error> { + let gate = crate::auth::Gate::new( + self.auth.clone(), + msg.track_namespace.to_owned(), + crate::auth::Direction::Subscribe, + ); + self.serve_publish_namespace_stream( + stream, + IncomingNamespace { + msg, + peer, + declared, + gate, + }, + ) + .await + } + + async fn serve_publish_namespace_stream( + &mut self, + mut stream: Stream, + incoming: IncomingNamespace<'_>, + ) -> Result<(), Error> { + let IncomingNamespace { + msg, + peer, + declared, + gate, + } = incoming; if let Some(error) = super::namespace::refusal(&msg.track_namespace, self.version) { return super::request_stream::reject(stream, error).await; } @@ -1367,7 +1415,7 @@ where // request-token), even where the limit covers it, and a verified one is exempt from the // limit while it lasts. let token = match &msg.authorization_token { - Some(token) if self.auth.token_decides(crate::auth::Direction::Subscribe) => Some(token.0.clone()), + Some(token) if gate.token_decides() => Some(token.0.clone()), _ => None, }; let (admit, token_grant) = match token { @@ -1379,7 +1427,7 @@ where true => None, false => Some(std::sync::Arc::new(self.acquire_announce()?)), }; - let verdict = self.auth.verify_request(token, slot.clone()); + let verdict = gate.verify(token, slot.clone()); // The session closing or the peer withdrawing the announce withdraws the token. let mut session = self.session.clone(); let version = self.version; diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 292d49d18c8..90e3c70aff4 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -269,7 +269,9 @@ impl Publisher { auth: Option, ) -> Result<(), Error> { let mut run = AnnounceRun::new(crate::PathOwned::default(), self_origin, version); - run.auth = auth; + run.auth = auth.map(|handle| { + crate::auth::Gate::new(handle, crate::Path::empty().to_owned(), crate::auth::Direction::Publish) + }); kio::wait(|waiter| run.poll(stream, origin, announced, waiter)).await } } @@ -947,7 +949,11 @@ impl AnnounceServe { }; let announced = origin.announced(); let mut run = AnnounceRun::new(prefix, self.shared.self_origin, self.shared.version); - run.auth = Some(self.shared.auth.clone()); + run.auth = Some(crate::auth::Gate::new( + self.shared.auth.clone(), + run.prefix.clone(), + crate::auth::Direction::Publish, + )); self.state = AnnounceState::Run { origin, announced, run }; } } @@ -971,7 +977,7 @@ struct AnnounceRun { phase: AnnouncePhase, // Our grant and the ceiling on the peer: only what both let us publish is // announced, and a shrink withdraws what they no longer cover. - auth: Option, + auth: Option, epoch: u64, // What we may announce right now. permit: crate::auth::Permit, @@ -1311,8 +1317,7 @@ impl AnnounceRun { // advertises something it would withdraw, or abort over, a moment later. if let Some(auth) = &self.auth { ready!(auth.poll_setup_answered(waiter)); - if let Poll::Ready(permit) = auth.poll_permit(crate::auth::Direction::Publish, &mut self.epoch, waiter) - { + if let Poll::Ready(permit) = auth.poll_permit(&mut self.epoch, waiter) { self.permit = permit; } } @@ -1336,7 +1341,7 @@ impl AnnounceRun { // A grant change applies before the next update, so a route it no longer // covers is withdrawn rather than re-sent. if let Some(auth) = &self.auth - && let Poll::Ready(permit) = auth.poll_permit(crate::auth::Direction::Publish, &mut self.epoch, waiter) + && let Poll::Ready(permit) = auth.poll_permit(&mut self.epoch, waiter) && self.permit != permit { self.regrant(stream, origin, permit)?; From 4f09ce03e01055ebe21ddd525ff31ae52a7a8b68 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 09:44:36 -0700 Subject: [PATCH 2/2] chore(net): mark the JS auth Gate as private plumbing Co-Authored-By: Claude Opus 5.5 --- js/net/src/auth_gate.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/js/net/src/auth_gate.ts b/js/net/src/auth_gate.ts index 299e5d9cf28..b32dd0fe718 100644 --- a/js/net/src/auth_gate.ts +++ b/js/net/src/auth_gate.ts @@ -2,6 +2,7 @@ import type { Dispose, Getter } from "@moq/signals"; import type { Grant } from "./auth.ts"; import type * as Path from "./path.ts"; +// private: plumbing /** Whether our grant still lets us serve one incoming request, built when it is dispatched. @internal */ export class Gate { #grant?: Getter;