feat(net): re-authorize a live session's grant in place

doc/bin/relay/auth.md

@@ -64,9 +64,15 @@ The client and relay snapshot each accepted grant's deadline on a monotonic
 clock, so later polls, outages, and wall-clock adjustments do not restart it.
 
 **Revalidate and outage.** On the cadence the relay POSTs `revalidate` with the
-same request. A grant applies: a changed `root` or `mounts`, or one that no longer covers
-what the session holds closes it with `Unauthorized` (the live session is not
-resized in place), and so does a flipped `peer`; a changed `tier` keeps the
+same request. A grant applies: its patterns resize 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. Subscriptions and fetches outside a narrower grant
+reset with `Unauthorized`, the broadcasts it published outside 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` or `mounts` closes the session with `Unauthorized`, and so does a
+flipped `peer`; a changed `tier` keeps the
 session and moves its stats: its presence counts under the new tier from then
 on, as does each group and subscription it starts afterwards, while one already
 in flight finishes where it began. A 401 or 403 closes the session now, as does a 2xx

doc/lib/rs/moq-net.md

@@ -115,6 +115,29 @@ shrinks withdraws the announcements and cancels the subscriptions it no longer
 covers, and leaves the session up. On moq-lite each cancelled stream resets with
 `StreamError::Unauthorized`.
 
+`session.auth().authorize(&grant)` authorizes the peer for `grant` on a live
+session; calling it again re-authorizes, narrower or wider. It works on every
+version, whether or not the peer speaks AUTH: the session enforces it itself, and
+never past its origin handles. A narrower grant ends what falls outside at once: announcements to the peer
+retract, its new requests are refused, its subscriptions and fetches reset with
+`Unauthorized`, and the broadcasts it published abort, so local readers see
+`Unauthorized` too. A wider one brings back what the old grant held back:
+announcements are made again, the peer's own announcements return to the origin,
+and new requests are accepted. The rest of the session carries on. When the
+session answers the peer's connection credential itself, the peer is sent the
+grant.
+
+```rust
+// Deafen alice's audio: the peer may still subscribe to her video.
+let video: Pattern = "room/alice/video/**".parse()?;
+session.auth().authorize(&auth::Grant {
+    subscribe: video.into(),
+    ..auth::Grant::all()
+});
+// Undeafen.
+session.auth().authorize(&auth::Grant::all());
+```
+
 ## Patterns
 
 `Pattern` describes a set of paths; `Patterns` is a union reduced by

js/net/src/lite/publisher.test.ts

@@ -1,5 +1,8 @@
 import { expect, mock, spyOn, test } from "bun:test";
 import { Signal } from "@moq/signals";
+import type { Grant } from "../auth.ts";
+import { Producer as BroadcastProducer } from "../broadcast.ts";
+import { StreamCode, toStreamCode } from "../error.ts";
 import { Producer as GroupProducer } from "../group.ts";
 import { randomHop } from "../hop.ts";
 import { hooks } from "../internal.ts";
@@ -1641,3 +1644,94 @@ test("lite draft-05: a group that goes stale while its stream opens writes nothi
 		origin.close();
 	}
 });
+
+function grantOf(publish: string): Grant {
+	return { publish: new Path.Patterns([Path.Pattern.subtree(publish)]), subscribe: new Path.Patterns([]) };
+}
+
+// The resets carrying our UNAUTHORIZED code, by message.
+function unauthorizedResets(resets: { mock: { calls: unknown[][] } }): string[] {
+	return resets.mock.calls
+		.map(([reason]) => reason as Error)
+		.filter((reason) => toStreamCode(reason) === StreamCode.Unauthorized)
+		.map((reason) => reason.message);
+}
+
+// The grant watch is armed before the first check, so a shrink that lands while the broadcast
+// is still resolving resets the subscription instead of being missed for good.
+test("lite draft-06: a grant that shrinks while the broadcast resolves resets the subscription", async () => {
+	const pair = createMockTransportPair(ALPN_06);
+	const origin = new OriginProducer();
+	const handle = origin.dynamic(Path.from("live"));
+	const requests = handle.requested();
+	const grant = new Signal<Grant | undefined>(grantOf("live"));
+	const publisher = new Publisher(pair.server, Version.DRAFT_06, randomHop(), origin.consume(), grant);
+
+	const client = await Stream.open(pair.client);
+	const server = await Stream.accept(pair.server);
+	if (!server) throw new Error("publisher never accepted the subscribe stream");
+
+	const resets = spyOn(Writer.prototype, "reset");
+	const produced = new BroadcastProducer();
+	try {
+		const msg = new Subscribe({ id: 0n, broadcast: Path.from("live/cam"), track: "video", priority: 0 });
+		const serving = publisher.runSubscribe(msg, server);
+
+		// The subscription is parked resolving the broadcast when the grant shrinks.
+		const request = await requests.next();
+		if (request.done) throw new Error("the handler never saw the request");
+		grant.set(grantOf("other"));
+		await Promise.resolve();
+		request.value.accept(produced);
+		await serving;
+
+		expect(unauthorizedResets(resets)).toContain("unauthorized: live/cam");
+	} finally {
+		resets.mockRestore();
+		produced.close();
+		handle.close();
+		publisher.close();
+		client.close();
+		origin.close();
+	}
+});
+
+// A fetch holds its grant watch until the last frame: a group can stay open as long as its
+// track, so a check at accept alone would keep serving after a shrink.
+test("lite draft-06: a grant that shrinks mid-fetch resets the fetch", async () => {
+	const pair = createMockTransportPair(ALPN_06);
+	const origin = new OriginProducer();
+	const grant = new Signal<Grant | undefined>(grantOf("live"));
+	const publisher = new Publisher(pair.server, Version.DRAFT_06, randomHop(), origin.consume(), grant);
+
+	const broadcast = publish(origin, Path.from("live"));
+	const track = broadcast.createTrack("video");
+	// Left open, so the fetch is still serving when the grant shrinks.
+	const group = new GroupProducer(0);
+	group.writeString("first");
+	track.writeGroup(group);
+
+	const client = await Stream.open(pair.client);
+	const server = await Stream.accept(pair.server);
+	if (!server) throw new Error("publisher never accepted the fetch stream");
+
+	const resets = spyOn(Writer.prototype, "reset");
+	try {
+		const msg = new Fetch({ broadcast: Path.from("live"), track: "video", priority: 0, group: 0 });
+		const serving = publisher.runFetch(msg, server);
+		// The first frame is on the wire before the grant shrinks.
+		await client.reader.u8();
+
+		grant.set(grantOf("other"));
+		await serving;
+
+		expect(unauthorizedResets(resets)).toContain("unauthorized: live");
+	} finally {
+		resets.mockRestore();
+		group.close();
+		publisher.close();
+		client.close();
+		broadcast.close();
+		origin.close();
+	}
+});

js/net/src/lite/publisher.ts

@@ -574,10 +574,46 @@ export class Publisher {
 	 * @internal
 	 */
 	async runSubscribe(msg: Subscribe, stream: Stream) {
-		// Serve only what our grant lets us publish, and stop once it no longer does.
+		// Serve only what our grant lets us publish, and stop once it no longer does. The
+		// watch is armed before the first check and held to the end, so a shrink while the
+		// broadcast resolves is never missed.
+		const revoked = unauthorized(msg.broadcast);
+		let serving: track.Subscriber | undefined;
+		const watch = this.#watch(msg.broadcast, () => {
+			console.debug(`publish revoked: broadcast=${msg.broadcast} track=${msg.track}`);
+			serving?.close(revoked);
+			stream.abort(revoked);
+		});
+		try {
+			await this.#serveSubscribe(msg, stream, watch, (track) => {
+				serving = track;
+			});
+		} finally {
+			watch.dispose();
+		}
+	}
+
+	// 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,
+		watch: { revoked: () => boolean },
+		serving: (track: track.Subscriber) => void,
+	) {
 		// Checked before resolving, so a denied request never reaches the origin.
-		const denied = () => this.#denied(msg.broadcast);
-		if (denied()) {
+		if (watch.revoked()) {
 			stream.writer.reset(unauthorized(msg.broadcast));
 			return;
 		}
@@ -591,6 +627,8 @@ export class Publisher {
 			stream.writer.reset(error(err));
 			return;
 		}
+		// Revoked while the broadcast resolved: the watch already reset the stream.
+		if (watch.revoked()) return;
 		if (!front) {
 			console.debug(`publish unknown: broadcast=${msg.broadcast}`);
 			stream.writer.reset(new NotFound(`broadcast ${msg.broadcast}`));
@@ -606,6 +644,7 @@ export class Publisher {
 				end: endGroup === undefined ? undefined : { excluded: endGroup },
 			},
 		});
+		serving(track);
 		positionCursor(track, this.version, msg.startGroup);
 		hooks.replaceGroups(track, { end: endGroup === undefined ? undefined : { excluded: endGroup } });
 
@@ -615,14 +654,6 @@ export class Publisher {
 		let datagrams = Promise.resolve();
 		let controls: SubscriptionControls | undefined;
 
-		const revoked = unauthorized(msg.broadcast);
-		const disposeGrant = this.#grant?.subscribe(() => {
-			if (!denied()) return;
-			console.debug(`publish revoked: broadcast=${msg.broadcast} track=${track.name}`);
-			track.close(revoked);
-			stream.abort(revoked);
-		});
-
 		try {
 			let timescale: Timescale = Timescale.MILLI;
 
@@ -696,8 +727,6 @@ export class Publisher {
 			track.close(e);
 			stream.abort(e);
 			await Promise.all([datagrams, controls?.decoding]);
-		} finally {
-			disposeGrant?.();
 		}
 	}
 
@@ -711,7 +740,31 @@ export class Publisher {
 			stream.writer.reset(new Error("fetch requires moq-lite-05 or newer"));
 			return;
 		}
-		if (this.#denied(msg.broadcast)) {
+		// Like a subscription, the fetch holds its watch until the last frame: a group can stay
+		// 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, () => {
+			console.debug(`fetch revoked: broadcast=${msg.broadcast} track=${msg.track} group=${msg.group}`);
+			fetched?.close(revoked);
+			stream.abort(revoked);
+		});
+		try {
+			await this.#serveFetch(msg, stream, watch, (group) => {
+				fetched = group;
+			});
+		} finally {
+			watch.dispose();
+		}
+	}
+
+	async #serveFetch(
+		msg: Fetch,
+		stream: Stream,
+		watch: { revoked: () => boolean },
+		serving: (group: group.Consumer) => void,
+	) {
+		if (watch.revoked()) {
 			stream.writer.reset(unauthorized(msg.broadcast));
 			return;
 		}
@@ -730,6 +783,8 @@ export class Publisher {
 			stream.writer.reset(new NotFound(`broadcast ${msg.broadcast}`));
 			return;
 		}
+		// Revoked while the broadcast resolved: the watch already reset the stream.
+		if (watch.revoked()) return;
 
 		// The subscriber opened this stream, so its send order only ranked the request. Rank the
 		// response here, on the same scale as the group streams it competes with.
@@ -741,6 +796,12 @@ export class Publisher {
 			// come off the same front, so the metadata and the frames are one generation.
 			const info = await this.#resolveTrackInfo(front, msg.track);
 			group = await wireOf(front).fetchGroup(msg.track, msg.group, { priority: msg.priority });
+			serving(group);
+			// Revoked while the group resolved: the watch already reset the stream.
+			if (watch.revoked()) {
+				group.close(unauthorized(msg.broadcast));
+				return;
+			}
 			await this.#runFetchGroup(group, stream.writer, {
 				timescale: Timescale(info.timescale),
 				start: msg.startFrame,

js/net/src/lite/subscriber.test.ts

@@ -1,5 +1,6 @@
 import { expect, spyOn, test } from "bun:test";
 import { Signal } from "@moq/signals";
+import type { Grant } from "../auth.ts";
 import type { Probe as ProbeStats } from "../connection/stats.ts";
 import { error, reason, StreamCode, StreamError } from "../error.ts";
 import { HopSchema, isAnonymous, MAX_HOPS, Route, UNKNOWN_HOP } from "../hop.ts";
@@ -803,3 +804,33 @@ test("a fetch started after the subscriber closes rejects without opening a stre
 	expectCut(err, undefined);
 	expect(streams.length).toBe(0);
 });
+
+// The grant watch is armed before the first check, so a shrink while TRACK_INFO is still in
+// flight refuses the subscription rather than opening a SUBSCRIBE the grant no longer covers.
+test("a grant that shrinks while the subscription sets up refuses it", async () => {
+	const { quic, streams } = fakeSession();
+	const scoped = (prefix: string): Grant => ({
+		publish: new Path.Patterns([]),
+		subscribe: new Path.Patterns([Path.Pattern.subtree(prefix)]),
+	});
+	const grant = new Signal<Grant | undefined>(scoped("room"));
+	const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n), undefined, undefined, grant);
+
+	const track = subscriber.consume(Path.from("room/cam")).track("video").subscribe().ordered();
+	const next = track.nextGroup().catch((err: unknown) => err);
+
+	// Parked on TRACK_INFO when the grant shrinks.
+	await drainUntil(() => streams.length === 1);
+	await streams[0].reading;
+	grant.set(scoped("other"));
+	await answerTrackInfo(streams[0]);
+
+	const err = await next;
+	expect(err).toBeInstanceOf(StreamError);
+	expect((err as StreamError).code).toBe(StreamCode.Unauthorized);
+	// Nothing past the TRACK stream reached the wire.
+	expect(streams.length).toBe(1);
+
+	track.close();
+	subscriber.close();
+});