diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 05d2648c70..03b1eadd3a 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -29,7 +29,7 @@ subset drops. What a peer actually observes against this implementation: - **History is one group.** A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed. - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed, as is every track on drafts 14 through 16. A Rust reader that never subscribed takes the units from `TRACK_STATUS` on draft 18 and later. - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. Message parameters follow the negotiated draft's lists: drafts 14 and 15 ignore an unknown or misplaced parameter, draft 16 ignores one defined only for another message but closes the session on an unknown one, and draft 17 and later close it on both. -- **One credential per session**, carried in `SETUP` and forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A token attached to an individual request is ignored. An alias reference or a second token closes the session. +- **One credential per session**, carried in `SETUP` and forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. A token attached to an individual request is otherwise decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. `SUBSCRIBE_TRACKS` is refused unread, so its token is not decoded. - **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, range filters, `TRACK_STATUS` to a JavaScript publisher, `SUBSCRIBE_TRACKS`, and the fetch forms above. On draft 19 and later, and in Rust on drafts 14 through 16, a subscription update may change only priority; any other update ends that subscription. - **Datagrams** are a single normal object at object 0, forwarded without renumbering. Anything else is dropped, and a malformed one closes the session. Rust and JavaScript both carry them on every draft. - **Priority.** Higher is served first. The IETF default of 128 is this stack's 127, and a track that never sets one is 127. @@ -52,7 +52,7 @@ See [Encryption](/concept/hang#encryption). ### Deliberate deviations -These three answers differ from the draft on purpose. They are the +These four answers differ from the draft on purpose. They are the product's model, not bugs, and the relay does not change them. - **One publisher per path.** A broadcast path names one piece of content, so a @@ -76,6 +76,11 @@ product's model, not bugs, and the relay does not change them. requires an established upstream subscription before `SUBSCRIBE_OK`. From moq-lite 05 the track stream answers first, and a missing track is refused before `SUBSCRIBE_OK`. +- **An unknown token alias closes the session.** No alias is ever registered, + so a `DELETE` or `USE_ALIAS` token on a request names one that does not exist. + [Draft 21 ยง8.9](https://www.ietf.org/archive/id/draft-ietf-moq-transport-21.html#section-8.9) + rejects the message, but `UNKNOWN_AUTH_TOKEN_ALIAS` exists only as a session + termination code, so we close the session with it. ## MSF diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 9d794c17fa..639078a941 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -292,8 +292,12 @@ Sent when terminating the session, via the transport's session close. | ------- | ------------- | ----------- | | 0x11 | CONTROL_MESSAGE_TIMEOUT | The peer took too long to respond to a control message. | | ------- | ------------- | ----------- | +| 0x13 | AUTH_TOKEN_CACHE_OVERFLOW | A token registration would exceed the advertised token cache size. | +| ------- | ------------- | ----------- | | 0x15 | VERSION_NEGOTIATION_FAILED | No version could be negotiated. | | ------- | ------------- | ----------- | +| 0x17 | UNKNOWN_AUTH_TOKEN_ALIAS | A token named an alias that was never registered. | +| ------- | ------------- | ----------- | ### Stream Error Codes Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (STOP_SENDING). @@ -1552,6 +1556,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. - 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. - Added the Auth Stream (0x7) with AUTH, AUTH_OK, and AUTH_ERROR: either endpoint presents a token on its own stream and learns the [path patterns](#path-pattern) it may publish and subscribe to. The union of a session's open grants is its scope. A shrink withdraws what it no longer covers, and an announcement outside the scope closes the session with UNAUTHORIZED. A peer without the stream resets it. diff --git a/js/net/src/error.test.ts b/js/net/src/error.test.ts index c6d5756c2a..dc951463c8 100644 --- a/js/net/src/error.test.ts +++ b/js/net/src/error.test.ts @@ -209,7 +209,9 @@ test("the code tables match the spec", () => { expect(Number(SessionCode.Cancel)).toBe(0x0); expect(Number(SessionCode.Unauthorized)).toBe(0x2); expect(Number(SessionCode.GoawayTimeout)).toBe(0x10); + expect(Number(SessionCode.AuthTokenCacheOverflow)).toBe(0x13); expect(Number(SessionCode.Version)).toBe(0x15); + expect(Number(SessionCode.UnknownAuthTokenAlias)).toBe(0x17); expect(Number(StreamCode.Cancel)).toBe(0x1); expect(Number(StreamCode.SessionClosed)).toBe(0x3); expect(Number(StreamCode.TooFarBehind)).toBe(0x5); diff --git a/js/net/src/error.ts b/js/net/src/error.ts index c72b5236df..963538a8fe 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -44,8 +44,12 @@ export const SessionCode = Object.freeze( GoawayTimeout: 0x10 as SessionCode, /** A control message took too long. */ Timeout: 0x11 as SessionCode, + /** A token registration would exceed the advertised token cache size, which is 0 when unadvertised. */ + AuthTokenCacheOverflow: 0x13 as SessionCode, /** No version could be negotiated. */ Version: 0x15 as SessionCode, + /** A token named an alias that was never registered. */ + UnknownAuthTokenAlias: 0x17 as SessionCode, } as const), ); @@ -296,9 +300,13 @@ export class RefusedRedirect extends Error { * @public */ export class ProtocolViolation extends Error { - constructor(message: string, options?: { cause?: unknown }) { + /** The session code to close with: `ProtocolViolation` unless the draft names a more specific one. */ + readonly code: SessionCode; + + constructor(message: string, options?: { cause?: unknown; code?: SessionCode }) { super(message, options); this.name = "ProtocolViolation"; + this.code = options?.code ?? SessionCode.ProtocolViolation; } } diff --git a/js/net/src/ietf/cluster.test.ts b/js/net/src/ietf/cluster.test.ts index a4c2626c46..86082fc7e5 100644 --- a/js/net/src/ietf/cluster.test.ts +++ b/js/net/src/ietf/cluster.test.ts @@ -9,6 +9,7 @@ import * as Namespace from "./namespace.ts"; import { Parameters, SetupOption, SetupOptions } from "./parameters.ts"; import { PublishNamespace, PublishNamespaceUpdate } from "./publish_namespace.ts"; import { SubscribeNamespaceEntry } from "./subscribe_namespace.ts"; +import { MalformedAuthToken } from "./token.ts"; import { type IetfVersion, Version } from "./version.ts"; const VERSION = Version.DRAFT_19; @@ -241,7 +242,7 @@ test("Cluster: a negotiated PUBLISH_NAMESPACE may carry AUTHORIZATION_TOKEN", as await wr.u62(1n); await Namespace.encode(wr, Path.from("alice.hang")); const params = Cluster.intoParams(advert); - params.bytes.set(0x03n, new Uint8Array([1, 2, 3])); // AUTHORIZATION_TOKEN + params.bytes.set(0x03n, new Uint8Array([0x03, 0x00, 0xaa])); // AUTHORIZATION_TOKEN: USE_VALUE, type 0 await params.encode(wr, version); }); }, @@ -251,6 +252,20 @@ test("Cluster: a negotiated PUBLISH_NAMESPACE may carry AUTHORIZATION_TOKEN", as expect(decoded.cluster).toEqual(advert); }); +test("Cluster: a repeated AUTHORIZATION_TOKEN refuses only the advertisement", async () => { + // USE_VALUE, type 0, value 0xAA, twice. A repeated key codes as a delta of 0. + const tokens = [2, 0x03, 3, 0x03, 0x00, 0xaa, 0, 3, 0x03, 0x00, 0xaa]; + const frame = (body: number[]) => new Uint8Array([0, body.length, ...body]); + + // Request ID 1 for namespace "a". The repeat is refused before the missing HOP_PATH is. + const advert = PublishNamespace.decode(reader(frame([1, 1, 1, 0x61, ...tokens])), VERSION, true); + await expect(advert).rejects.toThrow(MalformedAuthToken); + await expect(advert).rejects.toMatchObject({ requestId: 1n }); + + const update = PublishNamespaceUpdate.decode(reader(frame([1, ...tokens])), VERSION); + await expect(update).rejects.toThrow(MalformedAuthToken); +}); + test("Cluster: every malformed PUBLISH_NAMESPACE update is a protocol violation", async () => { const valid = await encode(new PublishNamespaceUpdate({ requestId: 1n, update: { cost: 2n } })); const trailing = new Uint8Array(valid.byteLength + 1); diff --git a/js/net/src/ietf/cluster.ts b/js/net/src/ietf/cluster.ts index 0af006eb97..3d9e400a26 100644 --- a/js/net/src/ietf/cluster.ts +++ b/js/net/src/ietf/cluster.ts @@ -29,6 +29,7 @@ import { type Hop, HopSchema, MAX_HOPS, UNKNOWN_HOP } from "../hop.ts"; import type { Reader } from "../stream.ts"; import * as Varint from "../varint.ts"; import { Parameters, SetupOption, type SetupOptions } from "./parameters.ts"; +import { MalformedAuthToken } from "./token.ts"; import { type IetfVersion, Version } from "./version.ts"; /** @@ -222,6 +223,8 @@ export async function decodeParams( try { params = await Parameters.decode(r, version, message); } catch (err) { + // Keep a violation's own code, such as a request token's, and a refusal of the request alone. + if (err instanceof ProtocolViolation || err instanceof MalformedAuthToken) throw err; throw new ProtocolViolation(reason(err), { cause: err }); } diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 6707e79ca1..4bdbbfc9b5 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -6,6 +6,7 @@ import { createMockTransportPair } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; import { Stream, Writer } from "../stream.ts"; +import { Timescale } from "../time.ts"; import { Connection } from "./connection.ts"; import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; @@ -13,6 +14,7 @@ import { Group } from "./object.ts"; import { Parameters, SetupOption, SetupOptions } from "./parameters.ts"; import { initialMaxRequestId, MaxRequestId, RequestError, RequestOk } from "./request.ts"; import { Setup } from "./setup.ts"; +import { SubscribeError, SubscribeOk } from "./subscribe.ts"; import { SUBSCRIBE_TRACKS_ID, SubscribeNamespace, @@ -498,3 +500,241 @@ for (const [version, alpn] of [ }); } } + +/** + * A parameter block of AUTHORIZATION TOKENs. Draft-16 on codes each key as a delta from the + * one before, so a repeat is 0. + */ +function tokenParams(version: IetfVersion, ...tokens: (readonly number[])[]): number[] { + const delta = version !== Version.DRAFT_14 && version !== Version.DRAFT_15; + return [tokens.length, ...tokens.flatMap((token, i) => [i > 0 && delta ? 0 : 0x03, token.length, ...token])]; +} + +/** A SUBSCRIBE for `room/video` whose parameter block is `params`. */ +function subscribeWith(version: IetfVersion, params: readonly number[], requestId = 0): Uint8Array { + const ascii = (text: string) => [...new TextEncoder().encode(text)]; + // Request ID, then the namespace and track name. + const head = [requestId, 1, 4, ...ascii("room"), 5, ...ascii("video")]; + // Draft-14 carries priority, group order, forward, and a Largest Object filter inline. + const fields = version === Version.DRAFT_14 ? [0x80, 0x02, 0x01, 0x02] : []; + const body = [...head, ...fields, ...params]; + return new Uint8Array([0x03, 0, body.length, ...body]); +} + +/** A SUBSCRIBE for `room/video` whose only parameter is an AUTHORIZATION TOKEN holding `token`. */ +function subscribeWithToken(version: IetfVersion, token: readonly number[]): Uint8Array { + return subscribeWith(version, tokenParams(version, token)); +} + +/** An update to that SUBSCRIBE whose only parameters are AUTHORIZATION TOKENs holding `tokens`. */ +function updateWithTokens(version: IetfVersion, ...tokens: (readonly number[])[]): Uint8Array { + // Draft-14 names its own Request ID and the subscription's, then the range, priority, + // and forward inline. Drafts 15 and 16 name both IDs, and later drafts only the request + // it updates. + const head = + version === Version.DRAFT_14 + ? [2, 0, 0, 0, 0, 0x80, 0x01] + : version === Version.DRAFT_15 || version === Version.DRAFT_16 + ? [2, 0] + : [0]; + const body = [...head, ...tokenParams(version, ...tokens)]; + return new Uint8Array([0x02, 0, body.length, ...body]); +} + +/** + * A request's AUTHORIZATION TOKEN decodes by the SETUP option's rules, so one this endpoint + * cannot use closes the session with the draft's code. Draft-14 carries the request on the + * control stream, a strict draft on its own stream. + */ +for (const [version, alpn] of [ + [Version.DRAFT_14, ALPN.DRAFT_14], + [Version.DRAFT_22, ALPN.DRAFT_22], +] as const) { + for (const [name, token, code] of [ + ["REGISTER", [0x01, 0x07, 0x00], SessionCode.AuthTokenCacheOverflow], + ["DELETE", [0x00, 0x07], SessionCode.UnknownAuthTokenAlias], + ["USE_ALIAS", [0x02, 0x07], SessionCode.UnknownAuthTokenAlias], + ["USE_VALUE without a type", [0x03], SessionCode.KeyValueFormatting], + ["an unknown alias type", [0x04, 0x00], SessionCode.KeyValueFormatting], + ] as const) { + test(`a request token with ${name} closes ${alpn}`, async () => { + const logged = spyOn(console, "error").mockImplementation(() => void 0); + const pair = createMockTransportPair(alpn); + const control = await Stream.open(pair.server, { version }); + const connection = new Connection({ + url: new URL("https://example.com"), + quic: pair.server, + control, + maxRequestId: 100n, + version, + client: false, + }); + try { + const stream = + version === Version.DRAFT_14 + ? await Stream.accept(pair.client, version) + : await Stream.open(pair.client, { version }); + if (!stream) throw new Error("no control stream"); + await stream.writer.write(subscribeWithToken(version, token)); + + const info = await pair.client.closed; + expect(info.closeCode).toBe(code); + } finally { + logged.mockRestore(); + connection.abort(); + } + }); + } +} + +/** + * An update's AUTHORIZATION TOKEN decodes like the request's, so one this endpoint cannot use + * closes the session with the draft's code, not only the subscription it updates. + */ +for (const [version, alpn] of [ + [Version.DRAFT_14, ALPN.DRAFT_14], + [Version.DRAFT_22, ALPN.DRAFT_22], +] as const) { + for (const [name, token, code] of [ + ["REGISTER", [0x01, 0x07, 0x00], SessionCode.AuthTokenCacheOverflow], + ["DELETE", [0x00, 0x07], SessionCode.UnknownAuthTokenAlias], + ["USE_ALIAS", [0x02, 0x07], SessionCode.UnknownAuthTokenAlias], + ["USE_VALUE without a type", [0x03], SessionCode.KeyValueFormatting], + ] as const) { + test(`an update token with ${name} closes ${alpn}`, async () => { + const logged = spyOn(console, "error").mockImplementation(() => void 0); + const pair = createMockTransportPair(alpn); + const control = await Stream.open(pair.server, { version }); + const origin = new OriginProducer(); + const broadcast = origin.createBroadcast(Path.from("room")); + broadcast.announce(); + broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const connection = new Connection({ + url: new URL("https://example.com"), + quic: pair.server, + control, + maxRequestId: 100n, + version, + client: false, + publish: origin.consume(), + }); + try { + const stream = + version === Version.DRAFT_14 + ? await Stream.accept(pair.client, version) + : await Stream.open(pair.client, { version }); + if (!stream) throw new Error("no control stream"); + await stream.writer.write(subscribeWith(version, [0])); + // Draft-14's control stream also carries the PUBLISH_NAMESPACE for `room`. + for (;;) { + const type = await stream.reader.u53(); + if (type === SubscribeOk.id) break; + await stream.reader.read(await stream.reader.u16()); + } + await SubscribeOk.decode(stream.reader, version); + await stream.writer.write(updateWithTokens(version, token)); + + const info = await pair.client.closed; + expect(info.closeCode).toBe(code); + } finally { + logged.mockRestore(); + connection.abort(); + origin.close(); + } + }); + } +} + +/** Skip the messages ahead of the first `id`, such as draft-14's PUBLISH_NAMESPACE. */ +async function skipTo(reader: Stream["reader"], id: number) { + for (;;) { + if ((await reader.u53()) === id) return; + await reader.read(await reader.u16()); + } +} + +/** A server publishing `room/video`, and the client's stream to send requests on. */ +async function serving(version: IetfVersion, alpn: string) { + const pair = createMockTransportPair(alpn); + const control = await Stream.open(pair.server, { version }); + const origin = new OriginProducer(); + const broadcast = origin.createBroadcast(Path.from("room")); + broadcast.announce(); + broadcast.createTrack("video", { timescale: Timescale.MILLI }); + const connection = new Connection({ + url: new URL("https://example.com"), + quic: pair.server, + control, + maxRequestId: 100n, + version, + client: false, + publish: origin.consume(), + }); + // Through draft 16 every request rides the control stream. + const control16 = version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16; + const stream = control16 ? await Stream.accept(pair.client, version) : await Stream.open(pair.client, { version }); + if (!stream) throw new Error("no control stream"); + const close = () => { + connection.abort(); + origin.close(); + }; + return { pair, stream, close }; +} + +/** USE_VALUE, type 0, value 0xAA. */ +const TOKEN = [0x03, 0x00, 0xaa] as const; + +/** + * Section 8.9: a request whose tokens repeat is rejected MALFORMED_AUTH_TOKEN, which draft-15 + * moved from 0x10 to 0x4, and the session stays open for the next request. + */ +for (const [version, alpn, malformed] of [ + [Version.DRAFT_14, ALPN.DRAFT_14, 0x10], + [Version.DRAFT_22, ALPN.DRAFT_22, 0x4], +] as const) { + test(`a repeated request token rejects only the request ${alpn}`, async () => { + const { pair, stream, close } = await serving(version, alpn); + try { + await stream.writer.write(subscribeWith(version, tokenParams(version, TOKEN, TOKEN))); + if (version === Version.DRAFT_14) { + await skipTo(stream.reader, SubscribeError.id); + expect((await SubscribeError.decode(stream.reader, version)).errorCode).toBe(malformed); + } else { + expect(await stream.reader.u53()).toBe(RequestError.id); + expect((await RequestError.decode(stream.reader, version)).errorCode).toBe(malformed); + } + + const next = version === Version.DRAFT_14 ? stream : await Stream.open(pair.client, { version }); + await next.writer.write(subscribeWith(version, tokenParams(version, TOKEN, [0x03, 0x00, 0xbb]), 2)); + await skipTo(next.reader, SubscribeOk.id); + await SubscribeOk.decode(next.reader, version); + } finally { + close(); + } + }); +} + +/** + * An update whose tokens repeat is answered MALFORMED_AUTH_TOKEN, not NOT_SUPPORTED, naming + * the update's own Request ID on draft 16. + */ +for (const [version, alpn, requestId] of [ + [Version.DRAFT_16, ALPN.DRAFT_16, 2n], + [Version.DRAFT_22, ALPN.DRAFT_22, undefined], +] as const) { + test(`a repeated update token is malformed ${alpn}`, async () => { + const { stream, close } = await serving(version, alpn); + try { + await stream.writer.write(subscribeWith(version, [0])); + await skipTo(stream.reader, SubscribeOk.id); + await SubscribeOk.decode(stream.reader, version); + await stream.writer.write(updateWithTokens(version, TOKEN, TOKEN)); + await skipTo(stream.reader, RequestError.id); + const error = await RequestError.decode(stream.reader, version); + expect(error.errorCode).toBe(0x4); + expect(error.requestId).toBe(requestId); + } finally { + close(); + } + }); +} diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 892618bab8..74a6f83eae 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -6,7 +6,7 @@ import type { Established } from "../connection/established.ts"; import type { Drain } from "../connection/goaway.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; import { type Transport, transportOf } from "../connection/transport.ts"; -import { error, fromClose, ProtocolViolation, SessionCode, StreamCode, StreamError } from "../error.ts"; +import { error, fromClose, ProtocolViolation, StreamCode, StreamError } from "../error.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import * as Path from "../path.ts"; import { type Reader, Readers, type Stream } from "../stream.ts"; @@ -15,20 +15,24 @@ import { registerWire } from "../wire.ts"; import { ControlStreamAdapter, NativeSession, RequestWindowError, type Session } from "./adapter.ts"; import { AuthMessage, supported as authSupported, IetfAuthWire } from "./auth.ts"; import * as Cluster from "./cluster.ts"; -import { Fetch, FetchHeader } from "./fetch.ts"; +import { type RequestKind, toRequestCode } from "./error.ts"; +import { Fetch, FetchError, FetchHeader } from "./fetch.ts"; import { GoAway } from "./goaway.ts"; import { Group } from "./object.ts"; -import { Publish } from "./publish.ts"; -import { PublishNamespace } from "./publish_namespace.ts"; +import { Publish, PublishError } from "./publish.ts"; +import { PublishNamespace, PublishNamespaceError } from "./publish_namespace.ts"; import { Publisher } from "./publisher.ts"; -import { Subscribe, SubscribeUpdate } from "./subscribe.ts"; +import { RequestError } from "./request.ts"; +import { Subscribe, SubscribeError, SubscribeUpdate } from "./subscribe.ts"; import { SUBSCRIBE_TRACKS_ID, SubscribeNamespace, + SubscribeNamespaceError, SubscribeNamespaceLegacy, SubscribeOptions, } from "./subscribe_namespace.ts"; import { Subscriber } from "./subscriber.ts"; +import { MalformedAuthToken } from "./token.ts"; import { TrackStatusRequest } from "./track.ts"; import { type IetfVersion, Version, versionName } from "./version.ts"; @@ -273,7 +277,7 @@ export class Connection implements Established { // The peer broke the protocol, so losing the stream is not enough: nothing stops it // repeating the violation on the next one. #violated(err: ProtocolViolation) { - this.#close({ closeCode: SessionCode.ProtocolViolation, reason: err.message }); + this.#close({ closeCode: err.code, reason: err.message }); } async #run(early: Reader[]): Promise { @@ -343,12 +347,12 @@ export class Connection implements Established { // to the same request_id + namespace. We never send PUBLISH, so a legacy // request for PUBLISH alone is refused, and one for both gets only NAMESPACE. case BigInt(SubscribeNamespace.id): { - const msg = await SubscribeNamespace.decode(stream.reader, this.#session.version); - await this.#publisher.runSubscribeNamespace(msg, stream); + const decoded = SubscribeNamespace.decode(stream.reader, this.#session.version); + const msg = await this.#request(stream, "subscribe_namespace", decoded); + if (msg) await this.#publisher.runSubscribeNamespace(msg, stream); break; } case BigInt(SubscribeNamespaceLegacy.id): { - const legacy = await SubscribeNamespaceLegacy.decode(stream.reader, this.#session.version); // Draft-16 carries this on its own stream, so the control adapter never sees the ID. // Draft-14 and 15 already admitted it off the control stream. A refused request // still spends its ID, so it is admitted and released like any other. @@ -356,6 +360,19 @@ export class Connection implements Established { this.#session instanceof ControlStreamAdapter && this.#session.version === Version.DRAFT_16 ? this.#session : undefined; + let legacy: SubscribeNamespaceLegacy; + try { + legacy = await SubscribeNamespaceLegacy.decode(stream.reader, this.#session.version); + } catch (err) { + if (!(err instanceof MalformedAuthToken) || err.requestId === undefined) throw err; + adapter?.admitRequest(err.requestId); + try { + await this.#refuseToken(stream, "subscribe_namespace", err.requestId); + } finally { + adapter?.releaseRequest(err.requestId); + } + break; + } if (adapter) adapter.admitRequest(legacy.requestId); try { if (legacy.subscribeOptions === SubscribeOptions.PUBLISH) { @@ -380,8 +397,9 @@ export class Connection implements Established { } // Publisher handles incoming requests case BigInt(Subscribe.id): { - const msg = await Subscribe.decode(stream.reader, this.#session.version); - await this.#publisher.runSubscribe(msg, stream); + const decoded = Subscribe.decode(stream.reader, this.#session.version); + const msg = await this.#request(stream, "subscribe", decoded); + if (msg) await this.#publisher.runSubscribe(msg, stream); break; } case BigInt(SUBSCRIBE_TRACKS_ID): { @@ -393,23 +411,27 @@ export class Connection implements Established { break; } case BigInt(TrackStatusRequest.id): { - const msg = await TrackStatusRequest.decode(stream.reader, this.#session.version); - await this.#publisher.runTrackStatusRequest(msg, stream); + const decoded = TrackStatusRequest.decode(stream.reader, this.#session.version); + const msg = await this.#request(stream, "track_status", decoded); + if (msg) await this.#publisher.runTrackStatusRequest(msg, stream); break; } case BigInt(Fetch.id): { - const msg = await Fetch.decode(stream.reader, this.#session.version); - await this.#publisher.runFetch(msg, stream); + const decoded = Fetch.decode(stream.reader, this.#session.version); + const msg = await this.#request(stream, "fetch", decoded); + if (msg) await this.#publisher.runFetch(msg, stream); break; } // Subscriber handles incoming notifications case BigInt(PublishNamespace.id): { - const msg = await PublishNamespace.decode( + const decoded = PublishNamespace.decode( stream.reader, this.#session.version, Cluster.negotiated(this.#cluster), ); + const msg = await this.#request(stream, "publish_namespace", decoded); + if (!msg) break; // Unless we declined it, we declare that advertisements to us must be // solicited (MoQ Solicit), and writing the option at all proves the peer @@ -441,8 +463,9 @@ export class Connection implements Established { break; } case BigInt(Publish.id): { - const msg = await Publish.decode(stream.reader, this.#session.version); - await this.#subscriber.runPublish(msg, stream); + const decoded = Publish.decode(stream.reader, this.#session.version); + const msg = await this.#request(stream, "publish", decoded); + if (msg) await this.#subscriber.runPublish(msg, stream); break; } @@ -451,6 +474,45 @@ export class Connection implements Established { } } + /** + * Finish decoding a request, refusing it instead when its AUTHORIZATION TOKENs repeat. + * Section 8.9 rejects that message rather than the session, unlike any other decode error. + * + * @returns undefined once refused. + */ + async #request(stream: Stream, kind: RequestKind, decoded: Promise): Promise { + try { + return await decoded; + } catch (err) { + if (!(err instanceof MalformedAuthToken) || err.requestId === undefined) throw err; + await this.#refuseToken(stream, kind, err.requestId); + return undefined; + } + } + + /** Refuse a request MALFORMED_AUTH_TOKEN, ending its stream. Mirrors Rust's `token::reject`. */ + async #refuseToken(stream: Stream, kind: RequestKind, requestId: bigint) { + const version = this.#session.version; + // Each draft-14 error message has the REQUEST_ERROR body that draft-15 folded them into. + const id14 = { + subscribe: SubscribeError.id, + fetch: FetchError.id, + publish: PublishError.id, + publish_namespace: PublishNamespaceError.id, + subscribe_namespace: SubscribeNamespaceError.id, + // TRACK_STATUS_ERROR + track_status: 0x0f, + }[kind]; + const legacy = version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16; + await stream.writer.u53(version === Version.DRAFT_14 ? id14 : RequestError.id); + await new RequestError({ + requestId: legacy ? requestId : undefined, + errorCode: toRequestCode("malformed_auth_token", kind, version), + reasonPhrase: "repeated AUTHORIZATION TOKEN", + }).encode(stream.writer, version); + stream.close(); + } + // A malformed OBJECT_DATAGRAM is the peer breaking the protocol, like an invalid stream type. async #runDatagrams() { try { diff --git a/js/net/src/ietf/error.test.ts b/js/net/src/ietf/error.test.ts index 12d415cbcb..5e51ecf68c 100644 --- a/js/net/src/ietf/error.test.ts +++ b/js/net/src/ietf/error.test.ts @@ -86,7 +86,8 @@ test("fetch refusal codes follow each draft registry", () => { test("only registered codes reach the wire", () => { for (const version of ALL) { for (const kind of KINDS) { - for (const condition of CONDITIONS) { + // Sent but never read back as itself, so it stays out of the round trip. + for (const condition of [...CONDITIONS, "malformed_auth_token" as const]) { const code = toRequestCode(condition, kind, version); expect(registered(kind, version)).toContain(code); } @@ -106,6 +107,19 @@ test("every emitted code round trips", () => { } }); +/** + * Draft-15 swapped MALFORMED_AUTH_TOKEN and the missing-track code, and draft-14 gives + * PUBLISH_ERROR no value for it. + */ +test("a malformed token uses the draft's own number", () => { + for (const kind of KINDS) { + expect(toRequestCode("malformed_auth_token", kind, Version.DRAFT_14)).toBe(kind === "publish" ? 0x0 : 0x10); + for (const version of ALL.slice(1)) { + expect(toRequestCode("malformed_auth_token", kind, version)).toBe(0x4); + } + } +}); + /** * Draft-14 numbers a missing track 0x4 and draft-15 moved it to 0x10, which is draft-14's * MALFORMED_AUTH_TOKEN. Getting this backwards tells a peer its token is broken when the diff --git a/js/net/src/ietf/error.ts b/js/net/src/ietf/error.ts index 21a5a0b526..f19449b586 100644 --- a/js/net/src/ietf/error.ts +++ b/js/net/src/ietf/error.ts @@ -99,6 +99,8 @@ export type RequestCondition = | "uninterested" /** The track's content could not be parsed. */ | "malformed_track" + /** A well-formed AUTHORIZATION TOKEN this endpoint cannot use, such as a repeat. */ + | "malformed_auth_token" /** A GOAWAY is draining the session, so no new request is accepted. */ | "going_away"; @@ -158,6 +160,16 @@ function malformedTrack(kind: RequestKind, version: IetfVersion): number | undef return kind === "fetch" ? MALFORMED_TRACK_14 : undefined; } +/** Draft-15 swapped MALFORMED_AUTH_TOKEN and the missing-track code. */ +const MALFORMED_AUTH_TOKEN_14 = 0x10; +const MALFORMED_AUTH_TOKEN = 0x4; + +/** The value for an unusable token, or undefined where none is assigned: draft-14 PUBLISH_ERROR. */ +function malformedAuthToken(kind: RequestKind, version: IetfVersion): number | undefined { + if (version !== Version.DRAFT_14) return MALFORMED_AUTH_TOKEN; + return kind === "publish" ? undefined : MALFORMED_AUTH_TOKEN_14; +} + /** The value for a draining session, or undefined before draft-17 registered one. */ function goingAway(version: IetfVersion): number | undefined { return version >= Version.DRAFT_17 ? GOING_AWAY : undefined; @@ -199,6 +211,8 @@ export function toRequestCode(condition: RequestCondition, kind: RequestKind, ve : (uninterested(kind, version) ?? INTERNAL_ERROR); case "malformed_track": return malformedTrack(kind, version) ?? INTERNAL_ERROR; + case "malformed_auth_token": + return malformedAuthToken(kind, version) ?? INTERNAL_ERROR; case "going_away": return goingAway(version) ?? INTERNAL_ERROR; } diff --git a/js/net/src/ietf/message.ts b/js/net/src/ietf/message.ts index 06be4faacd..4a080e9a97 100644 --- a/js/net/src/ietf/message.ts +++ b/js/net/src/ietf/message.ts @@ -1,4 +1,5 @@ import { Reader, Writer } from "../stream.ts"; +import { MalformedAuthToken } from "./token.ts"; // Encodes a message with a u16 (16-bit) size prefix as per draft-14. A type `id` is written // only once the body fits, so an oversized message leaves nothing on the stream. @@ -58,7 +59,14 @@ export async function decode(reader: Reader, f: (r: Reader) => Promise): P const data = await reader.read(size); const limit = new Reader(undefined, data, reader.version); - const msg = await f(limit); + let msg: T; + try { + msg = await f(limit); + } catch (err) { + // Every request starts with its Request ID, which the refusal names. + if (err instanceof MalformedAuthToken) err.requestId = await new Reader(undefined, data, reader.version).u62(); + throw err; + } // Check that we consumed exactly the right number of bytes if (!(await limit.done())) { diff --git a/js/net/src/ietf/parameters.ts b/js/net/src/ietf/parameters.ts index 7ec969bbb4..24c6e299e2 100644 --- a/js/net/src/ietf/parameters.ts +++ b/js/net/src/ietf/parameters.ts @@ -2,13 +2,14 @@ import { ProtocolViolation } from "../error.ts"; import type { Reader, Writer } from "../stream.ts"; import * as Varint from "../varint.ts"; import * as Filter from "./filter.ts"; +import { AUTHORIZATION_TOKEN, type Token, tokenFromRequest, unique } from "./token.ts"; import { type IetfVersion, Version } from "./version.ts"; /// Setup Option key constants (separate namespace from Message Parameters). export const SetupOption = { Path: 1n, MaxRequestId: 2n, - AuthorizationToken: 3n, + AuthorizationToken: AUTHORIZATION_TOKEN, MaxAuthTokenCacheSize: 4n, Authority: 5n, Implementation: 7n, @@ -221,8 +222,8 @@ const MSG_PARAM_ROUTE_COST = 0x40b58n; const MSG_PARAM_HIDDEN = 0x40b5en; // Bytes parameter IDs (odd) -/// AUTHORIZATION TOKEN. Ignored: the session's grant is what authorizes a request. -const MSG_PARAM_AUTHORIZATION_TOKEN = 0x03n; +/// AUTHORIZATION TOKEN. Decoded, but grants nothing: the session's grant is what authorizes a request. +const MSG_PARAM_AUTHORIZATION_TOKEN = AUTHORIZATION_TOKEN; const MSG_PARAM_LARGEST_OBJECT = 0x09n; const MSG_PARAM_SUBSCRIPTION_FILTER = 0x21n; /// FILL_PARAMETERS, draft-20's request for a backfill. @@ -254,6 +255,7 @@ function draft20(version: IetfVersion): boolean { /** True when this draft's definition of `message` includes `id`. */ function allows(message: ControlMessage, version: IetfVersion, id: bigint): boolean { + const d14 = version === Version.DRAFT_14; const d15 = version === Version.DRAFT_15; const d16 = version === Version.DRAFT_16; const d17 = version === Version.DRAFT_17; @@ -294,6 +296,8 @@ function allows(message: ControlMessage, version: IetfVersion, id: bigint): bool include ); case "subscribe-ok": + // Draft-14 defines only MAX_CACHE_DURATION here. + if (d14) return id === MSG_PARAM_MAX_CACHE_DURATION; return (id === MSG_PARAM_MAX_CACHE_DURATION && d15) || expires || largest || (order && d15); case "subscribe-update": case "request-update": @@ -432,15 +436,20 @@ export class Parameters { #filter: Filter.Filter | undefined; /** Every instance of a parameter that may repeat, decoded only; we never send one. */ #repeated: Map; + /** Every AUTHORIZATION TOKEN, checked for repeats once the block decoded. */ + #tokens: Token[]; constructor() { this.vars = new Map(); this.bytes = new Map(); this.#locations = new Map(); this.#repeated = new Map(); + this.#tokens = []; } - #repeat(id: bigint, value: Uint8Array) { + #repeat(id: bigint, value: Uint8Array, version: IetfVersion) { + // Every token decodes, so one this endpoint cannot use closes the session. + if (id === MSG_PARAM_AUTHORIZATION_TOKEN) this.#tokens.push(tokenFromRequest(value, version)); const values = this.#repeated.get(id) ?? []; values.push(value); this.#repeated.set(id, values); @@ -719,7 +728,11 @@ export class Parameters { * draft defines for a different message is ignored through draft-16 and closes the * session from draft-17 on. An id the draft does not define at all closes from * draft-16 on. Draft-14 and draft-15 ignore anything the message does not list. - * Omitting `message` keeps the previous decode, which stores every id. + * Omitting `message`, as a draft-14 request does, stores every id. + * + * Only a request may carry an AUTHORIZATION TOKEN, and every one decodes. + * + * @throws {MalformedAuthToken} once the block decoded, for tokens that repeat. */ static async decode(r: Reader, version: IetfVersion, message?: ControlMessage): Promise { const count = await r.u53(); @@ -766,7 +779,7 @@ export class Parameters { const size = await r.u53(); const bytes = await r.read(size); if (MSG_PARAM_REPEATABLE.includes(id)) { - params.#repeat(id, bytes); + params.#repeat(id, bytes, version); } else if (id === MSG_PARAM_LARGEST_OBJECT) { if (params.#locations.has(id)) { throw new Error(`duplicate message parameter id: ${id.toString()}`); @@ -790,7 +803,7 @@ export class Parameters { if (MSG_PARAM_REPEATABLE.includes(id)) { const size = await r.u53(); - params.#repeat(id, await r.read(size)); + params.#repeat(id, await r.read(size), version); continue; } @@ -837,6 +850,7 @@ export class Parameters { } } + unique(params.#tokens); return params; } } diff --git a/js/net/src/ietf/publish_namespace.ts b/js/net/src/ietf/publish_namespace.ts index ad0da3ec60..144e2cf66c 100644 --- a/js/net/src/ietf/publish_namespace.ts +++ b/js/net/src/ietf/publish_namespace.ts @@ -5,6 +5,7 @@ import * as Cluster from "./cluster.ts"; import * as Message from "./message.ts"; import * as Namespace from "./namespace.ts"; import { Parameters } from "./parameters.ts"; +import { MalformedAuthToken } from "./token.ts"; import { type IetfVersion, Version } from "./version.ts"; // In draft-14, ANNOUNCE is renamed to PUBLISH_NAMESPACE @@ -112,7 +113,12 @@ export class PublishNamespaceUpdate { try { return await Message.decode(r, (rd) => PublishNamespaceUpdate.#decode(rd, version)); } catch (err) { - if (err instanceof ProtocolViolation || err instanceof SessionError || err instanceof StreamError) + if ( + err instanceof ProtocolViolation || + err instanceof SessionError || + err instanceof StreamError || + err instanceof MalformedAuthToken + ) throw err; if (typeof err === "object" && err !== null) { const source = (err as { source?: unknown }).source; diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index ad639c1d97..9b299057c2 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -1,5 +1,6 @@ import { expect, jest, mock, spyOn, test } from "bun:test"; import { Signal } from "@moq/signals"; +import type { Grant } from "../auth.ts"; import type { Producer as BroadcastProducer } from "../broadcast.ts"; import * as Epoch from "../epoch.ts"; import { error } from "../error.ts"; @@ -14,6 +15,7 @@ import type { Producer as TrackProducer } from "../track.ts"; import { wireOf } from "../wire.ts"; import { ControlStreamAdapter, NativeSession, type Session } from "./adapter.ts"; import type * as Cluster from "./cluster.ts"; +import { toRequestCode } from "./error.ts"; import { Fetch, FetchHeader } from "./fetch.ts"; import { Frame, Group as GroupMessage } from "./object.ts"; import { PublishDone } from "./publish.ts"; @@ -115,7 +117,13 @@ function publisher( requiresSolicitation = false, session, cluster, - }: { requiresSolicitation?: boolean; session?: Session; cluster?: Cluster.Hops } = {}, + grant, + }: { + requiresSolicitation?: boolean; + session?: Session; + cluster?: Cluster.Hops; + grant?: Signal; + } = {}, ): { pub: Publisher; origin: OriginProducer } { const origin = new OriginProducer(); const inner = session ?? new NativeSession(transport, VERSION, true); @@ -126,6 +134,7 @@ function publisher( publish: origin.consume(), requiresSolicitation, cluster, + grant, }), origin, }; @@ -220,6 +229,49 @@ test("FETCH and a non-forwarding SUBSCRIBE get NOT_SUPPORTED", async () => { } }); +/** A request's USE_VALUE token decodes but grants nothing yet, so a SUBSCRIBE outside the grant is still refused. */ +test("a request token does not widen the grant", async () => { + for (const version of [Version.DRAFT_14, Version.DRAFT_22] as const) { + const pair = createMockTransportPair(ALPNS[version]); + const session = new NativeSession(pair.server, version, true); + const nothing = new Signal({ + publish: new Path.Patterns([]), + subscribe: new Path.Patterns([]), + }); + const { pub, origin } = publisher(pair.server, { session, grant: nothing }); + publish(origin, Path.from("room")); + + const ascii = (text: string) => [...new TextEncoder().encode(text)]; + // Draft-14 carries priority, group order, forward, and a Largest Object filter inline. + const fields = version === Version.DRAFT_14 ? [0x80, 0x02, 0x01, 0x02] : []; + const token = [0x03, 0x00, 0xaa]; // USE_VALUE, type 0 + const body = [0, 1, 4, ...ascii("room"), 5, ...ascii("video"), ...fields, 1, 0x03, token.length, ...token]; + const msg = await Subscribe.decode( + new Reader(undefined, new Uint8Array([0, body.length, ...body]), version), + version, + ); + + const written: Uint8Array[] = []; + const stream = new Stream({ + readable: new ReadableStream(), + writable: new WritableStream({ + write: (chunk) => { + written.push(new Uint8Array(chunk)); + }, + }), + version, + }); + await pub.runSubscribe(msg, stream); + await stream.writer.closed; + origin.close(); + + // SUBSCRIBE_ERROR on draft-14, REQUEST_ERROR after; the code follows the Length and any Request ID. + const wire = written.flatMap((chunk) => Array.from(chunk)); + expect(wire[0]).toBe(0x05); + expect(wire[version === Version.DRAFT_14 ? 4 : 3]).toBe(toRequestCode("unauthorized", "subscribe", version)); + } +}); + // The header is part of the group's lifetime too. If it blocks on flow control, advancing // the live edge must reset the stream without waiting for that write to finish. test("a blocked group header is reset when the group expires", async () => { diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 265c2d38da..07563572b8 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -4,7 +4,7 @@ import { enforceGrant } from "../auth_session.ts"; import type * as broadcast from "../broadcast.ts"; import { Withdrawal } from "../connection/withdrawal.ts"; import * as DatagramStream from "../datagram_stream.ts"; -import { controlTimeout, error, reason, StreamCode, StreamError } from "../error.ts"; +import { controlTimeout, error, ProtocolViolation, reason, StreamCode, StreamError } from "../error.ts"; import type * as group from "../group.ts"; import type { Hop, Route } from "../hop.ts"; import { hiddenBelow, hooks, presented } from "../internal.ts"; @@ -44,6 +44,7 @@ import { SubscribeNamespaceError, SubscribeNamespaceOk, } from "./subscribe_namespace.ts"; +import { MalformedAuthToken } from "./token.ts"; import type { TrackStatusRequest } from "./track.ts"; import { type IetfVersion, Version } from "./version.ts"; @@ -499,32 +500,52 @@ export class Publisher { return stream.writer.closed; } if ((await stream.reader.u53()) !== 0x02) throw new Error("unexpected message on subscribe stream"); + // Drafts 15 and 16 name the update's own Request ID; later drafts name none. + const refuse = async ( + condition: "not_supported" | "malformed_auth_token", + reasonPhrase: string, + requestId?: bigint, + ) => { + await stream.writer.u53(RequestError.id); + await new RequestError({ + requestId, + errorCode: toRequestCode(condition, "subscribe", version), + reasonPhrase, + }).encode(stream.writer, version); + return new UpdateFailed(); + }; if (legacy) { // Not applied yet, but read anyway: an unread update keeps the adapter's // stream from closing, which is how UNSUBSCRIBE arrives. - await SubscribeUpdate.decode(stream.reader, version); + try { + await SubscribeUpdate.decode(stream.reader, version); + } catch (err) { + if (!(err instanceof MalformedAuthToken)) throw err; + // Draft 14 answers no update, so the refusal only ends the subscription. + if (version === Version.DRAFT_14) throw new UpdateFailed(); + throw await refuse("malformed_auth_token", err.message, err.requestId); + } continue; } - const params = await Message.decode(stream.reader, async (r) => { - await r.u62(); - if (version === Version.DRAFT_17) await r.u62(); - return Parameters.decode(r, version, "request-update"); - }); + let params: Parameters; + try { + params = await Message.decode(stream.reader, async (r) => { + await r.u62(); + if (version === Version.DRAFT_17) await r.u62(); + return Parameters.decode(r, version, "request-update"); + }); + } catch (err) { + // Section 8.9 rejects the update, which says more than that we do not support it. + if (err instanceof MalformedAuthToken) throw await refuse("malformed_auth_token", err.message); + throw err; + } const unsupported = params.forward === false || params.bytes.size > 0 || params.rangeFilters || params.trackPropertyFilter || [...params.vars.keys()].some((key) => key !== 0x10n && key !== 0x20n); - if (unsupported) { - await stream.writer.u53(RequestError.id); - await new RequestError({ - requestId: undefined, - errorCode: toRequestCode("not_supported", "subscribe", version), - reasonPhrase: "REQUEST_UPDATE parameters not supported", - }).encode(stream.writer, version); - throw new UpdateFailed(); - } + if (unsupported) throw await refuse("not_supported", "REQUEST_UPDATE parameters not supported"); // update() replaces every option, so carry the rest over as Rust does. if (params.subscriberPriority !== undefined) track.update({ ...track.subscription.peek(), priority: fromWire(params.subscriberPriority) }); @@ -623,6 +644,9 @@ export class Publisher { } finally { disposeGrant(); } + // 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. + if (publishError instanceof ProtocolViolation) throw publishError; // PUBLISH_DONE waits until every stream this subscription will open is closed, as // the draft requires, so its count is final. The subscriber leaving cancels the @@ -681,6 +705,7 @@ export class Publisher { finished = true; stream.close(); } catch (err: unknown) { + if (err instanceof ProtocolViolation) throw err; const e = error(err); console.warn(`publish error: broadcast=${name} track=${track.name} error=${reason(e)}`); stream.abort(e); diff --git a/js/net/src/ietf/subscribe.ts b/js/net/src/ietf/subscribe.ts index bc9d341e0c..1eed75c3ea 100644 --- a/js/net/src/ietf/subscribe.ts +++ b/js/net/src/ietf/subscribe.ts @@ -300,7 +300,7 @@ export class SubscribeOk { largest = { groupId, objectId }; } - properties.maxCacheDuration = (await Parameters.decode(r, version)).maxCacheDuration; + properties.maxCacheDuration = (await Parameters.decode(r, version, "subscribe-ok")).maxCacheDuration; } else { // v15+: parameters followed by Track Properties (draft-17+). LARGEST_OBJECT is // required on every draft once the track has content, so rejecting it would tear diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index fd33200d55..9e6b94852d 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -54,6 +54,7 @@ import { SubscribeNamespaceOk, UnsubscribeNamespace, } from "./subscribe_namespace.ts"; +import { MalformedAuthToken } from "./token.ts"; import { Version } from "./version.ts"; // Bound on how long stream-open plus SUBSCRIBE_OK may take. Browsers cap @@ -1061,7 +1062,21 @@ export class Subscriber { ); } - const update = await PublishNamespaceUpdate.decode(stream.reader, version); + let update: PublishNamespaceUpdate; + try { + update = await PublishNamespaceUpdate.decode(stream.reader, version); + } catch (err) { + if (!(err instanceof MalformedAuthToken)) throw err; + // Section 8.9 rejects the update, which withdraws the advertisement as any + // refused update does. Only draft-17+ updates arrive here, so no Request ID. + await stream.writer.u53(RequestError.id); + await new RequestError({ + errorCode: toRequestCode("malformed_auth_token", "publish_namespace", version), + reasonPhrase: err.message, + }).encode(stream.writer, version); + stream.close(); + break; + } // The parameters exist only on a session that negotiated the extension; // anywhere else they are the peer's violation. diff --git a/js/net/src/ietf/token.test.ts b/js/net/src/ietf/token.test.ts index 5d4b4137d9..f2365e70c1 100644 --- a/js/net/src/ietf/token.test.ts +++ b/js/net/src/ietf/token.test.ts @@ -1,9 +1,9 @@ import { expect, test } from "bun:test"; -import { SessionCode, SessionError } from "../error.ts"; +import { ProtocolViolation, SessionCode } from "../error.ts"; import { Reader, Writer } from "../stream.ts"; import * as Varint from "../varint.ts"; import { SetupOption, SetupOptions } from "./parameters.ts"; -import { TOKEN_OUT_OF_BAND, type Token, tokenFromSetup, tokenIntoSetup } from "./token.ts"; +import { TOKEN_OUT_OF_BAND, type Token, tokenFromRequest, tokenFromSetup, tokenIntoSetup } from "./token.ts"; import { type IetfVersion, Version } from "./version.ts"; const VERSIONS: IetfVersion[] = [ @@ -54,16 +54,21 @@ async function received(params: SetupOptions, version: IetfVersion): Promise varint(field, version)), value]; - const raw = new Uint8Array(parts.reduce((total, part) => total + part.length, 0)); + const out = new Uint8Array(parts.reduce((total, part) => total + part.length, 0)); let offset = 0; for (const part of parts) { - raw.set(part, offset); + out.set(part, offset); offset += part.length; } + return out; +} + +/** The Token structure as a SETUP option. */ +function structure(version: IetfVersion, fields: bigint[], value: Uint8Array = new Uint8Array()): SetupOptions { const params = new SetupOptions(); - params.setBytes(SetupOption.AuthorizationToken, raw); + params.setBytes(SetupOption.AuthorizationToken, raw(version, fields, value)); return params; } @@ -71,7 +76,7 @@ function sessionCode(fn: () => unknown): SessionCode | undefined { try { fn(); } catch (err) { - if (err instanceof SessionError) return err.code; + if (err instanceof ProtocolViolation) return err.code; throw err; } return undefined; @@ -119,7 +124,8 @@ test("REGISTER is a value", () => { } }); -test("an alias reference is a protocol violation", () => { +/** Section 9.1.4: nothing can have been registered before SETUP, so an alias there is a violation. */ +test("a SETUP alias reference is a protocol violation", () => { for (const version of VERSIONS) { for (const aliasType of [0x0n, 0x2n]) { const params = structure(version, [aliasType, 7n]); @@ -130,7 +136,7 @@ test("an alias reference is a protocol violation", () => { test("an undecodable structure is a formatting error", () => { for (const version of VERSIONS) { - for (const fields of [[], [0x3n], [0x1n], [0x1n, 7n], [0x4n, 0n]]) { + for (const fields of [[], [0x3n], [0x1n], [0x1n, 7n], [0x0n], [0x2n], [0x4n, 0n]]) { const params = structure(version, fields); expect(sessionCode(() => tokenFromSetup(params, version))).toBe(SessionCode.KeyValueFormatting); } @@ -153,3 +159,37 @@ test("two tokens are refused", async () => { ); } }); + +test("a request token by value decodes", () => { + for (const version of VERSIONS) { + expect(tokenFromRequest(raw(version, [0x3n, TOKEN.kind], TOKEN.value), version)).toEqual(TOKEN); + } +}); + +/** Only SETUP falls back to a value: on a request, a registration overflows the cache. */ +test("a request registration overflows the cache", () => { + for (const version of VERSIONS) { + const token = raw(version, [0x1n, 7n, TOKEN.kind], TOKEN.value); + expect(sessionCode(() => tokenFromRequest(token, version))).toBe(SessionCode.AuthTokenCacheOverflow); + } +}); + +test("a request alias reference is an unknown alias", () => { + for (const version of VERSIONS) { + for (const aliasType of [0x0n, 0x2n]) { + const token = raw(version, [aliasType, 7n]); + expect(sessionCode(() => tokenFromRequest(token, version))).toBe(SessionCode.UnknownAuthTokenAlias); + } + } +}); + +test("an undecodable request token is a formatting error", () => { + for (const version of VERSIONS) { + // A registration, deletion, or alias use missing a field it names, or an alias use + // carrying a token type. + for (const fields of [[], [0x3n], [0x1n], [0x1n, 7n], [0x0n], [0x2n], [0x2n, 7n, 0n], [0x4n, 0n]]) { + const token = raw(version, fields); + expect(sessionCode(() => tokenFromRequest(token, version))).toBe(SessionCode.KeyValueFormatting); + } + } +}); diff --git a/js/net/src/ietf/token.ts b/js/net/src/ietf/token.ts index 818466696f..2139e05604 100644 --- a/js/net/src/ietf/token.ts +++ b/js/net/src/ietf/token.ts @@ -1,10 +1,11 @@ -import { SessionCode, SessionError } from "../error.ts"; +import { ProtocolViolation, SessionCode } from "../error.ts"; import * as Varint from "../varint.ts"; -import { SetupOption, type SetupOptions } from "./parameters.ts"; +import type { SetupOptions } from "./parameters.ts"; import { type IetfVersion, Version } from "./version.ts"; /** - * The `AUTHORIZATION TOKEN` Setup Option (draft-ietf-moq-transport-21 section 9.1.4). + * The `AUTHORIZATION TOKEN` Setup Option (draft-ietf-moq-transport-21 section 9.1.4) and + * Message Parameter. * * The value is the Token structure of section 8.9: an Alias Type, then fields that type * selects. We advertise no `MAX_AUTH_TOKEN_CACHE_SIZE`, so its default of 0 means no alias @@ -16,6 +17,13 @@ import { type IetfVersion, Version } from "./version.ts"; * @internal */ +/** + * The `AUTHORIZATION TOKEN` key, the same as a Setup Option and as a Message Parameter. + * + * @internal + */ +export const AUTHORIZATION_TOKEN = 0x03n; + /** Retire a registered alias. */ const DELETE = 0x0n; /** Register an alias for this type and value, then use them. */ @@ -26,7 +34,7 @@ const USE_ALIAS = 0x2n; const USE_VALUE = 0x3n; /** - * A credential presented in a SETUP's `AUTHORIZATION TOKEN` option. + * A credential presented in an `AUTHORIZATION TOKEN` option or parameter. * * @internal */ @@ -62,17 +70,71 @@ function leadingOnes(version: IetfVersion): boolean { * A second token is already refused as a duplicate option by {@link SetupOptions}: one * credential per connection. * - * @throws {SessionError} `ProtocolViolation` for an alias reference, which nothing before - * SETUP could have registered, or `KeyValueFormatting` for a structure that cannot be decoded. + * @throws {ProtocolViolation} carrying the session code to close with. * @internal */ export function tokenFromSetup(params: SetupOptions, version: IetfVersion): Token | undefined { - const raw = params.getBytes(SetupOption.AuthorizationToken); + const raw = params.getBytes(AUTHORIZATION_TOKEN); if (raw === undefined) return undefined; + return decode(raw, version, true); +} + +/** + * A request whose tokens repeat one another, which section 8.9 rejects with + * MALFORMED_AUTH_TOKEN rather than closing the session. + * + * @internal + */ +export class MalformedAuthToken extends Error { + /** The refused request. Every request starts with its Request ID, so framing fills it in. */ + requestId?: bigint; + + constructor() { + super("repeated AUTHORIZATION TOKEN"); + this.name = "MalformedAuthToken"; + } +} + +/** + * Refuse a request whose tokens repeat one another. + * + * Section 8.9 lets a token repeat only with a different type or value, after resolving + * aliases, of which we accept none. Checked once every parameter decoded, so a token that + * closes the session wins over a repeat. + * + * @throws {MalformedAuthToken} + * @internal + */ +export function unique(tokens: Token[]) { + const same = (a: Token, b: Token) => + a.kind === b.kind && a.value.length === b.value.length && a.value.every((byte, i) => byte === b.value[i]); + for (let i = 0; i < tokens.length; i++) { + if (tokens.slice(0, i).some((prev) => same(prev, tokens[i]))) throw new MalformedAuthToken(); + } +} + +/** + * Decode an `AUTHORIZATION TOKEN` message parameter on a request, by the SETUP option's rules + * except that a registration overflows the cache instead of falling back to a value. + * + * @throws {ProtocolViolation} carrying the session code to close with. + * @internal + */ +export function tokenFromRequest(raw: Uint8Array, version: IetfVersion): Token { + return decode(raw, version, false); +} +/** + * Decode a Token structure, refusing what this endpoint cannot accept. + * + * @throws {ProtocolViolation} `AuthTokenCacheOverflow` for a registration outside SETUP, + * `UnknownAuthTokenAlias` for an alias reference on a request (`ProtocolViolation` in SETUP), + * or `KeyValueFormatting` for a structure that cannot be decoded. + */ +function decode(raw: Uint8Array, version: IetfVersion, setup: boolean): Token { // Section 8.9: a structure that cannot be decoded closes with KEY_VALUE_FORMATTING_ERROR. const malformed = (cause?: unknown) => - new SessionError(SessionCode.KeyValueFormatting, { cause, reason: "malformed AUTHORIZATION TOKEN" }); + new ProtocolViolation("malformed AUTHORIZATION TOKEN", { cause, code: SessionCode.KeyValueFormatting }); const unvarint = (buf: Uint8Array): [bigint, Uint8Array] => { try { return leadingOnes(version) ? Varint.decodeLeadingOnes(buf) : Varint.decodeBigInt(buf); @@ -81,18 +143,36 @@ export function tokenFromSetup(params: SetupOptions, version: IetfVersion): Toke } }; + // Every field an alias type names decodes before its meaning applies, so a truncated + // structure is malformed whatever its type. let [aliasType, rest] = unvarint(raw); switch (aliasType) { case USE_VALUE: break; - // With no cache, section 9.1.4 treats a registration as a value; the alias is unused. case REGISTER: [, rest] = unvarint(rest); + // Section 8.9: a registration past the cache size of 0 closes the session. With no + // cache, section 9.1.4 treats one in SETUP as a value and the alias is unused. + if (!setup) { + unvarint(rest); + throw new ProtocolViolation("AUTHORIZATION TOKEN registration", { + code: SessionCode.AuthTokenCacheOverflow, + }); + } break; - // Section 9.1.4: nothing can have been registered before SETUP. case DELETE: - case USE_ALIAS: - throw new SessionError(SessionCode.ProtocolViolation, { reason: "AUTHORIZATION TOKEN alias in SETUP" }); + case USE_ALIAS: { + const [, trailing] = unvarint(rest); + if (trailing.length > 0) throw malformed(); + // Section 9.1.4: nothing can have been registered before SETUP. + if (setup) throw new ProtocolViolation("AUTHORIZATION TOKEN alias in SETUP"); + // Section 9.1.3: a cache size of 0 prohibits aliases, so none was ever registered. + // Section 8.9 rejects the message, but 0x17 is only a session code (section 12.2), + // so the session closes; doc/concept/standard.md records the deviation. + throw new ProtocolViolation("unknown AUTHORIZATION TOKEN alias", { + code: SessionCode.UnknownAuthTokenAlias, + }); + } default: throw malformed(); } @@ -115,5 +195,5 @@ export function tokenIntoSetup(params: SetupOptions, token: Token, version: Ietf out.set(aliasType, 0); out.set(kind, aliasType.length); out.set(token.value, aliasType.length + kind.length); - params.setBytes(SetupOption.AuthorizationToken, out); + params.setBytes(AUTHORIZATION_TOKEN, out); } diff --git a/js/net/src/lite/connection.ts b/js/net/src/lite/connection.ts index 51b96cec6e..78891d1bf3 100644 --- a/js/net/src/lite/connection.ts +++ b/js/net/src/lite/connection.ts @@ -6,7 +6,7 @@ import type { Established } from "../connection/established.ts"; import type { Drain } from "../connection/goaway.ts"; import { type Probe, type Stats, transportStats } from "../connection/stats.ts"; import { type Transport, transportOf } from "../connection/transport.ts"; -import { closeError, error, fromClose, ProtocolViolation, SessionCode, StreamCode, sessionCause } from "../error.ts"; +import { closeError, error, fromClose, ProtocolViolation, StreamCode, sessionCause } from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import type { Consumer as OriginConsumer } from "../origin.ts"; import * as Path from "../path.ts"; @@ -365,7 +365,7 @@ export class Connection implements Established { .catch((err: unknown) => { stream.stop(err); if (err instanceof ProtocolViolation) { - this.#quic.close({ closeCode: SessionCode.ProtocolViolation, reason: err.message }); + this.#quic.close({ closeCode: err.code, reason: err.message }); } }); } diff --git a/quest/m1/auth/request-token.md b/quest/m1/auth/request-token.md index 2be861a55f..b75969c524 100644 --- a/quest/m1/auth/request-token.md +++ b/quest/m1/auth/request-token.md @@ -80,7 +80,10 @@ silently ignored on moq-lite, an outbound update sniffer, unread - A namespace renews only from draft 17; a token-bearing draft-16 REQUEST_UPDATE for a PUBLISH_NAMESPACE is still ignored. Implement it (decided 2026-10-09: draft-16 allows it, and the Drafts decision covers 16), - with a test that a refused one sends PUBLISH_NAMESPACE_CANCEL. + with a test that a refused one sends PUBLISH_NAMESPACE_CANCEL. Its tokens + also decode by the request rules, in Rust and JS: REGISTER, an alias, or a + malformed structure closes the session, and a repeat is refused + MALFORMED_AUTH_TOKEN (deferred from #5107). Decided 2026-10-09: `set_request_token` keeps the typed `setup::Token` and the encoder writes only `USE_VALUE`. Alias forms wait for a consumer that diff --git a/rs/moq-net/src/coding/decode.rs b/rs/moq-net/src/coding/decode.rs index 4be934995f..fa33535ef6 100644 --- a/rs/moq-net/src/coding/decode.rs +++ b/rs/moq-net/src/coding/decode.rs @@ -93,6 +93,16 @@ pub enum DecodeError { /// The field does not exist in the negotiated protocol version. #[error("unsupported version")] Version, + + /// The value is one the draft answers by closing the session with this code, such as an + /// `AUTHORIZATION TOKEN` naming an alias. + #[error(transparent)] + Session(crate::SessionError), + + /// A request's `AUTHORIZATION TOKEN` is well formed but unusable, such as a repeat of + /// another, which rejects the request rather than closing the session. + #[error("malformed authorization token")] + MalformedAuthToken, } impl DecodeError { diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index edc758b9ef..81c05dc983 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -46,10 +46,19 @@ pub enum SessionError { #[error("control message timeout")] Timeout, + /// A token registration would exceed the advertised `MAX_AUTH_TOKEN_CACHE_SIZE`, which is + /// 0 when unadvertised. + #[error("auth token cache overflow")] + AuthTokenCacheOverflow, + /// No version could be negotiated. #[error("version negotiation failed")] Version, + /// A token named an alias that was never registered. + #[error("unknown auth token alias")] + UnknownAuthTokenAlias, + /// An application-chosen code, offset into the 64+ range on the wire. #[error("app code={0}")] App(u16), @@ -71,7 +80,9 @@ impl SessionError { Self::TooManyRequests => 0x7, Self::GoawayTimeout => 0x10, Self::Timeout => 0x11, + Self::AuthTokenCacheOverflow => 0x13, Self::Version => 0x15, + Self::UnknownAuthTokenAlias => 0x17, Self::App(app) => *app as u32 + 64, Self::Unknown(code) => *code, } @@ -92,7 +103,9 @@ impl SessionError { 0x7 => Self::TooManyRequests, 0x10 => Self::GoawayTimeout, 0x11 => Self::Timeout, + 0x13 => Self::AuthTokenCacheOverflow, 0x15 => Self::Version, + 0x17 => Self::UnknownAuthTokenAlias, code @ 64.. => match u16::try_from(code - 64) { Ok(app) => Self::App(app), Err(_) => Self::Unknown(code), @@ -549,7 +562,7 @@ impl From for Error { impl From<&Error> for SessionError { fn from(err: &Error) -> Self { match err { - Error::Session(err) => err.clone(), + Error::Session(err) | Error::Decode(coding::DecodeError::Session(err)) => err.clone(), // App codes share the 64+ range in both registries. Error::Stream(StreamError::App(app)) => Self::App(*app), // A stream-scoped code has no meaning in this registry; don't forward the number. @@ -595,6 +608,7 @@ impl From<&Error> for StreamError { Error::Session(_) => Self::Internal, Error::Cancel | Error::Closed => Self::Cancel, Error::SessionClosed => Self::Session(SessionError::Cancel), + Error::Decode(coding::DecodeError::Session(err)) => Self::Session(err.clone()), Error::Old => Self::Old, Error::Evicted => Self::Evicted, Error::Lagged => Self::TooFarBehind, @@ -659,7 +673,9 @@ mod tests { SessionError::TooManyRequests, SessionError::GoawayTimeout, SessionError::Timeout, + SessionError::AuthTokenCacheOverflow, SessionError::Version, + SessionError::UnknownAuthTokenAlias, SessionError::App(0), SessionError::App(404), ]; @@ -675,7 +691,9 @@ mod tests { assert_eq!(SessionError::Unauthorized.to_code(), 0x2); assert_eq!(SessionError::TooManyRequests.to_code(), 0x7); assert_eq!(SessionError::GoawayTimeout.to_code(), 0x10); + assert_eq!(SessionError::AuthTokenCacheOverflow.to_code(), 0x13); assert_eq!(SessionError::Version.to_code(), 0x15); + assert_eq!(SessionError::UnknownAuthTokenAlias.to_code(), 0x17); // The reserved 32-47 range, and anything else unregistered, keeps its value instead // of being given a meaning. A peer on the old placeholders (0x20-0x22) lands here. diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 9b9b4b77d2..6d9eccfdc5 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -141,7 +141,7 @@ pub fn from_stream_code(code: u32, version: Version) -> StreamError { /// without the draft in hand tells the peer the opposite of what happened. pub(crate) mod request { use super::Version; - use crate::{Error, SessionError, StreamError}; + use crate::{DecodeError, Error, SessionError, StreamError}; /// Which request the error answers, so draft-14 picks the right registry. /// @@ -207,6 +207,12 @@ pub(crate) mod request { /// The same condition from draft-15 on. const MALFORMED_TRACK: u64 = 0x12; + /// A well-formed AUTHORIZATION TOKEN this endpoint cannot use, as draft-14 numbers it. + const MALFORMED_AUTH_TOKEN_14: u64 = 0x10; + + /// The same condition from draft-15 on, where 0x10 became DOES_NOT_EXIST. + const MALFORMED_AUTH_TOKEN: u64 = 0x4; + /// The value for "the thing you asked for is not here", or `None` where the draft and /// request assign none. fn does_not_exist(kind: Kind, version: Version) -> Option { @@ -267,6 +273,16 @@ pub(crate) mod request { } } + /// The value for an unusable token, or `None` where the draft and request assign none. + /// Draft-14 registers none on PUBLISH_ERROR. + fn malformed_auth_token(kind: Kind, version: Version) -> Option { + match (version, kind) { + (Version::Draft14, Kind::Publish) => None, + (Version::Draft14, _) => Some(MALFORMED_AUTH_TOKEN_14), + _ => Some(MALFORMED_AUTH_TOKEN), + } + } + /// The value for a draining session, or `None` before draft-17 registered one. fn going_away(version: Version) -> Option { match version { @@ -307,6 +323,7 @@ pub(crate) mod request { Kind::Subscribe | Kind::Fetch | Kind::TrackStatus => does_not_exist(kind, version), Kind::Publish | Kind::PublishNamespace | Kind::SubscribeNamespace => uninterested(kind, version), }, + Error::Decode(DecodeError::MalformedAuthToken) => malformed_auth_token(kind, version), // Our own parse failure is, from the peer's side, a malformed track. Error::Decode(_) | Error::BoundsExceeded(_) @@ -449,6 +466,20 @@ pub(crate) mod request { )); } + /// Draft-15 swapped MALFORMED_AUTH_TOKEN and the missing-track code, and draft-14 + /// gives PUBLISH_ERROR no value for it. + #[test] + fn a_malformed_token_uses_the_draft_s_own_number() { + let err = Error::Decode(DecodeError::MalformedAuthToken); + for kind in KINDS { + let draft14 = if kind == Kind::Publish { INTERNAL_ERROR } else { 0x10 }; + assert_eq!(to_code(&err, kind, Version::Draft14), draft14); + for version in ALL.into_iter().skip(1) { + assert_eq!(to_code(&err, kind, version), 0x4); + } + } + } + /// GOING_AWAY arrived in draft-17. Sent to an earlier peer, 0x6 is unassigned there, /// so say INTERNAL_ERROR outright rather than a number that means nothing. #[test] @@ -546,9 +577,11 @@ pub(crate) mod request { } } + // Sent but never read back as itself: a peer's MALFORMED_AUTH_TOKEN stays remote. + let malformed = Error::Decode(DecodeError::MalformedAuthToken); for version in ALL { for kind in KINDS { - for err in &EVERY_ERROR { + for err in EVERY_ERROR.iter().chain([&malformed]) { let code = to_code(err, kind, version); assert!( registered(kind, version).contains(&code), diff --git a/rs/moq-net/src/ietf/fetch.rs b/rs/moq-net/src/ietf/fetch.rs index 15ca89921a..53f5df5283 100644 --- a/rs/moq-net/src/ietf/fetch.rs +++ b/rs/moq-net/src/ietf/fetch.rs @@ -196,14 +196,14 @@ impl Message for Fetch<'_> { let _required_request_id_delta = buf.varint()?; } - // The token is ignored: the session's grant is what authorizes the request. + // The token decodes but grants nothing: the session's grant is what authorizes the request. let (fetch_type, subscriber_priority, group_order, range_filters, fill_timeout, properties_wanted) = match version { Version::Draft14 => { let subscriber_priority = buf.u8()?; let group_order = GroupOrder::decode(buf, version)?; let fetch_type = FetchType::decode(buf, version)?; - Parameters::skip(buf)?; + super::token::unique(&Parameters::skip_request(buf)?)?; ( fetch_type, Some(subscriber_priority), @@ -216,7 +216,7 @@ impl Message for Fetch<'_> { Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18 | Version::Draft19 => { let fetch_type = FetchType::decode(buf, version)?; decode_params!(buf, version, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, 0x0A => fill_timeout: Option where !matches!(version, Version::Draft15 | Version::Draft16 | Version::Draft17), 0x20 => subscriber_priority: Option, 0x22 => group_order: Option, @@ -225,6 +225,7 @@ impl Message for Fetch<'_> { 0x27 => priority_filter: Vec where has_range_filters(version), 0x28 => object_property_filter: Vec where has_range_filters(version), ); + super::token::unique(&authorization_token)?; let range_filters = [ subgroup_filter, object_id_filter, @@ -248,7 +249,7 @@ impl Message for Fetch<'_> { let namespace = decode_namespace(buf)?; let track = Cow::Owned(buf.string()?); decode_params!(buf, version, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, 0x0A => fill_timeout: Option, 0x20 => subscriber_priority: Option, 0x21 => filter: Option, @@ -259,6 +260,7 @@ impl Message for Fetch<'_> { 0x28 => object_property_filter: Vec, 0x35 => include_properties: Option, ); + super::token::unique(&authorization_token)?; let range_filters = [ subgroup_filter, object_id_filter, @@ -354,7 +356,7 @@ impl Message for FetchOk { let end_of_track = buf.bool()?; let end_location = Location::decode(buf, version)?; let properties = super::Properties { - max_cache_duration: Parameters::skip(buf)?.map(std::time::Duration::from_millis), + max_cache_duration: Parameters::skip_response(buf)?.map(std::time::Duration::from_millis), ..Default::default() }; Ok(Self { diff --git a/rs/moq-net/src/ietf/message.rs b/rs/moq-net/src/ietf/message.rs index 99f1f73ae5..b6bd52e669 100644 --- a/rs/moq-net/src/ietf/message.rs +++ b/rs/moq-net/src/ietf/message.rs @@ -27,7 +27,7 @@ impl Encode for T { /// A control message body not decoded yet: a `u16` length, then that many bytes. /// /// Read after the message type, when the type decides how to decode the rest. -#[derive(Debug)] +#[derive(Clone, Debug)] pub struct Body(pub bytes::Bytes); impl Decode for Body { diff --git a/rs/moq-net/src/ietf/parameters.rs b/rs/moq-net/src/ietf/parameters.rs index 6b2005b485..6a64130f68 100644 --- a/rs/moq-net/src/ietf/parameters.rs +++ b/rs/moq-net/src/ietf/parameters.rs @@ -167,13 +167,29 @@ impl Encode for Parameters { } impl Parameters { - /// Consume a draft-14 message parameter block, returning its first MAX_CACHE_DURATION - /// (0x04), the one parameter we act on. - /// + /// Consume a draft-14 request's parameter block, returning every AUTHORIZATION TOKEN + /// (0x03), each decoded by [`super::token::RequestToken`]'s rules. + pub fn skip_request(r: &mut Decoder<'_>) -> Result, DecodeError> { + let mut tokens = Vec::new(); + Self::skip(r, Some(&mut tokens))?; + Ok(tokens) + } + + /// Consume a draft-14 response's parameter block, returning its first + /// MAX_CACHE_DURATION (0x04), the one parameter we act on. + pub fn skip_response(r: &mut Decoder<'_>) -> Result, DecodeError> { + Self::skip(r, None) + } + /// Draft-14 section 9.2 has a receiver ignore unrecognized parameters and allow their /// duplicates, and lets AUTHORIZATION TOKEN repeat, so unlike [`Parameters::decode`], - /// which SETUP uses, a repeat is not refused. - pub fn skip(r: &mut Decoder<'_>) -> Result, DecodeError> { + /// which SETUP uses, a repeat is not refused. A request collects its tokens into `tokens`; + /// on a message that does not list them, a response, they are ignored like any other + /// parameter. + fn skip( + r: &mut Decoder<'_>, + mut tokens: Option<&mut Vec>, + ) -> Result, DecodeError> { let count = r.varint()?; if count > MAX_PARAMS { return Err(DecodeError::TooMany); @@ -190,6 +206,11 @@ impl Parameters { cache_duration.get_or_insert(value); } } + _ if let Some(tokens) = tokens.as_deref_mut() + && kind == u64::from(ParameterBytes::AuthorizationToken) => + { + tokens.push(super::token::RequestToken::param_decode(r, Version::Draft14)?); + } _ => { let len = usize::try_from(r.varint()?).map_err(|_| DecodeError::BoundsExceeded)?; if len > MAX_KVP_VALUE_LEN { @@ -1211,24 +1232,33 @@ mod tests { /// instance instead of failing the message as a duplicate. #[test] fn test_param_repeat_allowed() { + use super::super::token::RequestToken; + use crate::setup::Token; + + let token = |kind| { + RequestToken(Token { + kind, + value: vec![0xAA], + }) + }; for version in [Version::Draft15, Version::Draft16, Version::Draft17, Version::Draft20] { let mut buf = Vec::new(); let mut w = Encoder::new(&mut buf, version.into()); w.varint(2).unwrap(); w.varint(0x03).unwrap(); - Opaque(vec![0xAA]).param_encode(&mut w, version).unwrap(); + token(0).param_encode(&mut w, version).unwrap(); // The second key: absolute before draft-16, a zero delta after. let second: u64 = if version == Version::Draft15 { 0x03 } else { 0 }; w.varint(second).unwrap(); - Opaque(vec![0xBB]).param_encode(&mut w, version).unwrap(); + token(1).param_encode(&mut w, version).unwrap(); let mut r = Decoder::new(&buf, version.into()); - let tokens = (|| -> Result, DecodeError> { - decode_params!(&mut r, version, 0x03 => tokens: Vec); + let tokens = (|| -> Result, DecodeError> { + decode_params!(&mut r, version, 0x03 => tokens: Vec); Ok(tokens) })() .unwrap_or_else(|e| panic!("{version}: {e}")); - assert_eq!(tokens, vec![Opaque(vec![0xAA]), Opaque(vec![0xBB])], "{version}"); + assert_eq!(tokens, vec![token(0), token(1)], "{version}"); assert!(r.is_empty(), "{version}"); } } @@ -1240,13 +1270,35 @@ mod tests { #[rustfmt::skip] let block = [ 0x04, // Number of Parameters - 0x03, 0x01, 0xAA, // AUTHORIZATION TOKEN - 0x03, 0x01, 0xBB, // and again + 0x03, 0x03, 0x03, 0x00, 0xAA, // AUTHORIZATION TOKEN: USE_VALUE, type 0 + 0x03, 0x03, 0x03, 0x01, 0xBB, // and again, type 1 0x3E, 0x05, // an unknown varint parameter 0x3E, 0x06, // and again ]; let mut r = Decoder::new(&block, Version::Draft14.into()); - Parameters::skip(&mut r).unwrap(); + Parameters::skip_request(&mut r).unwrap(); + assert!(r.is_empty()); + } + + /// A draft-14 request decodes every token; a response ignores one, as section 9.2 has + /// for a parameter on a message that does not list it. + #[test] + fn test_skip_decodes_draft14_request_tokens() { + #[rustfmt::skip] + let block = [ + 0x02, // Number of Parameters + 0x03, 0x03, 0x03, 0x00, 0xAA, // AUTHORIZATION TOKEN: USE_VALUE, type 0 + 0x03, 0x02, 0x02, 0x07, // and a USE_ALIAS, never registered + ]; + let mut r = Decoder::new(&block, Version::Draft14.into()); + let err = Parameters::skip_request(&mut r).unwrap_err(); + assert!( + matches!(err, DecodeError::Session(crate::SessionError::UnknownAuthTokenAlias)), + "{err:?}" + ); + + let mut r = Decoder::new(&block, Version::Draft14.into()); + assert_eq!(Parameters::skip_response(&mut r).unwrap(), None); assert!(r.is_empty()); } } diff --git a/rs/moq-net/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs index 843ddfedf4..a65e9fade4 100644 --- a/rs/moq-net/src/ietf/publish.rs +++ b/rs/moq-net/src/ietf/publish.rs @@ -314,7 +314,7 @@ impl Message for Publish<'_> { }; let forward = r.bool()?; // parameters - Parameters::skip(r)?; + super::token::unique(&Parameters::skip_request(r)?)?; Ok(Self { request_id, @@ -340,7 +340,7 @@ impl Message for Publish<'_> { // OBJECT_DELIVERY_TIMEOUT is also legal in draft-15. The values are dropped. decode_params!(r, version, 0x02 => _object_delivery_timeout: Option where version == Version::Draft15 || Filter::is_draft20(version), - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, 0x06 => _subgroup_delivery_timeout: Option where Filter::is_draft20(version), 0x08 => _expires: Option, 0x09 => largest_location: Option, @@ -349,6 +349,7 @@ impl Message for Publish<'_> { 0x21 => _filter: Option where Filter::is_draft20(version), 0x22 => group_order: Option where version == Version::Draft15 || Filter::is_draft20(version), ); + super::token::unique(&authorization_token)?; let mut properties = Properties::decode(r, version)?; properties.group_order = properties.group_order.or(group_order); @@ -440,7 +441,7 @@ impl Message for PublishOk { let filter = Filter::decode(r, version)?; // no parameters - Parameters::skip(r)?; + Parameters::skip_response(r)?; Ok(Self { request_id, diff --git a/rs/moq-net/src/ietf/publish_namespace.rs b/rs/moq-net/src/ietf/publish_namespace.rs index d3713f9b47..a54e146125 100644 --- a/rs/moq-net/src/ietf/publish_namespace.rs +++ b/rs/moq-net/src/ietf/publish_namespace.rs @@ -37,12 +37,13 @@ impl PublishNamespace<'_> { } let track_namespace = decode_namespace(r)?; - // The token is ignored: the session's grant is what authorizes the request. + // The token decodes but grants nothing: the session's grant is what authorizes the request. decode_params!(r, version, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, cluster::HOP_PATH => hops: Option, cluster::ROUTE_COST => cost: Option, ); + super::token::unique(&authorization_token)?; let cluster = match negotiated { true => Some(cluster::Advert { @@ -144,12 +145,13 @@ impl Message for PublishNamespaceUpdate { } _ => RequestId::decode(r, version)?, }; - // The token is ignored: the session's grant is what authorizes the request. + // The token decodes but grants nothing: the session's grant is what authorizes the request. decode_params!(r, version, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, cluster::HOP_PATH => hops: Option, cluster::ROUTE_COST => cost: Option, ); + super::token::unique(&authorization_token)?; Ok(Self { request_id, hops, cost }) } } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 8fb4fff3b0..a0632320aa 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -13,7 +13,7 @@ use std::{ use crate::transport::poll::SendStream as _; use crate::{ - AsPath, Error, Timescale, Timestamp, + AsPath, DecodeError, Error, Timescale, Timestamp, coding::{Encode as _, Encoder, Stream, Writer}, ietf::{self, Control, EndLocation, FetchHeader, FetchType, Filter, GroupOrder, Location, RequestId}, track::Subscription, @@ -540,7 +540,12 @@ where }); let task = match id { ietf::Subscribe::ID => { - let msg = ietf::Subscribe::decode_msg(&mut data, this.version)?; + let msg = match ietf::Subscribe::decode_msg(&mut data, this.version) { + Err(DecodeError::MalformedAuthToken) => { + return Ok(this.reject_token(stream, request::Kind::Subscribe, body.clone(), serve)); + } + msg => msg?, + }; if !data.is_empty() { return Err(Error::WrongSize); } @@ -555,7 +560,12 @@ where .maybe_boxed() } ietf::Fetch::ID => { - let msg = ietf::Fetch::decode_msg(&mut data, this.version)?; + let msg = match ietf::Fetch::decode_msg(&mut data, this.version) { + Err(DecodeError::MalformedAuthToken) => { + return Ok(this.reject_token(stream, request::Kind::Fetch, body.clone(), serve)); + } + msg => msg?, + }; if !data.is_empty() { return Err(Error::WrongSize); } @@ -572,17 +582,25 @@ where // to the same request_id + namespace. We never send PUBLISH, so a legacy // request for PUBLISH alone is refused, and one for both gets only NAMESPACE. ietf::SubscribeNamespace::ID | ietf::SubscribeNamespaceLegacy::ID => { - let (msg, options) = if id == ietf::SubscribeNamespace::ID { - let msg = ietf::SubscribeNamespace::decode_msg(&mut data, this.version)?; - (msg, ietf::SubscribeOptions::Namespace) + let decoded = if id == ietf::SubscribeNamespace::ID { + ietf::SubscribeNamespace::decode_msg(&mut data, this.version) + .map(|msg| (msg, ietf::SubscribeOptions::Namespace)) } else { - let legacy = ietf::SubscribeNamespaceLegacy::decode_msg(&mut data, this.version)?; - let msg = ietf::SubscribeNamespace { - request_id: legacy.request_id, - namespace: legacy.namespace, - hidden: legacy.hidden, - }; - (msg, legacy.subscribe_options) + ietf::SubscribeNamespaceLegacy::decode_msg(&mut data, this.version).map(|legacy| { + let msg = ietf::SubscribeNamespace { + request_id: legacy.request_id, + namespace: legacy.namespace, + hidden: legacy.hidden, + }; + (msg, legacy.subscribe_options) + }) + }; + let (msg, options) = match decoded { + Err(DecodeError::MalformedAuthToken) => { + let kind = request::Kind::SubscribeNamespace; + return Ok(this.reject_token(stream, kind, body.clone(), serve)); + } + decoded => decoded?, }; if !data.is_empty() { return Err(Error::WrongSize); @@ -623,7 +641,12 @@ where .maybe_boxed() } ietf::TrackStatus::ID => { - let msg = ietf::TrackStatus::decode_msg(&mut data, this.version)?; + let msg = match ietf::TrackStatus::decode_msg(&mut data, this.version) { + Err(DecodeError::MalformedAuthToken) => { + return Ok(this.reject_token(stream, request::Kind::TrackStatus, body.clone(), serve)); + } + msg => msg?, + }; if !data.is_empty() { return Err(Error::WrongSize); } @@ -643,6 +666,25 @@ where Ok(task) } + /// Refuse a request whose AUTHORIZATION TOKEN repeats (section 8.9), which unlike any + /// other decode error leaves the session open. + fn reject_token( + &self, + stream: Stream, + kind: request::Kind, + body: ietf::Body, + serve: Option, + ) -> MaybeSendBox<'static, ()> { + let version = self.version; + async move { + let _serve = serve; + if let Err(err) = super::token::reject(stream, kind, body, version).await { + tracing::debug!(%err, "token refusal failed"); + } + } + .maybe_boxed() + } + /// Handle a SUBSCRIBE on its bidi stream. fn run_subscribe_stream( self, @@ -976,6 +1018,15 @@ where match event { Ok(None) => return Some(Ok(())), Err(Error::Cancel) => return None, + // An update the draft answers by closing the session, such as one whose + // AUTHORIZATION TOKEN names an alias, closes it rather than only ending this + // subscription, as the same token on the request itself would. + Err(err @ Error::Decode(crate::coding::DecodeError::Session(_))) => { + self.session + .clone() + .close(crate::SessionError::from(&err).to_code(), &err.to_string()); + return Some(Err(err)); + } Err(err) => return Some(Err(err)), Ok(Some(update)) => { // Draft 14 answers no update. Drafts 15 and 16 answer on the control @@ -983,18 +1034,18 @@ where let answer = self.version != Version::Draft14; let answer_id = matches!(self.version, Version::Draft15 | Version::Draft16).then_some(update.request_id); - if update.unsupported { + if let Some(err) = update.refused { + let reason = match err { + Error::Unsupported => "REQUEST_UPDATE parameters not supported", + _ => "repeated AUTHORIZATION TOKEN", + }; let result = if answer { - self.write_subscribe_error( - &mut stream.writer, - update.request_id, - &Error::Unsupported, - "REQUEST_UPDATE parameters not supported", - ) - .await + self.write_subscribe_error(&mut stream.writer, update.request_id, &err, reason) + .await } else { Ok(()) }; + // Either way the update failed, which ends the subscription. return Some(result.and(Err(Error::Unsupported))); } if let Some(priority) = update.priority @@ -4491,8 +4542,8 @@ mod serve_tests { ) .await; - // REQUEST_UPDATE, Request ID 2, one AUTHORIZATION TOKEN (0x03) parameter. - request.push(&[0x02, 0, 5, 2, 1, 0x03, 1, 0x00]); + // REQUEST_UPDATE, Request ID 2, one AUTHORIZATION TOKEN (0x03): USE_VALUE, type 0. + request.push(&[0x02, 0, 6, 2, 1, 0x03, 2, 0x03, 0x00]); moq_net_sim::timeout(std::time::Duration::from_secs(10), serve) .await .expect("the failed update ends the subscription") @@ -4641,6 +4692,156 @@ mod serve_tests { } } + /// An update whose only parameters are AUTHORIZATION TOKENs holding `tokens`, framed for + /// the subscription's stream. + fn update_with_tokens(version: Version, tokens: &[&[u8]]) -> Vec { + let mut body = match version { + Version::Draft14 => { + let mut body = Vec::new(); + ietf::SubscribeUpdate { + request_id: RequestId(2), + subscription_request_id: Some(RequestId(0)), + start_location: Location { group: 0, object: 0 }, + end_group: 0, + subscriber_priority: 10, + forward: true, + } + .encode_msg(&mut Encoder::new(&mut body, version.into()), version) + .unwrap(); + // Replace the empty parameter block with the token. + assert_eq!(body.pop(), Some(0)); + body + } + // Then the request it updates, through draft 16. Every parameter but the token is + // left as it is. + Version::Draft15 | Version::Draft16 => vec![2, 0], + _ => vec![2], + }; + body.extend(token_params(version, tokens)); + let mut frame = vec![ietf::SubscribeUpdate::ID as u8, 0, body.len() as u8]; + frame.extend(body); + frame + } + + /// An update's AUTHORIZATION TOKEN that this endpoint cannot use closes the session with + /// the draft's code, as it would on the SUBSCRIBE, not only the subscription. + #[moq_net_sim::test] + async fn an_update_token_that_cannot_be_used_closes_the_session() { + for version in [Version::Draft14, Version::Draft22] { + for (token, expected, why) in [ + ( + &[0x01, 0x07, 0x00][..], + crate::SessionError::AuthTokenCacheOverflow, + "REGISTER", + ), + (&[0x00, 0x07][..], crate::SessionError::UnknownAuthTokenAlias, "DELETE"), + ( + &[0x02, 0x07][..], + crate::SessionError::UnknownAuthTokenAlias, + "USE_ALIAS", + ), + ( + &[0x03][..], + crate::SessionError::KeyValueFormatting, + "USE_VALUE without a type", + ), + ] { + let h = serve(version); + let mut session = ScriptedSession::per_stream(vec![update_with_tokens(version, &[token])]); + let mut stream = Stream::open(&mut session, version).await.unwrap(); + let mut serving = TrackServe::new( + h.session.clone(), + h.track.subscribe(None), + RequestId(0), + version, + ServeRange::default(), + None, + ); + let mut finished = false; + let served = h + .publisher + .run_subscription(&mut stream, &mut serving, &mut finished, async {}) + .await; + assert!(matches!(served, Some(Err(_))), "{version} {why}: {served:?}"); + let codes: Vec = h.log.closes().into_iter().map(|(code, _)| code).collect(); + assert_eq!(codes, vec![expected.to_code()], "{version} {why}"); + } + } + } + + /// A parameter block of AUTHORIZATION TOKENs. Draft-16 on codes each key as a delta from + /// the one before, so a repeat is 0. + fn token_params(version: Version, tokens: &[&[u8]]) -> Vec { + let mut params = vec![tokens.len() as u8]; + for (i, token) in tokens.iter().enumerate() { + let delta = i > 0 && !matches!(version, Version::Draft14 | Version::Draft15); + params.extend([if delta { 0 } else { 0x03 }, token.len() as u8]); + params.extend(*token); + } + params + } + + /// USE_VALUE, type 0, value 0xAA. + const TOKEN: &[u8] = &[0x03, 0x00, 0xAA]; + + /// Section 8.9: a request whose tokens repeat is rejected MALFORMED_AUTH_TOKEN, which + /// draft-15 moved from 0x10 to 0x4, and the session stays open. A token with another + /// value is not a repeat. + #[moq_net_sim::test] + async fn a_repeated_request_token_rejects_only_the_request() { + for (version, malformed) in [(Version::Draft14, 0x10), (Version::Draft22, 0x4)] { + let request = |tokens: &[&[u8]]| { + let mut body = track_status("room", true, version); + assert_eq!(body.pop(), Some(0), "{version}"); + body.extend(token_params(version, tokens)); + body + }; + + let code = refusal(version, ietf::TrackStatus::ID, request(&[TOKEN, TOKEN])).await; + assert_eq!(code, malformed, "{version}"); + + let h = serve(version); + let other = [0x03, 0x00, 0xBB]; + assert_eq!( + answer_code(&h, request(&[TOKEN, &other]), version).await, + None, + "{version}" + ); + } + } + + /// An update whose tokens repeat is answered MALFORMED_AUTH_TOKEN rather than + /// NOT_SUPPORTED, and fails only its subscription. + #[moq_net_sim::test] + async fn a_repeated_update_token_is_malformed() { + for version in [Version::Draft16, Version::Draft22] { + let h = serve(version); + let mut session = ScriptedSession::per_stream(vec![update_with_tokens(version, &[TOKEN, TOKEN])]); + let mut stream = Stream::open(&mut session, version).await.unwrap(); + let mut serving = TrackServe::new( + h.session.clone(), + h.track.subscribe(None), + RequestId(0), + version, + ServeRange::default(), + None, + ); + let mut finished = false; + let served = h + .publisher + .run_subscription(&mut stream, &mut serving, &mut finished, async {}) + .await; + assert!(matches!(served, Some(Err(Error::Unsupported))), "{version}: {served:?}"); + assert!(h.log.closes().is_empty(), "{version}: the session closed"); + + let wire = session.log.writes.lock().unwrap().clone(); + let mut buf = Decoder::new(&wire, version.into()); + assert_eq!(buf.varint().unwrap(), ietf::RequestError::ID, "{version}"); + let error = ietf::RequestError::decode(&mut buf, version).unwrap(); + assert_eq!(error.error_code, 0x4, "{version}"); + } + } + #[moq_net_sim::test] async fn namespace_requester_fin_is_version_gated() { for version in [Version::Draft17, Version::Draft18, Version::Draft19, Version::Draft22] { @@ -4925,6 +5126,31 @@ mod serve_tests { } } + /// A request's USE_VALUE token decodes but grants nothing yet, so a TRACK_STATUS outside + /// the session grant is still refused. + #[moq_net_sim::test] + async fn a_request_token_does_not_widen_the_grant() { + let nothing = crate::auth::Grant { + publish: Default::default(), + subscribe: Default::default(), + expires: None, + }; + for version in [Version::Draft14, Version::Draft22] { + let mut h = serve(version); + let auth = crate::auth::Handle::new(false); + auth.authorize(¬hing); + h.publisher = h.publisher.clone().with_auth(auth); + + let mut body = track_status("room", true, version); + // Replace the empty parameter block with an AUTHORIZATION TOKEN: USE_VALUE, type 0. + assert_eq!(body.pop(), Some(0), "{version}"); + body.extend([0x01, 0x03, 0x03, 0x03, 0x00, 0xAA]); + + let unauthorized = request::to_code(&Error::Unauthorized, request::Kind::TrackStatus, version); + assert_eq!(answer_code(&h, body, version).await, Some(unauthorized), "{version}"); + } + } + /// A TRACK_STATUS answer parked on its request stream when the grant narrows is reset, /// never delivered once the stream drains. #[moq_net_sim::test] diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index de82091117..cce5dca693 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -84,7 +84,9 @@ pub(super) struct Update { /// The first Request ID: the update's own on drafts 14-16, which their answer names. pub(super) request_id: super::RequestId, pub(super) priority: Option, - pub(super) unsupported: bool, + /// Why the update is refused, which fails the subscription: [`Error::Unsupported`], or a + /// [`DecodeError::MalformedAuthToken`] for tokens that repeat. + pub(super) refused: Option, } impl Update { @@ -112,14 +114,14 @@ impl Update { let _end_group = data.varint()?; let priority = data.u8()?; let forward = data.bool()?; - Parameters::skip(data)?; + let tokens = Parameters::skip_request(data)?; if !data.is_empty() { return Err(DecodeError::InvalidValue); } return Ok(Self { request_id, priority: Some(priority), - unsupported: !forward, + refused: Self::refused(&tokens, !forward), }); } // The request it updates, which the adapter already routed by. @@ -133,7 +135,7 @@ impl Update { } decode_params!(data, version, 0x02 => object_timeout: Option, - 0x03 => token: Vec, + 0x03 => token: Vec, 0x06 => subgroup_timeout: Option where !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17), 0x10 => forward: Option, 0x20 => priority: Option, @@ -149,23 +151,33 @@ impl Update { if !data.is_empty() { return Err(DecodeError::InvalidValue); } + let unsupported = forward == Some(false) + || filter.is_some() + || fill.is_some() + || object_timeout.is_some() + || subgroup_timeout.is_some() + || !token.is_empty() + || !subgroup_filter.is_empty() + || !object_filter.is_empty() + || !priority_filter.is_empty() + || !property_filter.is_empty() + || !track_filter.is_empty() + || new_group.is_some(); Ok(Self { request_id, priority, - unsupported: forward == Some(false) - || filter.is_some() - || fill.is_some() - || object_timeout.is_some() - || subgroup_timeout.is_some() - || !token.is_empty() - || !subgroup_filter.is_empty() - || !object_filter.is_empty() - || !priority_filter.is_empty() - || !property_filter.is_empty() - || !track_filter.is_empty() - || new_group.is_some(), + refused: Self::refused(&token, unsupported), }) } + + /// Section 8.9 rejects an update whose tokens repeat, which says more than that we do not + /// support it. + fn refused(tokens: &[super::token::RequestToken], unsupported: bool) -> Option { + match super::token::unique(tokens) { + Err(err) => Some(err.into()), + Ok(()) => unsupported.then_some(Error::Unsupported), + } + } } impl Decode for Update { @@ -196,6 +208,6 @@ mod tests { let mut buf = bytes::Bytes::from_static(&[0x02, 0x01, 0x20, 0x05]); let update = crate::coding::decode_buf(&mut buf, Version::Draft18, Update::decode_body).unwrap(); assert_eq!(update.priority, Some(5)); - assert!(!update.unsupported); + assert!(update.refused.is_none()); } } diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 92d3aea4f3..6301f5bbb7 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -2718,4 +2718,117 @@ mod tests { assert!(!driver.is_finished(), "the repeated withdrawal ended the session"); assert!(log.closes().is_empty(), "closed the session: {:?}", log.closes()); } + + /// A SUBSCRIBE for `room/video` whose parameters are one AUTHORIZATION TOKEN holding + /// `token`, framed as a control message. + fn subscribe_with_token(version: Version, token: &[u8]) -> Vec { + let msg = ietf::Subscribe { + request_id: RequestId(0), + track_namespace: crate::Path::new("room"), + track_name: "video".into(), + subscriber_priority: 128, + group_order: ietf::GroupOrder::Descending, + filter: ietf::Filter::NextObject, + fill: None, + properties_wanted: true, + forward: true, + range_filters: false, + }; + let mut body = Vec::new(); + let mut w = crate::coding::Encoder::new(&mut body, version.into()); + match version { + Version::Draft14 => { + msg.encode_msg(&mut w, version).unwrap(); + // Replace the empty parameter block with the token. + assert_eq!(body.pop(), Some(0)); + } + // The request ID, namespace and name; every parameter but the token defaults. + _ => w.slice(&[0, 1, 4, b'r', b'o', b'o', b'm', 5, b'v', b'i', b'd', b'e', b'o']), + } + let mut w = crate::coding::Encoder::new(&mut body, version.into()); + w.varint(1).unwrap(); + w.varint(u64::from(ietf::ParameterBytes::AuthorizationToken)).unwrap(); + w.bytes(token).unwrap(); + + let mut payload = Vec::new(); + let mut w = crate::coding::Encoder::new(&mut payload, version.into()); + w.varint(ietf::Subscribe::ID).unwrap(); + w.u16(body.len() as u16); + w.slice(&body); + payload + } + + /// A request's AUTHORIZATION TOKEN decodes by the SETUP option's rules, so one this + /// endpoint cannot use closes the session with the draft's code. Draft-14 carries the + /// request on the control stream, a strict draft on its own stream. + #[moq_net_sim::test] + async fn a_request_token_that_cannot_be_used_closes_the_session() { + for version in [Version::Draft14, Version::Draft22] { + for (token, expected, why) in [ + ( + &[0x01, 0x07, 0x00][..], + SessionError::AuthTokenCacheOverflow, + "REGISTER", + ), + (&[0x00, 0x07][..], SessionError::UnknownAuthTokenAlias, "DELETE"), + (&[0x02, 0x07][..], SessionError::UnknownAuthTokenAlias, "USE_ALIAS"), + ( + &[0x03][..], + SessionError::KeyValueFormatting, + "USE_VALUE without a type", + ), + ( + &[0x04, 0x00][..], + SessionError::KeyValueFormatting, + "an unknown alias type", + ), + ] { + let payload = subscribe_with_token(version, token); + let mut session = match version { + Version::Draft14 => crate::lite::test_transport::ScriptedSession::new(payload), + _ => { + crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_bidis(vec![payload]) + } + }; + let log = session.log.clone(); + let setup = match version { + Version::Draft14 => Some(Stream::open(&mut session, version).await.unwrap()), + _ => None, + }; + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let (driver, _goaway, _) = start(Config { + runtime: crate::time::Clock::sim(), + session, + setup, + request_id_max: None, + limits: Default::default(), + client: false, + publish: Some(origin.consume()), + subscribe: None, + peer_hop: None, + cost: None, + version, + path: None, + authority: None, + peer_setup_stream: None, + peer_declared: Some(peer::Peer::default()), + auth: crate::auth::Handle::new(false), + early_unis: Vec::new(), + extensions: Default::default(), + }) + .unwrap(); + + let err = moq_net_sim::timeout(std::time::Duration::from_secs(10), driver) + .await + .unwrap_or_else(|_| panic!("{version:?} {why}: the session stayed up")) + .expect_err("the token must fail the session"); + assert_eq!(SessionError::from(&err), expected, "{version:?} {why}"); + assert_eq!( + log.closes(), + vec![(expected.to_code(), err.to_string())], + "{version:?} {why}" + ); + } + } + } } diff --git a/rs/moq-net/src/ietf/subscribe.rs b/rs/moq-net/src/ietf/subscribe.rs index 3d3f4dcd51..6b3f658489 100644 --- a/rs/moq-net/src/ietf/subscribe.rs +++ b/rs/moq-net/src/ietf/subscribe.rs @@ -69,7 +69,7 @@ impl Message for Subscribe<'_> { let forward = r.bool()?; let filter = Filter::decode(r, version)?; - Parameters::skip(r)?; + super::token::unique(&Parameters::skip_request(r)?)?; Ok(Self { request_id, @@ -85,12 +85,12 @@ impl Message for Subscribe<'_> { }) } _ => { - // The token is ignored: the session's grant is what authorizes the request. + // The token decodes but grants nothing: the session's grant is what authorizes the request. // NEW_GROUP_REQUEST is ignored too, as the draft lets a publisher without // dynamic groups do. decode_params!(r, version, 0x02 => _object_delivery_timeout: Option, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, // 0x04 is MAX_CACHE_DURATION in draft-15 (a publisher parameter) and not a // message parameter at all in draft-16. RENDEZVOUS_TIMEOUT arrives in draft-17. 0x04 => rendezvous_timeout: Option where !matches!(version, Version::Draft15 | Version::Draft16), @@ -107,6 +107,7 @@ impl Message for Subscribe<'_> { 0x32 => _new_group_request: Option where has_new_group_request(version), 0x35 => include_properties: Option where Filter::is_draft20(version), ); + super::token::unique(&authorization_token)?; let range_filters = [ subgroup_filter, @@ -276,7 +277,7 @@ impl Message for SubscribeOk { largest = Some(Location::decode(r, version)?); } - properties.max_cache_duration = Parameters::skip(r)?.map(std::time::Duration::from_millis); + properties.max_cache_duration = Parameters::skip_response(r)?.map(std::time::Duration::from_millis); } _ => { // GROUP_ORDER and MAX_CACHE_DURATION are legal here only in draft-15. Draft-16 @@ -433,7 +434,7 @@ impl Message for SubscribeUpdate { let end_group = r.varint()?; let subscriber_priority = r.u8()?; let forward = r.bool()?; - Parameters::skip(r)?; + Parameters::skip_request(r)?; Ok(Self { request_id, @@ -449,7 +450,7 @@ impl Message for SubscribeUpdate { let subscription_request_id = Some(RequestId::decode(r, version)?); decode_params!(r, version, 0x02 => _object_delivery_timeout: Option, - 0x03 => _authorization_token: Vec, + 0x03 => _authorization_token: Vec, 0x10 => forward: Option, 0x20 => subscriber_priority: Option, 0x21 => _filter: Option, @@ -478,7 +479,7 @@ impl Message for SubscribeUpdate { // consumed so a legal update does not fail the session. decode_params!(r, version, 0x02 => _object_delivery_timeout: Option, - 0x03 => _authorization_token: Vec, + 0x03 => _authorization_token: Vec, 0x06 => _subgroup_delivery_timeout: Option where !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17), 0x10 => forward: Option, 0x20 => subscriber_priority: Option, diff --git a/rs/moq-net/src/ietf/subscribe_namespace.rs b/rs/moq-net/src/ietf/subscribe_namespace.rs index 6bf2ff68a5..b77c53269f 100644 --- a/rs/moq-net/src/ietf/subscribe_namespace.rs +++ b/rs/moq-net/src/ietf/subscribe_namespace.rs @@ -74,11 +74,12 @@ impl Message for SubscribeNamespace<'_> { } let request_id = RequestId::decode(r, version)?; let namespace = decode_namespace(r)?; - // The token is ignored: the session's grant is what authorizes the request. + // The token decodes but grants nothing: the session's grant is what authorizes the request. decode_params!(r, version, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, HIDDEN_PARAM => hidden: Option, ); + super::token::unique(&authorization_token)?; Ok(Self { request_id, @@ -170,16 +171,17 @@ impl Message for SubscribeNamespaceLegacy<'_> { _ => SubscribeOptions::Namespace, }; - // The token is ignored: the session's grant is what authorizes the request. + // The token decodes but grants nothing: the session's grant is what authorizes the request. // FORWARD is legal here in draft-15 through draft-17. The value is checked (only // 0 or 1) and then dropped: a namespace subscription has no objects to forward. // Draft-14 has no such parameter, so an unlisted id is ignored. Draft-18 moved // FORWARD onto SUBSCRIBE_TRACKS and closes if it shows up on the new message. decode_params!(r, version, - 0x03 => _authorization_token: Vec, + 0x03 => authorization_token: Vec, 0x10 => _forward: Option where matches!(version, Version::Draft15 | Version::Draft16 | Version::Draft17), HIDDEN_PARAM => hidden: Option, ); + super::token::unique(&authorization_token)?; Ok(Self { request_id, diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 35eee6993a..f72e969df3 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1145,7 +1145,12 @@ where let mut data = body.decoder(this.version); let task = match id { ietf::Publish::ID => { - let msg = ietf::Publish::decode_msg(&mut data, this.version)?; + let msg = match ietf::Publish::decode_msg(&mut data, this.version) { + Err(DecodeError::MalformedAuthToken) => { + return Ok(this.reject_token(stream, request::Kind::Publish, body.clone())); + } + msg => msg?, + }; if !data.is_empty() { return Err(Error::WrongSize); } @@ -1160,7 +1165,12 @@ where ietf::PublishNamespace::ID => { // A negotiated session that omits HOP_PATH fails the decode here, which // the dispatcher turns into the protocol violation the draft requires. - let msg = ietf::PublishNamespace::decode_body(&mut data, this.version, peer.negotiated())?; + let msg = match ietf::PublishNamespace::decode_body(&mut data, this.version, peer.negotiated()) { + Err(DecodeError::MalformedAuthToken) => { + return Ok(this.reject_token(stream, request::Kind::PublishNamespace, body.clone())); + } + msg => msg?, + }; if !data.is_empty() { return Err(Error::WrongSize); } @@ -1188,6 +1198,23 @@ where Ok(task) } + /// Refuse a request whose AUTHORIZATION TOKEN repeats (section 8.9), which unlike any + /// other decode error leaves the session open. + fn reject_token( + &self, + stream: Stream, + kind: request::Kind, + body: ietf::Body, + ) -> MaybeSendBox<'static, ()> { + let version = self.version; + async move { + if let Err(err) = super::token::reject(stream, kind, body, version).await { + tracing::debug!(%err, "token refusal failed"); + } + } + .maybe_boxed() + } + /// What the peer declared about being solicited (MoQ Solicit). /// /// Read once by the dispatch loop and handed to each stream rather than awaited per @@ -1365,7 +1392,21 @@ where return Ok(()); } - let msg = ietf::PublishNamespaceUpdate::decode_msg(&mut data, self.version)?; + let msg = match ietf::PublishNamespaceUpdate::decode_msg(&mut data, self.version) { + // Section 8.9 rejects the update, which withdraws the advertisement as any + // refused update does. + Err(DecodeError::MalformedAuthToken) => { + let request_id = ietf::RequestId::decode(&mut body.decoder(self.version), self.version)?; + let err = Error::Decode(DecodeError::MalformedAuthToken); + self.write_error(stream, request_id, &err, "repeated AUTHORIZATION TOKEN") + .await?; + if stream.writer.finish().is_ok() { + let _ = stream.writer.closed().await; + } + return Ok(()); + } + msg => msg?, + }; // Junk inside the declared size would otherwise be applied silently, which // is the one decode path that skipped the check the others make. if !data.is_empty() { diff --git a/rs/moq-net/src/ietf/token.rs b/rs/moq-net/src/ietf/token.rs index e5e1c3558d..615508d279 100644 --- a/rs/moq-net/src/ietf/token.rs +++ b/rs/moq-net/src/ietf/token.rs @@ -1,16 +1,17 @@ -//! The `AUTHORIZATION TOKEN` Setup Option (draft-ietf-moq-transport-21 section 9.1.4). +//! The `AUTHORIZATION TOKEN` Setup Option (draft-ietf-moq-transport-21 section 9.1.4) and +//! Message Parameter. //! //! The value is the Token structure of section 8.9: an Alias Type, then fields that //! type selects. We advertise no `MAX_AUTH_TOKEN_CACHE_SIZE`, so its default of 0 means //! no alias is ever registered and every token arrives by value. use crate::{ - Error, SessionError, - coding::{Decoder, EncodeError, Encoder}, + DecodeError, Error, SessionError, + coding::{Decode as _, Decoder, EncodeError, Encoder, Stream}, setup::Token, }; -use super::{ParameterBytes, Parameters, Version}; +use super::{Message as _, Param, ParameterBytes, Parameters, RequestId, Version, error::request}; /// Retire a registered alias. const DELETE: u64 = 0x0; @@ -28,37 +29,130 @@ const USE_VALUE: u64 = 0x3; pub fn from_setup(params: &Parameters, version: Version) -> Result, Error> { params .get_bytes(ParameterBytes::AuthorizationToken) - .map(|value| decode(value, version)) + .map(|value| decode(value, version, true).map_err(Error::Session)) .transpose() } /// Present `token` in our SETUP, by value. #[cfg_attr(not(test), expect(dead_code))] pub fn into_setup(params: &mut Parameters, token: &Token, version: Version) -> Result<(), EncodeError> { + params.set_bytes(ParameterBytes::AuthorizationToken, encode(token, version)?); + Ok(()) +} + +/// An `AUTHORIZATION TOKEN` parameter on a request message, decoded by the SETUP option's +/// rules except that a registration overflows the cache instead of falling back to a value. +/// +/// The parameter may repeat, so a request decodes it as a `Vec` and every instance is checked. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RequestToken(pub Token); + +impl Param for RequestToken { + fn param_encode(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { + w.bytes(&encode(&self.0, version)?) + } + + fn param_decode(r: &mut Decoder<'_>, version: Version) -> Result { + let value = r.bytes()?; + Ok(Self(decode(value, version, false).map_err(DecodeError::Session)?)) + } +} + +/// Refuse a request whose tokens repeat one another. +/// +/// Section 8.9 lets a token repeat only with a different type or value, after resolving +/// aliases, of which we accept none. Checked once every parameter decoded, so a token that +/// closes the session wins over a repeat. +pub fn unique(tokens: &[RequestToken]) -> Result<(), DecodeError> { + for (i, token) in tokens.iter().enumerate() { + if tokens[..i].contains(token) { + return Err(DecodeError::MalformedAuthToken); + } + } + Ok(()) +} + +/// Refuse a request that decoded to [`DecodeError::MalformedAuthToken`], ending its stream. +/// +/// Section 8.9 rejects the message rather than the session. Every request starts with its +/// Request ID, so it is read from the undecoded `body`. +pub(super) async fn reject( + mut stream: Stream, + kind: request::Kind, + body: super::Body, + version: Version, +) -> Result<(), Error> { + let request_id = RequestId::decode(&mut body.decoder(version), version)?; + // Each draft-14 error message has the REQUEST_ERROR body that draft-15 folded them into. + let id = match (version, kind) { + (Version::Draft14, request::Kind::Subscribe) => super::SubscribeError::ID, + (Version::Draft14, request::Kind::Fetch) => super::FetchError::ID, + (Version::Draft14, request::Kind::Publish) => super::PublishError::ID, + (Version::Draft14, request::Kind::PublishNamespace) => super::PublishNamespaceError::ID, + (Version::Draft14, request::Kind::SubscribeNamespace) => super::SubscribeNamespaceError::ID, + (Version::Draft14, request::Kind::TrackStatus) => super::TRACK_STATUS_ERROR_14, + _ => super::RequestError::ID, + }; + let err = Error::Decode(DecodeError::MalformedAuthToken); + stream.writer.varint(id).await?; + stream + .writer + .encode(&super::RequestError { + request_id: matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16).then_some(request_id), + error_code: request::to_code(&err, kind, version), + reason_phrase: "repeated AUTHORIZATION TOKEN".into(), + retry_interval: 0, + }) + .await?; + // The peer dropping the stream once it has the rejection is a normal end. + let _ = stream.writer.close().await; + Ok(()) +} + +/// Encode `token` by value. +fn encode(token: &Token, version: Version) -> Result, EncodeError> { let mut value = Vec::new(); let mut w = Encoder::new(&mut value, version.into()); w.varint(USE_VALUE)?; w.varint(token.kind)?; w.slice(&token.value); - params.set_bytes(ParameterBytes::AuthorizationToken, value); - Ok(()) + Ok(value) } -/// Decode a Token structure, refusing what a SETUP cannot carry. -fn decode(buf: &[u8], version: Version) -> Result { +/// Decode a Token structure, refusing what this endpoint cannot accept. +fn decode(buf: &[u8], version: Version, setup: bool) -> Result { // Section 8.9: a structure that cannot be decoded closes with KEY_VALUE_FORMATTING_ERROR. - let malformed = |_| Error::Session(SessionError::KeyValueFormatting); + let malformed = |_| SessionError::KeyValueFormatting; let mut r = Decoder::new(buf, version.into()); + // Every field an alias type names decodes before its meaning applies, so a truncated + // structure is malformed whatever its type. match r.varint().map_err(malformed)? { USE_VALUE => {} - // With no cache, section 9.1.4 treats a registration as a value; the alias is unused. REGISTER => { r.varint().map_err(malformed)?; + // Section 8.9: a registration past the cache size of 0 closes the session. With no + // cache, section 9.1.4 treats one in SETUP as a value and the alias is unused. + if !setup { + r.varint().map_err(malformed)?; + return Err(SessionError::AuthTokenCacheOverflow); + } + } + DELETE | USE_ALIAS => { + r.varint().map_err(malformed)?; + if !r.is_empty() { + return Err(SessionError::KeyValueFormatting); + } + // Section 9.1.4: nothing can have been registered before SETUP. + if setup { + return Err(SessionError::ProtocolViolation); + } + // Section 9.1.3: a cache size of 0 prohibits aliases, so none was ever registered. + // Section 8.9 rejects the message, but 0x17 is only a session code (section 12.2), + // so the session closes; doc/concept/standard.md records the deviation. + return Err(SessionError::UnknownAuthTokenAlias); } - // Section 9.1.4: nothing can have been registered before SETUP. - DELETE | USE_ALIAS => return Err(Error::ProtocolViolation), - _ => return Err(Error::Session(SessionError::KeyValueFormatting)), + _ => return Err(SessionError::KeyValueFormatting), } let kind = r.varint().map_err(malformed)?; @@ -73,6 +167,22 @@ mod tests { use super::*; use crate::coding::{Decode, Encode}; + /// Section 8.9: tokens may repeat only with a different type or value. + #[test] + fn a_repeated_token_is_malformed() { + let token = |kind, value: &[u8]| { + RequestToken(Token { + kind, + value: value.to_vec(), + }) + }; + assert!(unique(&[token(0, b"a"), token(1, b"a"), token(0, b"b")]).is_ok()); + assert!(matches!( + unique(&[token(0, b"a"), token(1, b"a"), token(0, b"a")]), + Err(DecodeError::MalformedAuthToken) + )); + } + const VERSIONS: [Version; 9] = [ Version::Draft14, Version::Draft15, @@ -203,20 +313,102 @@ mod tests { } } + /// Section 9.1.4: nothing can have been registered before SETUP, so an alias there is a + /// protocol violation. #[test] - fn an_alias_reference_is_a_protocol_violation() { + fn a_setup_alias_reference_is_a_protocol_violation() { for version in VERSIONS { for alias_type in [DELETE, USE_ALIAS] { let params = structure(version, &[alias_type, 7], &[]); let err = from_setup(¶ms, version).unwrap_err(); assert!( - matches!(err, Error::ProtocolViolation), + matches!(err, Error::Session(SessionError::ProtocolViolation)), "{version:?} {alias_type}: {err:?}" ); } } } + /// A request's token as a message parameter value: the length-prefixed structure. + fn request(version: Version, fields: &[u64], value: &[u8]) -> Result { + let mut raw = Vec::new(); + let mut w = Encoder::new(&mut raw, version.into()); + for field in fields { + w.varint(*field).unwrap(); + } + w.slice(value); + let mut param = Vec::new(); + Encoder::new(&mut param, version.into()).bytes(&raw).unwrap(); + + let mut r = Decoder::new(¶m, version.into()); + let token = RequestToken::param_decode(&mut r, version)?; + assert!(r.is_empty(), "{version:?}"); + Ok(token) + } + + #[test] + fn a_request_token_round_trips_on_every_draft() { + for version in VERSIONS { + let mut param = Vec::new(); + RequestToken(token()) + .param_encode(&mut Encoder::new(&mut param, version.into()), version) + .unwrap(); + let mut r = Decoder::new(¶m, version.into()); + assert_eq!( + RequestToken::param_decode(&mut r, version).unwrap(), + RequestToken(token()), + "{version:?}" + ); + } + } + + /// Only SETUP falls back to a value: on a request, a registration overflows the cache. + #[test] + fn a_request_registration_overflows_the_cache() { + for version in VERSIONS { + let err = request(version, &[REGISTER, 7, token().kind], &token().value).unwrap_err(); + assert!( + matches!(err, DecodeError::Session(SessionError::AuthTokenCacheOverflow)), + "{version:?}: {err:?}" + ); + } + } + + #[test] + fn a_request_alias_reference_is_an_unknown_alias() { + for version in VERSIONS { + for alias_type in [DELETE, USE_ALIAS] { + let err = request(version, &[alias_type, 7], &[]).unwrap_err(); + assert!( + matches!(err, DecodeError::Session(SessionError::UnknownAuthTokenAlias)), + "{version:?} {alias_type}: {err:?}" + ); + } + } + } + + #[test] + fn an_undecodable_request_token_is_a_formatting_error() { + for version in VERSIONS { + for (fields, why) in [ + (&[][..], "no alias type"), + (&[USE_VALUE][..], "no token type"), + (&[REGISTER][..], "a registration with no alias"), + (&[REGISTER, 7][..], "a registration with no token type"), + (&[DELETE][..], "a deletion with no alias"), + (&[USE_ALIAS][..], "an alias use with no alias"), + (&[USE_ALIAS, 7, 0][..], "an alias use with a token type"), + (&[0x4, 0][..], "an unknown alias type"), + ] { + let err = request(version, fields, &[]).unwrap_err(); + assert!( + matches!(err, DecodeError::Session(SessionError::KeyValueFormatting)), + "{version:?} {why}: {err:?}" + ); + } + } + } + #[test] fn an_undecodable_structure_is_a_formatting_error() { for version in VERSIONS { @@ -225,6 +417,8 @@ mod tests { (&[USE_VALUE][..], "no token type"), (&[REGISTER][..], "no alias"), (&[REGISTER, 7][..], "no token type after the alias"), + (&[DELETE][..], "a deletion with no alias"), + (&[USE_ALIAS][..], "an alias use with no alias"), (&[0x4, 0][..], "an unknown alias type"), ] { let params = structure(version, fields, &[]);