feat(auth): a request token is one more auth on requests()

doc/bin/relay/auth.md

@@ -143,7 +143,11 @@ if its first credential expired earlier, after its attached grants stop. Each
 attached JWT counts against its own `--limit-token` cap until the connection
 ends. The initial credential remains responsible for the connection's stats tier.
 Certificate sessions cannot renew a certificate in band, and public rules
-cannot verify a presented token; both refuse it as unsupported. A burst of
+cannot verify a presented token; both refuse it as unsupported. A moq-transport
+`AUTHORIZATION TOKEN` on a `SUBSCRIBE` or `PUBLISH_NAMESPACE` is verified the
+same way on any draft, but its grant authorizes only that request and ends with
+it; it never joins the union. A `PUBLISH_NAMESPACE` grant must cover the whole
+namespace: `room/**`, not just `room`. A burst of
 tokens past the session's pending-verification bound is refused. On moq-lite, an
 auth server that cannot answer, by timeout or outage, refuses a token as a
 transient error, and `Connection::auth()` presents it again on its reconnect

doc/concept/standard.md

@@ -29,8 +29,8 @@ subset drops. What a peer actually observes against this implementation:
 - **History is one group.** In Rust, Object ID gaps must be marked by Prior Object ID Gap properties for a group to be fetchable. A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed, and so does a reader whose publisher refuses `TRACK_STATUS` as `NOT_SUPPORTED`.
 - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed, as is every track on drafts 14 through 16. A Rust reader that never subscribed takes the units from `TRACK_STATUS` on draft 18 and later.
 - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. Message parameters follow the negotiated draft's lists: drafts 14 and 15 ignore an unknown or misplaced parameter, draft 16 ignores one defined only for another message but closes the session on an unknown one, and draft 17 and later close it on both.
-- **One `SETUP` credential**, forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second `SETUP` token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. A token attached to an individual request is otherwise decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` is refused unread, so its token is not decoded. On draft 17 and later, the negotiated [MoQ Auth extension](/concept/moq-lite#authorization) accepts additional session tokens in band.
-- **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, range filters, `SUBSCRIBE_TRACKS`, and the fetch forms above. On draft 19 and later, and in Rust on drafts 14 through 16, a subscription update may change only priority; any other update ends that subscription.
+- **One `SETUP` credential**, forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second `SETUP` token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. Where the Rust side authorizes the peer or verifies tokens, a token on a `SUBSCRIBE` or `PUBLISH_NAMESPACE` is verified like an in-band session token and authorizes that one request alone, even where that authorization already covers it; it never widens the session. It is refused with `NOT_SUPPORTED` when nothing verifies it, when it is empty, or when its Token Type is anything but 0 (out of band). A side that does neither, JavaScript included, ignores the token. Rust refuses a request carrying more than one distinct token with `NOT_SUPPORTED`, since it verifies one; JavaScript, which verifies none, ignores them all. A token-authorized `SUBSCRIBE` still stays within the grant the peer gave us. A token-authorized `PUBLISH_NAMESPACE` needs a grant covering everything beneath the namespace, and the tracks we subscribe under it rely on that announce instead of our limit, ending with it. An update renews the token, except on a drafts 14 and 15 `PUBLISH_NAMESPACE`, which has no update, and a refused renewal ends only that request. A token on any other request is decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` is refused unread, so its token is not decoded. On draft 17 and later, the negotiated [MoQ Auth extension](/concept/moq-lite#authorization) accepts additional session tokens in band.
+- **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, range filters, `SUBSCRIBE_TRACKS`, and the fetch forms above. On draft 19 and later, and in Rust on drafts 14 through 16, a subscription update may change only priority or renew a request token; any other update ends that subscription.
 - **Objects and datagrams.** Rust relays preserve Object IDs, gaps, object properties, FIRST\_OBJECT, and END\_OF\_GROUP status objects between moq-transport peers for subgroup 0. Datagrams retain their IDs, properties, priority, and status. Plain Rust readers and moq-lite peers see consecutive frames and normal datagram payloads; JavaScript supports only normal datagrams at Object ID 0. A malformed datagram closes the session; a relayed one the downstream draft cannot carry is dropped. A relayed Timestamp leaves in the units the downstream response declares, and is dropped where it declares none (drafts 14 through 16), as for a plain frame; an immutable one that would change or vanish is refused. Property 0x06 is a draft-ietf-moq-loc-03 Timestamp only before draft 19, which reassigns it to SUBGROUP\_DELIVERY\_TIMEOUT, so it is dropped across that boundary, or refused when immutable. From draft 15, a status object carrying properties is refused, and one relayed from draft 14 leaves without them. A `FETCH` carries the END\_OF\_GROUP object on drafts 14 and 15; later drafts have no FETCH status, so their response ends at the last frame. A group cannot resume across route changes by frame ordinal once Object IDs skip ahead of the ordinals on either route; recovery refuses it instead of splicing different objects.
 - **Priority.** Higher is served first. The IETF default of 128 is this stack's 127, and a track that never sets one is 127.
 - **Size.** An object extension block larger than 64 KiB ends that subgroup stream. The session stays up. This cap is ours, not the draft's.

drafts/draft-lcurley-moq-lite.md

@@ -301,6 +301,8 @@ Sent when terminating the session, via the transport's session close.
 | ------- | ------------- | ----------- |
 |  0x17  | UNKNOWN_AUTH_TOKEN_ALIAS | A token named an alias that was never registered. |
 | ------- | ------------- | ----------- |
+|  0x1B  | TOO_MANY_REQUEST_UPDATES | The peer left more request updates unanswered than the endpoint allows. |
+| ------- | ------------- | ----------- |
 |  0x30  | NOT_SUPPORTED | The requested feature, token, or grant is unsupported. |
 | ------- | ------------- | ----------- |
 
@@ -1590,7 +1592,7 @@ The `Message Length` describes the payload size on the wire.
 - Added announce compression: ANNOUNCE_START gains `Path Base` and `Path Keep` to copy the head of a live advertisement's suffix, and ANNOUNCE_START and ANNOUNCE_UPDATE gain `Hop Base` and `Hop Keep` to copy the tail of a live advertisement's Hop ID list.
 - Capped the Message Length of every message except FRAME at 65,535 bytes.
 - Added the TOO_MANY_REQUESTS (0x7) session code, closing a session whose peer goes past the endpoint's bound on subscriptions or announcements.
-- Added the AUTH_TOKEN_CACHE_OVERFLOW (0x13) and UNKNOWN_AUTH_TOKEN_ALIAS (0x17) session codes, with moq-transport's meaning.
+- Added the AUTH_TOKEN_CACHE_OVERFLOW (0x13), UNKNOWN_AUTH_TOKEN_ALIAS (0x17), and TOO_MANY_REQUEST_UPDATES (0x1B) session codes, with moq-transport's meaning.
 - Limited AUTH_OK Expires to 2^53 - 1 milliseconds, with an out-of-range value closing the session with PROTOCOL_VIOLATION.
 - Invalid UTF-8 or an oversized AUTH_ERROR reason closes the session with PROTOCOL_VIOLATION.
 - An Auth Stream message that fails to decode, has trailing body bytes, or has an unexpected type, or a second AUTH from the opener, closes the session with PROTOCOL_VIOLATION.

js/net/src/auth.test.ts

@@ -61,46 +61,6 @@ async function connect(opts: { publish?: OriginProducer; serverPublish?: OriginP
 
 // Every case runs on moq-lite-07-wip and on moq-transport with the MoQ Auth extension.
 describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", (protocol) => {
-	test("pending application tokens have a per-session bound", async () => {
-		const { client, server } = await connect({ publish: new OriginProducer(), protocol });
-		try {
-			const requests = server.auth.requests();
-			const setup = await requests.next();
-			if (!setup) throw new Error("setup token missing");
-			setup.accept(grant(["a"], []));
-			let refused!: () => void;
-			const overflow = new Promise<void>((resolve) => {
-				refused = resolve;
-			});
-			const pending = Array.from({ length: 129 }, (_, index) =>
-				client.auth.add(`token-${index}`).then(
-					(token) => ({ token }),
-					(err: unknown) => {
-						refused();
-						return { err };
-					},
-				),
-			);
-			await withTimeout(overflow, 1000, "pending token bound did not refuse overflow");
-			for (let index = 0; index < 128; index++) {
-				const request = await requests.next();
-				if (!request) throw new Error("pending token missing");
-				request.accept(grant(["a"], []));
-			}
-			const results = await Promise.all(pending);
-			const errors = results.flatMap((result) => ("err" in result ? [result.err] : []));
-			expect(errors).toHaveLength(1);
-			// moq-transport's AUTH_ERROR registry only distinguishes unsupported from unauthorized.
-			expect((errors[0] as SessionError).code).toBe(
-				protocol === Lite.ALPN_07_WIP ? SessionCode.TooManyRequests : SessionCode.Unauthorized,
-			);
-			for (const result of results) if ("token" in result) result.token.close();
-		} finally {
-			client.abort();
-			server.abort();
-		}
-	});
-
 	test("both sides learn their default grant", async () => {
 		const { client, server } = await connect({ publish: new OriginProducer(), protocol });
 
@@ -497,7 +457,7 @@ test("withdrawn queued tokens are pruned before the next application decision",
 		if (!setup) throw new Error("setup token missing");
 		setup.accept(grant(["a"], []));
 		const wire = new LiteAuthWire(transport, Lite.Version.DRAFT_07);
-		for (let index = 0; index < 128; index++) {
+		for (let index = 0; index < 3; index++) {
 			const stream = await wire.present(new TextEncoder().encode("withdrawn"));
 			await stream.writer.close();
 			await stream.reader.closed;

js/net/src/auth.ts

@@ -111,6 +111,9 @@ export interface Auth {
 	 *
 	 * Call it as soon as connect or accept resolves, in the same task: whoever answers is
 	 * decided once, right after. Throws once that has happened.
+	 *
+	 * The peer's stream credit bounds how many tokens wait unanswered, since each holds its
+	 * stream open until answered or withdrawn. A withdrawn token leaves the queue.
 	 */
 	requests(): Requests;
 }

js/net/src/auth_session.ts

@@ -213,19 +213,23 @@ class IssuedGrant implements Issued {
 /** The peer's tokens, queued for the application. */
 class RequestQueue implements Requests {
 	#queue: PeerRequest[] = [];
+	// How many requests survived the last prune: pruning again only once the queue doubles
+	// keeps a push amortized O(1).
+	#survivors = 0;
 	#waiters: ((request: PeerRequest | undefined) => void)[] = [];
 	#closed = false;
 
-	/** Queue the token for the application, or refuse it when closed or full. */
+	/** Queue the token for the application, or refuse it when closed. */
 	push(request: PeerRequest): void {
 		if (this.#closed) {
 			request.reject(SessionCode.Unauthorized, "not accepting tokens");
 			return;
 		}
-		if (this.#queue.length >= 128) this.#queue = this.#queue.filter((queued) => !queued.ended);
-		if (this.#queue.length >= 128) {
-			request.reject(SessionCode.TooManyRequests, "too many pending tokens");
-			return;
+		// A withdrawn token gave its stream back, so it must not stay queued while the
+		// application is busy: that would let a peer that keeps withdrawing grow the queue.
+		if (this.#queue.length >= 2 * Math.max(this.#survivors, 1)) {
+			this.#queue = this.#queue.filter((queued) => !queued.ended);
+			this.#survivors = this.#queue.length;
 		}
 		const waiter = this.#waiters.shift();
 		if (waiter) waiter(request);

js/net/src/error.test.ts

@@ -217,6 +217,7 @@ test("the code tables match the spec", () => {
 	expect(Number(SessionCode.AuthTokenCacheOverflow)).toBe(0x13);
 	expect(Number(SessionCode.Version)).toBe(0x15);
 	expect(Number(SessionCode.UnknownAuthTokenAlias)).toBe(0x17);
+	expect(Number(SessionCode.TooManyRequestUpdates)).toBe(0x1b);
 	expect(Number(StreamCode.Cancel)).toBe(0x1);
 	expect(Number(StreamCode.SessionClosed)).toBe(0x3);
 	expect(Number(StreamCode.TooFarBehind)).toBe(0x5);

js/net/src/error.ts

@@ -54,6 +54,8 @@ export const SessionCode = Object.freeze(
 		UnknownAuthTokenAlias: 0x17 as SessionCode,
 		/** The peer cannot support the requested feature; moq-lite only. */
 		Unsupported: 0x30 as SessionCode,
+		/** The peer left more request updates unanswered than MAX_REQUEST_UPDATES allows. */
+		TooManyRequestUpdates: 0x1b as SessionCode,
 	} as const),
 );
 

js/net/src/ietf/cluster.test.ts

@@ -283,12 +283,27 @@ test("Cluster: every malformed PUBLISH_NAMESPACE update is a protocol violation"
 		await expect(PublishNamespaceUpdate.decode(reader(bytes), VERSION)).rejects.toThrow(ProtocolViolation);
 	}
 
-	// The message did not exist before draft-17.
+	// The message did not exist before draft-16.
 	await expect(
-		PublishNamespaceUpdate.decode(reader(new Uint8Array([0, 0]), Version.DRAFT_16), Version.DRAFT_16),
+		PublishNamespaceUpdate.decode(reader(new Uint8Array([0, 0]), Version.DRAFT_15), Version.DRAFT_15),
 	).rejects.toThrow(ProtocolViolation);
 });
 
+test("Cluster: a draft-16 PUBLISH_NAMESPACE update names the advertisement", async () => {
+	const version = Version.DRAFT_16;
+	const update = new PublishNamespaceUpdate({ requestId: 2n, namespaceRequestId: 0n, update: {} });
+	const decoded = await PublishNamespaceUpdate.decode(reader(await encode(update, version), version), version);
+	expect(decoded.requestId).toBe(2n);
+	expect(decoded.namespaceRequestId).toBe(0n);
+
+	// A repeated token refuses the update, keyed by its own Request ID.
+	const tokens = [2, 0x03, 3, 0x03, 0x00, 0xaa, 0, 3, 0x03, 0x00, 0xaa];
+	const body = [2, 0, ...tokens];
+	const repeated = PublishNamespaceUpdate.decode(reader(new Uint8Array([0, body.length, ...body]), version), version);
+	await expect(repeated).rejects.toThrow(MalformedAuthToken);
+	await expect(repeated).rejects.toMatchObject({ requestId: 2n });
+});
+
 test("Cluster: a reset while decoding an update stays stream-local", async () => {
 	const reset = Object.assign(new Error("reset"), {
 		source: "stream" as const,

js/net/src/ietf/publish_namespace.ts

@@ -84,16 +84,33 @@ export class PublishNamespaceUpdate {
 
 	/** The update's own Request ID; every REQUEST_UPDATE consumes one. */
 	requestId: bigint;
+	/** Draft-16 only: the PUBLISH_NAMESPACE this updates, which later drafts name by stream. */
+	namespaceRequestId?: bigint;
 	update: Cluster.Update;
 
-	constructor({ requestId, update }: { requestId: bigint; update: Cluster.Update }) {
+	constructor({
+		requestId,
+		namespaceRequestId,
+		update,
+	}: {
+		requestId: bigint;
+		namespaceRequestId?: bigint;
+		update: Cluster.Update;
+	}) {
 		this.requestId = requestId;
+		this.namespaceRequestId = namespaceRequestId;
 		this.update = update;
 	}
 
 	async #encode(w: Writer, version: IetfVersion): Promise<void> {
-		PublishNamespaceUpdate.#modern(version);
+		PublishNamespaceUpdate.#supported(version);
 		await w.u62(this.requestId);
+		if (version === Version.DRAFT_16) {
+			if (this.namespaceRequestId === undefined) {
+				throw new Error("a draft-16 REQUEST_UPDATE names its PUBLISH_NAMESPACE");
+			}
+			await w.u62(this.namespaceRequestId);
+		}
 		if (version === Version.DRAFT_17) {
 			await w.u62(0n); // required_request_id_delta (draft-17 only, removed in draft-18 per #1615)
 		}
@@ -128,20 +145,21 @@ export class PublishNamespaceUpdate {
 		}
 	}
 
-	static #modern(version: IetfVersion) {
-		if (version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16) {
-			throw new Error("REQUEST_UPDATE on PUBLISH_NAMESPACE requires draft-17+");
+	static #supported(version: IetfVersion) {
+		if (version === Version.DRAFT_14 || version === Version.DRAFT_15) {
+			throw new Error("REQUEST_UPDATE on PUBLISH_NAMESPACE requires draft-16+");
 		}
 	}
 
 	static async #decode(r: Reader, version: IetfVersion): Promise<PublishNamespaceUpdate> {
-		PublishNamespaceUpdate.#modern(version);
+		PublishNamespaceUpdate.#supported(version);
 		const requestId = await r.u62();
+		const namespaceRequestId = version === Version.DRAFT_16 ? await r.u62() : undefined;
 		if (version === Version.DRAFT_17) {
 			await r.u62(); // required_request_id_delta (draft-17 only, removed in draft-18 per #1615)
 		}
 		const params = await Parameters.decode(r, version, "publish-namespace");
-		return new PublishNamespaceUpdate({ requestId, update: Cluster.updateFromParams(params) });
+		return new PublishNamespaceUpdate({ requestId, namespaceRequestId, update: Cluster.updateFromParams(params) });
 	}
 }