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..b32dd0fe718 --- /dev/null +++ b/js/net/src/auth_gate.ts @@ -0,0 +1,31 @@ +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; + + 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 da8f0b86290..4906818de8c 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"; @@ -304,6 +305,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. @@ -354,6 +357,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; @@ -390,7 +394,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); @@ -430,7 +434,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" }; @@ -728,11 +732,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; @@ -749,7 +751,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. @@ -1139,12 +1141,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}. @@ -1219,7 +1215,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; @@ -1438,7 +1434,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; @@ -1793,7 +1789,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; } @@ -1807,8 +1803,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)); }); @@ -1839,7 +1834,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; } @@ -1851,7 +1846,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)?;