diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index e756517983..1d15d8cb71 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -7,8 +7,9 @@ description: TOML reference for moq-relay `moq-relay relay.toml`. Every key is also a CLI flag and environment variable that joins the section and key: `listen.tls.cert` is `--listen-tls-cert` / -`MOQ_LISTEN_TLS_CERT`. The exception is `listen.bind`, spelled `--listen` / -`MOQ_LISTEN`. Precedence is CLI, then env, then file, then defaults. +`MOQ_LISTEN_TLS_CERT`. The exceptions: `listen.bind` is spelled `--listen` / +`MOQ_LISTEN`, and `listen.extensions` and `connect.extensions` are TOML only. +Precedence is CLI, then env, then file, then defaults. `moq-relay --help` lists every flag. ## \[listen] @@ -32,6 +33,10 @@ bind = "127.0.0.1:4444" [listen.unix] # Plaintext qmux over a Unix socket, gated by peer credentials. bind = "/run/moq/internal.sock" allow.uid = [1001] + +[listen.extensions] # moq-transport SETUP extensions to offer, each on by default. +auth = false # Decline MoQ Auth (draft 17+). +solicit = false # Decline MoQ Solicit, so peers may announce unasked. ``` `timeout` bounds the whole handshake, transport through MoQ SETUP, so a peer @@ -163,6 +168,8 @@ tls.cert = "relay.pem" # Present a client certificate (mTLS to pee tls.key = "relay.key" goaway.redirect = "same-host" # How far to trust a draining peer's redirect URI. goaway.handover = "10s" # Cap on how long the drained upstream keeps serving. +extensions.auth = false # Decline MoQ Auth on outbound moq-transport sessions. +extensions.solicit = false # Decline MoQ Solicit. Both default to on, as in [listen.extensions]. ``` A draining upstream may name a replacement URI. `same-host` follows it only diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 071a330b6d..05d2648c70 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -43,6 +43,9 @@ ignores unknown parameters: [cluster](/draft/moq-cluster) routing hop lists, [active-count](/draft/moq-active-count) to count the `NAMESPACE` messages before a `SUBSCRIBE_NAMESPACE` is caught up, and [probe](/draft/moq-probe) for bandwidth estimation. +Both libraries offer solicit and auth (draft 17 and later) unless a side's +`extensions` setting declines one, which connects it as a peer that does not +speak that extension. The pooled JavaScript `Connection` has no such setting yet. [moq-e2ee](/draft/moq-e2ee) is not a transport extension. It encrypts application payloads, so relays still forward named tracks they cannot read. See [Encryption](/concept/hang#encryption). diff --git a/js/net/src/connection/accept.ts b/js/net/src/connection/accept.ts index 92e836ffbf..bf7df5be70 100644 --- a/js/net/src/connection/accept.ts +++ b/js/net/src/connection/accept.ts @@ -3,6 +3,7 @@ import * as Lite from "../lite/index.ts"; import type { Consumer as OriginConsumer, Producer as OriginProducer } from "../origin.ts"; import { Stream } from "../stream.ts"; import type { Established } from "./established.ts"; +import * as Extensions from "./extensions.ts"; import { forwardAnnounced } from "./forward.ts"; import { exchangeSetup } from "./handshake.ts"; @@ -33,6 +34,9 @@ export interface AcceptProps { * nothing. The entries retract when the session dies; see the `consume` connect option. */ consume?: OriginProducer; + + /** The moq-transport extensions to offer, each on unless set to `false`. */ + extensions?: Extensions.Extensions; } /** The per-session wiring shared by every negotiated protocol path. */ @@ -41,6 +45,8 @@ type SessionProps = { publish?: OriginConsumer; /** Whether this side dialed; only the dialing side aborts on a publication its grant does not cover. */ client: boolean; + /** The moq-transport extensions this side offers. */ + extensions: Extensions.Offered; }; /** @@ -70,6 +76,7 @@ async function acceptInner( discovery: props.discovery ?? true, publish: props.publish, client: false, + extensions: Extensions.offered(props.extensions), }; if (protocol === Ietf.ALPN.DRAFT_22) { @@ -118,7 +125,12 @@ async function acceptAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup(transport, version, "moq-lite-js"); + const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup( + transport, + version, + "moq-lite-js", + wiring.extensions, + ); return new Ietf.Connection({ ...wiring, @@ -164,7 +176,7 @@ async function acceptSetup( const params = new Ietf.SetupOptions(); params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true)); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); - Ietf.solicitIntoSetup(params); + Extensions.intoSetup(params, wiring.extensions, version); Ietf.hiddenIntoSetup(params); const server = new Ietf.ServerSetup({ version, parameters: params }); @@ -224,7 +236,7 @@ async function acceptNegotiated( const params = new Ietf.SetupOptions(); params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true)); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); - Ietf.solicitIntoSetup(params); + Extensions.intoSetup(params, wiring.extensions, setupVersion); Ietf.hiddenIntoSetup(params); const server = new Ietf.ServerSetup({ version: selectedVersion, parameters: params }); diff --git a/js/net/src/connection/connect.ts b/js/net/src/connection/connect.ts index 4a6074b076..b8c9b55ff4 100644 --- a/js/net/src/connection/connect.ts +++ b/js/net/src/connection/connect.ts @@ -9,6 +9,7 @@ import * as Hex from "../util/hex.ts"; import { redact } from "../util/log.ts"; import { isWebTransportSupported } from "./browser.ts"; import type { Established } from "./established.ts"; +import * as Extensions from "./extensions.ts"; import { forwardAnnounced } from "./forward.ts"; import { exchangeSetup } from "./handshake.ts"; @@ -110,6 +111,9 @@ export interface ConnectProps { */ consume?: OriginProducer; + /** The moq-transport extensions to offer, each on unless set to `false`. */ + extensions?: Extensions.Extensions; + /** * Aborts the connection attempt with the signal's reason. An already-aborted * signal rejects before anything opens, and aborting after the connection is @@ -124,6 +128,8 @@ type SessionProps = { publish?: OriginConsumer; /** Whether this side dialed; only the dialing side aborts on a publication its grant does not cover. */ client: boolean; + /** The moq-transport extensions this side offers. */ + extensions: Extensions.Offered; }; // Save if WebSocket won the last race, so we won't give QUIC a head start next time. @@ -170,6 +176,7 @@ async function connectInner(url: URL, props: Omit, abort: P discovery: props.discovery ?? true, publish: props.publish, client: true, + extensions: Extensions.offered(props.extensions), }; if (props.transport) { @@ -310,7 +317,7 @@ async function negotiate(url: URL, session: WebTransport, wiring: SessionProps): const params = new Ietf.SetupOptions(); params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(false)); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode("moq-lite-js")); - Ietf.solicitIntoSetup(params); + Extensions.intoSetup(params, wiring.extensions, setupVersion); Ietf.hiddenIntoSetup(params); const client = new Ietf.ClientSetup({ @@ -367,7 +374,12 @@ async function handshakeAlpn( version: Ietf.IetfVersion, wiring: SessionProps, ): Promise { - const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup(session, version, "moq-lite-js"); + const { control, early, solicit, hidden, cluster, auth } = await exchangeSetup( + session, + version, + "moq-lite-js", + wiring.extensions, + ); return new Ietf.Connection({ ...wiring, diff --git a/js/net/src/connection/extensions.test.ts b/js/net/src/connection/extensions.test.ts new file mode 100644 index 0000000000..60f19e8982 --- /dev/null +++ b/js/net/src/connection/extensions.test.ts @@ -0,0 +1,105 @@ +import { expect, spyOn, test } from "bun:test"; +import { SessionCode } from "../error.ts"; +import { AuthMessage, AuthOk } from "../ietf/auth.ts"; +import { PublishNamespace } from "../ietf/publish_namespace.ts"; +import { RequestOk } from "../ietf/request.ts"; +import { ALPN, Version } from "../ietf/version.ts"; +import { createMockTransportPair } from "../mock.ts"; +import * as Path from "../path.ts"; +import { Stream } from "../stream.ts"; +import { accept } from "./accept.ts"; +import { type Extensions, offered } from "./extensions.ts"; +import { exchangeSetup } from "./handshake.ts"; + +const VERSION = Version.DRAFT_18; +const URL_ = new URL("https://example.com"); + +/** What each side learns from the other's SETUP when each offers `client` and `server`. */ +async function handshake(client: Extensions, server: Extensions) { + const pair = createMockTransportPair(ALPN.DRAFT_18); + const [dialed, accepted] = await Promise.all([ + exchangeSetup(pair.client, VERSION, "test", offered(client)), + exchangeSetup(pair.server, VERSION, "test", offered(server)), + ]); + return { dialed, accepted }; +} + +/** Accept a draft-18 session offering `extensions`, against a peer offering every extension. */ +async function acceptWith(extensions: Extensions) { + const pair = createMockTransportPair(ALPN.DRAFT_18); + const [peer, connection] = await Promise.all([ + exchangeSetup(pair.client, VERSION, "test", offered()), + accept({ transport: pair.server, url: URL_, extensions }), + ]); + return { pair, peer, connection }; +} + +test("every extension is offered by default", () => { + expect(offered()).toEqual({ auth: true, solicit: true }); + expect(offered({ solicit: false })).toEqual({ auth: true, solicit: false }); +}); + +test("a declined extension stays out of SETUP", async () => { + // Control: both sides offer both. + const all = await handshake({}, {}); + expect(all.accepted.solicit).toBe(true); + expect(all.accepted.auth).toBe(true); + expect(all.dialed.auth).toBe(true); + + const none = await handshake({ auth: false, solicit: false }, {}); + expect(none.accepted.solicit).toBeUndefined(); + expect(none.accepted.auth).toBe(false); + // Negotiated only when both offer it, so the declining side does not speak it either. + expect(none.dialed.auth).toBe(false); + expect(none.dialed.solicit).toBe(true); +}); + +test("an AUTH to a side that declined it closes the session", async () => { + // Control: offered by both, the connection credential is granted. + const offering = await acceptWith({}); + const granted = await Stream.open(offering.pair.client, { version: VERSION }); + await new AuthMessage(0n, new Uint8Array()).encode(granted.writer, VERSION); + expect(await granted.reader.u53()).toBe(AuthOk.id); + offering.connection.close(); + + const logged = spyOn(console, "error").mockImplementation(() => void 0); + try { + const declined = await acceptWith({ auth: false }); + const refused = await Stream.open(declined.pair.client, { version: VERSION }); + // The session closes on the message type alone, so the rest of the write may fail. + await new AuthMessage(0n, new Uint8Array()).encode(refused.writer, VERSION).catch(() => undefined); + const info = await declined.pair.client.closed; + expect(info.closeCode).toBe(SessionCode.ProtocolViolation); + } finally { + logged.mockRestore(); + } +}); + +test("a side that declined solicit takes an unsolicited announce", async () => { + const announce = async (session: Awaited>) => { + const stream = await Stream.open(session.pair.client, { version: VERSION }); + await stream.writer.u53(PublishNamespace.id); + await new PublishNamespace({ + requestId: 0n, + trackNamespace: Path.from("room"), + cluster: { hops: [session.peer.cluster.self], cost: 0n }, + }).encode(stream.writer, VERSION); + return stream; + }; + + const declined = await acceptWith({ solicit: false }); + const taken = await announce(declined); + expect(await taken.reader.u53()).toBe(RequestOk.id); + declined.connection.close(); + + // Control: our SETUP declared it, so the peer that implements it broke it. + const logged = spyOn(console, "error").mockImplementation(() => void 0); + try { + const declaring = await acceptWith({}); + await announce(declaring); + await declaring.pair.client.closed; + expect(logged).toHaveBeenCalled(); + } finally { + logged.mockRestore(); + } +}); diff --git a/js/net/src/connection/extensions.ts b/js/net/src/connection/extensions.ts new file mode 100644 index 0000000000..e029f37b47 --- /dev/null +++ b/js/net/src/connection/extensions.ts @@ -0,0 +1,27 @@ +import * as Ietf from "../ietf/index.ts"; + +/** + * The moq-transport extensions a session offers in its SETUP, each on unless set to `false`. + * Turning one off connects as a peer that does not speak it. moq-lite carries both in its + * core, so this only changes moq-transport sessions. + */ +export interface Extensions { + /** The MoQ Auth extension: tokens presented and granted on their own stream (draft-17+). */ + auth?: boolean; + /** The MoQ Solicit extension: the peer answers our SUBSCRIBE_NAMESPACE rather than announcing unasked. */ + solicit?: boolean; +} + +/** Every extension resolved to whether it is offered. */ +export type Offered = Required; + +/** Resolve each unset extension to offered. */ +export function offered(extensions?: Extensions): Offered { + return { auth: extensions?.auth ?? true, solicit: extensions?.solicit ?? true }; +} + +/** Write the options for the offered extensions into a SETUP. */ +export function intoSetup(params: Ietf.SetupOptions, extensions: Offered, version: Ietf.IetfVersion) { + if (extensions.solicit) Ietf.solicitIntoSetup(params); + if (extensions.auth) Ietf.Auth.intoSetup(params, version); +} diff --git a/js/net/src/connection/handshake.ts b/js/net/src/connection/handshake.ts index 04da453389..2903de9da6 100644 --- a/js/net/src/connection/handshake.ts +++ b/js/net/src/connection/handshake.ts @@ -2,6 +2,7 @@ import { StreamError } from "../error.ts"; import { type Hop, randomHop } from "../hop.ts"; import * as Ietf from "../ietf/index.ts"; import { Reader, Stream, UnexpectedEnd, Writer } from "../stream.ts"; +import * as Extensions from "./extensions.ts"; /** * Draft-17+ SETUP exchange. Each side opens a uni stream, writes its Setup @@ -12,16 +13,17 @@ import { Reader, Stream, UnexpectedEnd, Writer } from "../stream.ts"; * Returns the control stream, the uni streams that arrived before it (see * {@link receiveSetup}), plus what the peer's SETUP declared: whether it requires * solicitation, which decides whether we announce namespaces unprompted (see the MoQ Solicit - * extension), its Hop ID (see the MoQ Cluster extension), and whether it offered MoQ Auth. - * We declare all three ourselves on - * every session: we send SUBSCRIBE_NAMESPACE for each prefix we want, so an unsolicited - * advertisement can tell us nothing we won't have asked for, and a peer that knows our Hop - * ID can withhold the advertisements that already flowed through us. + * extension), its Hop ID (see the MoQ Cluster extension), and whether MoQ Auth is negotiated. + * We declare our Hop ID on every session, and Solicit and Auth unless `extensions` declines + * them: we send SUBSCRIBE_NAMESPACE for each prefix we want, so an unsolicited advertisement + * can tell us nothing we won't have asked for, and a peer that knows our Hop ID can withhold + * the advertisements that already flowed through us. */ export async function exchangeSetup( transport: WebTransport, version: Ietf.IetfVersion, implementation: string, + extensions: Extensions.Offered, ): Promise<{ control: Stream; early: Reader[]; @@ -33,9 +35,8 @@ export async function exchangeSetup( const encoder = new TextEncoder(); const params = new Ietf.SetupOptions(); params.setBytes(Ietf.SetupOption.Implementation, encoder.encode(implementation)); - Ietf.solicitIntoSetup(params); + Extensions.intoSetup(params, extensions, version); Ietf.hiddenIntoSetup(params); - Ietf.Auth.intoSetup(params, version); // One id per session, like the moq-lite connection: nothing in this process forwards // between sessions, so there is nothing for a shared id to detect. @@ -55,7 +56,8 @@ export async function exchangeSetup( solicit: received.solicit, hidden: received.hidden, cluster: { self, peer: received.cluster }, - auth: received.auth, + // Negotiated only when both sides offer it. + auth: received.auth && extensions.auth, }; } diff --git a/js/net/src/connection/index.ts b/js/net/src/connection/index.ts index 9989292eb6..1cbbb3762c 100644 --- a/js/net/src/connection/index.ts +++ b/js/net/src/connection/index.ts @@ -14,6 +14,7 @@ export { type WebTransportProps, } from "./connect.ts"; export type { Established } from "./established.ts"; +export type { Extensions } from "./extensions.ts"; export { Connection, type ConnectionProps } from "./pool.ts"; export type { Probe, Stats } from "./stats.ts"; export type { Transport } from "./transport.ts"; diff --git a/js/net/src/connection/pool.ts b/js/net/src/connection/pool.ts index c2e939d7fa..21a8613bf4 100644 --- a/js/net/src/connection/pool.ts +++ b/js/net/src/connection/pool.ts @@ -20,6 +20,7 @@ import { type WebTransportProps as WebTransportPropsType, } from "./connect.ts"; import type { Established as EstablishedType } from "./established.ts"; +import type { Extensions as ExtensionsType } from "./extensions.ts"; import type { GoawayProps, Redirect as RedirectType } from "./goaway.ts"; import { Reload, type ReloadDelay, type ReloadStatus } from "./reload.ts"; import type { Probe as ProbeType, Stats as StatsType } from "./stats.ts"; @@ -372,6 +373,8 @@ export namespace Connection { export type CertificateHash = CertificateHashType; /** An established one-shot session. */ export type Established = EstablishedType; + /** The moq-transport extensions a one-shot session offers. */ + export type Extensions = ExtensionsType; } /** Throw if `props` cannot be honored, rather than silently dropping them. */ diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 306b9f7b2a..6707e79ca1 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -1,4 +1,5 @@ import { expect, spyOn, test } from "bun:test"; +import { offered } from "../connection/extensions.ts"; import { exchangeSetup } from "../connection/handshake.ts"; import { SessionCode } from "../error.ts"; import { createMockTransportPair } from "../mock.ts"; @@ -298,7 +299,7 @@ test("uni streams before SETUP are held until it lands", async () => { parameters.setBytes(SetupOption.Implementation, new TextEncoder().encode("test")); await new Setup({ parameters }).encode(setup, version); - const { control, early, solicit, hidden, cluster } = await exchangeSetup(pair.server, version, "test"); + const { control, early, solicit, hidden, cluster } = await exchangeSetup(pair.server, version, "test", offered()); expect(early.length).toBe(2); let closed = false; @@ -348,7 +349,7 @@ test("malformed uni stream before SETUP fails the handshake", async () => { parameters.setBytes(SetupOption.Implementation, new TextEncoder().encode("test")); await new Setup({ parameters }).encode(setup, version); - await expect(exchangeSetup(pair.server, version, "test")).rejects.toThrow("leading-ones varint"); + await expect(exchangeSetup(pair.server, version, "test", offered())).rejects.toThrow("leading-ones varint"); }); /** An early stream that ends partway through its type died; it is skipped like an empty one. */ @@ -367,7 +368,7 @@ test("truncated uni stream before SETUP is skipped", async () => { parameters.setBytes(SetupOption.Implementation, new TextEncoder().encode("test")); await new Setup({ parameters }).encode(setup, version); - const { early } = await exchangeSetup(pair.server, version, "test"); + const { early } = await exchangeSetup(pair.server, version, "test", offered()); expect(early.length).toBe(0); }); diff --git a/js/net/src/ietf/connection.ts b/js/net/src/ietf/connection.ts index 37beef5614..892618bab8 100644 --- a/js/net/src/ietf/connection.ts +++ b/js/net/src/ietf/connection.ts @@ -83,6 +83,8 @@ export class Connection implements Established { // What the peer declared about being solicited; see {@link Ietf.solicitFromSetup}. #solicit: boolean | undefined; + // Whether our SETUP declared MoQ Solicit, which is what makes an unasked announce a fault. + #declaredSolicit: boolean; // The Hop IDs this session declared; see {@link Cluster}. #cluster?: Cluster.Hops; @@ -121,6 +123,7 @@ export class Connection implements Established { auth = false, early = [], requestWindow, + extensions, }: { url: URL; quic: WebTransport; @@ -144,12 +147,14 @@ export class Connection implements Established { * cannot negotiate the extension, as is a `peer` the peer never declared. */ cluster?: Cluster.Hops; - /** Whether the peer's SETUP offered MoQ Auth (draft-17+). */ + /** Whether MoQ Auth is negotiated (draft-17+): offered by both SETUPs. */ auth?: boolean; /** Uni streams that arrived before the peer's SETUP, type unread (v17+). */ early?: Reader[]; /** Requests the peer may hold open on drafts 14 to 16, the window our SETUP advertised (default `REQUEST_WINDOW`). */ requestWindow?: bigint; + /** What our own SETUP offered; every extension when omitted. */ + extensions?: { solicit: boolean }; }) { this.url = url; this.discovery = discovery; @@ -210,6 +215,7 @@ export class Connection implements Established { ready: this.#auth.setupAnswered(), }); this.#solicit = solicit; + this.#declaredSolicit = extensions?.solicit ?? true; this.#cluster = cluster; this.#subscriber = new Subscriber({ session: this.#session, @@ -405,18 +411,19 @@ export class Connection implements Established { Cluster.negotiated(this.#cluster), ); - // We always declare that advertisements to us must be solicited (MoQ - // Solicit), and writing the option at all proves the peer implements the - // extension, whichever value it chose. It also cannot have advertised - // before reading our SETUP, since our SETUP is what says whether + // Unless we declined it, we declare that advertisements to us must be + // solicited (MoQ Solicit), and writing the option at all proves the peer + // implements the extension, whichever value it chose. It also cannot have + // advertised before reading our SETUP, since our SETUP is what says whether // advertising unasked is allowed. So this is a bug in the peer, and a - // silent one on both sides if we tolerate it. + // silent one on both sides if we tolerate it. Without our declaration an + // unasked announce is what we invited. // // Draft-14/15 are exempt: they have no inline NAMESPACE, so a // PUBLISH_NAMESPACE request is also how a peer answers our // SUBSCRIBE_NAMESPACE there, and the message alone does not say which. const legacy = this.#session.version === Version.DRAFT_14 || this.#session.version === Version.DRAFT_15; - if (this.#solicit !== undefined && !legacy) { + if (this.#declaredSolicit && this.#solicit !== undefined && !legacy) { console.error( `unsolicited publish_namespace from a peer that implements MoQ Solicit: broadcast=${msg.trackNamespace}`, ); diff --git a/js/net/src/ietf/solicit.test.ts b/js/net/src/ietf/solicit.test.ts index be31565f46..c7937498b3 100644 --- a/js/net/src/ietf/solicit.test.ts +++ b/js/net/src/ietf/solicit.test.ts @@ -3,7 +3,7 @@ import { SetupOption, SetupOptions } from "./parameters.ts"; import { solicitFromSetup, solicitIntoSetup } from "./solicit.ts"; /** - * We declare it on every session, so a peer honoring it never sends us an unsolicited + * We declare it by default, so a peer honoring it never sends us an unsolicited * PUBLISH_NAMESPACE. The literal 1 is the wire contract, not an implementation detail. */ test("we declare that advertisements must be solicited", () => { diff --git a/js/net/src/ietf/solicit.ts b/js/net/src/ietf/solicit.ts index a01089dd4e..9ee96c3494 100644 --- a/js/net/src/ietf/solicit.ts +++ b/js/net/src/ietf/solicit.ts @@ -37,9 +37,9 @@ export function solicitFromSetup(params: SetupOptions): boolean | undefined { /** * Declare that advertisements to us must be solicited. * - * Unconditional, and true of every session: we send SUBSCRIBE_NAMESPACE for each prefix - * we want, so there is nothing an unsolicited PUBLISH_NAMESPACE could tell us that we - * will not have asked for. + * Declared unless the session's `extensions` declines it: we send SUBSCRIBE_NAMESPACE for + * each prefix we want, so there is nothing an unsolicited PUBLISH_NAMESPACE could tell us + * that we will not have asked for. * * @internal */ diff --git a/rs/moq-net/src/auth.rs b/rs/moq-net/src/auth.rs index db5028d076..e0dd718fb8 100644 --- a/rs/moq-net/src/auth.rs +++ b/rs/moq-net/src/auth.rs @@ -257,6 +257,12 @@ impl Handle { } } + /// Whether this side speaks AUTH: false on a version without it, on a side that does + /// not offer it, and once the peer turns out not to. + pub(crate) fn supported(&self) -> bool { + self.state.lock().supported + } + /// The union of every grant this side holds: `None` until the peer first /// answers a token (with a grant or a refusal), and forever on a version /// without AUTH. diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index d13d97f3c4..3f7e00141c 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -19,6 +19,7 @@ pub struct Client { cost: Option, peer_hop: Option, limits: crate::session::Limits, + extensions: setup::Extensions, } impl Client { @@ -94,6 +95,12 @@ impl Client { self } + /// Choose which moq-transport extensions to offer in SETUP. Defaults to all of them. + pub fn with_extensions(mut self, extensions: setup::Extensions) -> Self { + self.extensions = extensions; + self + } + /// Price this link, in the units the rest of the mesh uses (moq-lite-06+, and /// `moqt-17`+ via the MoQ Cluster extension). /// @@ -281,8 +288,9 @@ impl Client { // Draft-17+: SETUP is exchanged by the connection driver. // We advertise the request path in our SETUP for URL-less transports. - // The peer's SETUP decides whether AUTH is negotiated. - let auth = crate::auth::Handle::new(true); + // The peer's SETUP decides whether AUTH is negotiated. A handle that does not + // offer it keeps the option out of our SETUP and presents no credential. + let auth = crate::auth::Handle::new(self.extensions.auth); let (protocol, goaway, setup) = ietf::start(ietf::Config { runtime: runtime.clone(), limits: self.limits, @@ -301,6 +309,7 @@ impl Client { peer_declared: None, auth: auth.clone(), early_unis: Vec::new(), + extensions: self.extensions, })?; tracing::debug!(version = ?v, "connected"); @@ -381,7 +390,9 @@ impl Client { if let Some(authority) = &self.setup_authority { parameters.set_bytes(ietf::ParameterBytes::Authority, authority.clone().into_bytes()); } - ietf::solicit::into_setup(&mut parameters, ietf_encoding); + if self.extensions.solicit { + ietf::solicit::into_setup(&mut parameters, ietf_encoding); + } ietf::hidden::into_setup(&mut parameters, ietf_encoding); ietf::active_count::into_setup(&mut parameters, ietf_encoding); let parameters = parameters.encode_bytes(ietf_encoding)?; @@ -466,6 +477,7 @@ impl Client { peer_declared: Some(peer_declared), auth: auth.clone(), early_unis: Vec::new(), + extensions: self.extensions, })?; (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, setup) } diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 94c57e46ad..92d3aea4f3 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -74,12 +74,17 @@ pub struct Config { /// The session's auth handle, created before [`start`] so a server can take the /// peer's token requests during its handshake. Supports AUTH exactly when the - /// version can negotiate it; the peer's SETUP decides whether it does. + /// version can negotiate it and this side offers it; the peer's SETUP decides whether + /// it does. pub auth: crate::auth::Handle, /// Uni streams that arrived before that pre-read SETUP (see [`PeerSetup::early`]), /// classified by the session before it accepts any more. pub early_unis: Vec>, + + /// The extensions our SETUP offers. Drafts 14-16 already sent it, so there this only + /// records what it said. MoQ Auth is offered only while [`Self::auth`] supports it too. + pub extensions: crate::setup::Extensions, } pub(crate) struct Driver { @@ -126,6 +131,7 @@ where peer_declared, auth, early_unis, + extensions, } = config; // GOAWAY wiring: the public Session holds one half (drain trigger, received @@ -239,7 +245,8 @@ where tasks.clone(), goaway.going_away.clone(), ) - .with_auth(auth.clone()); + .with_auth(auth.clone()) + .with_solicit(extensions.solicit); subscriber.announces = announces; // GOAWAY send task: draft-14-16 carry GOAWAY on the shared control @@ -374,9 +381,23 @@ where let runtime = runtime.clone(); let session = session.clone(); let goaway = goaway.clone(); + let extensions = crate::setup::Extensions { + auth: extensions.auth && auth.supported(), + ..extensions + }; async move { - if let Err(err) = - run_setup(runtime, session, version, path, authority, self_origin, cost, goaway).await + if let Err(err) = run_setup( + runtime, + session, + version, + path, + authority, + self_origin, + cost, + extensions, + goaway, + ) + .await { tracing::warn!(%err, "setup send error"); } @@ -427,7 +448,8 @@ where tasks, goaway.going_away.clone(), ) - .with_auth(auth.clone()); + .with_auth(auth.clone()) + .with_solicit(extensions.solicit); // Our tokens, one Auth request each, once the peer's SETUP negotiates it. let present = auth::run_present( @@ -439,11 +461,13 @@ where version, goaway.going_away.clone(), ); - let serve = auth::Serve { + // Only a side that offered MoQ Auth answers the peer's Auth requests: one that + // arrives otherwise falls through to the protocol violation in `run_dispatch`. + let serve = auth.supported().then(|| auth::Serve { runtime: runtime.clone(), handle: auth.clone(), peer_grant, - }; + }); subscriber.announces = announces; let sub_ns_session = session.clone(); @@ -482,7 +506,7 @@ where publisher.clone(), subscriber.clone(), peer_setup.clone(), - Some(serve), + serve, version ))); let mut datagrams = std::pin::pin!(err_only(run_datagrams(session.clone(), subscriber.clone()))); @@ -714,8 +738,8 @@ fn peer_from_params(params: &ietf::Parameters, version: Version) -> Result( runtime: crate::time::Clock, @@ -725,6 +749,7 @@ async fn run_setup( authority: Option, self_origin: Hop, cost: Option, + extensions: crate::setup::Extensions, goaway: crate::goaway::Protocol, ) -> Result<(), Error> { let outer_version = crate::Version::Ietf(version); @@ -741,9 +766,13 @@ async fn run_setup( parameters.set_bytes(ietf::ParameterBytes::Authority, authority.into_bytes()); } cluster::peer_into_setup(&mut parameters, self_origin, cost, version); - solicit::into_setup(&mut parameters, version); + if extensions.solicit { + solicit::into_setup(&mut parameters, version); + } hidden::into_setup(&mut parameters, version); - auth::into_setup(&mut parameters, version); + if extensions.auth { + auth::into_setup(&mut parameters, version); + } active_count::into_setup(&mut parameters, version); let parameters = parameters.encode_bytes(version)?; @@ -1118,8 +1147,8 @@ where // negotiated, so dispatch waits for SETUP, which `run_unis` reads independently. let peer = subscriber.peer().await; - // An AUTH from a peer that did not negotiate MoQ Auth is an unknown request, which - // falls through to the protocol violation in `Bidi::serve`. + // An AUTH without MoQ Auth negotiated, offered by both sides, is an unknown request, + // which falls through to the protocol violation in `Bidi::serve`. let serve = match peer_setup.get().await.auth { true => serve, false => None, @@ -1464,6 +1493,7 @@ mod tests { }), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); @@ -1519,6 +1549,7 @@ mod tests { peer_declared: Some(peer::Peer::default()), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); let _driver = moq_net_sim::spawn(driver); @@ -1593,6 +1624,7 @@ mod tests { }), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); let driver = moq_net_sim::spawn(driver); @@ -1677,6 +1709,7 @@ mod tests { peer_declared, auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); let _driver = moq_net_sim::spawn(driver); @@ -1782,6 +1815,7 @@ mod tests { }), auth: handle.clone(), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); AuthSession { @@ -1838,6 +1872,159 @@ mod tests { assert_eq!(occurrences(log, &auth_type()), 1, "one AUTH for the setup token"); } + /// What the draft-18 SETUP that `start` writes says about MoQ Auth and MoQ Solicit. + async fn our_setup(client: bool, auth: crate::auth::Handle, extensions: crate::setup::Extensions) -> (bool, bool) { + const VERSION: Version = Version::Draft18; + + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()); + let log = session.log.clone(); + let (driver, _goaway, _) = start(Config { + runtime: crate::time::Clock::sim(), + session, + setup: None, + request_id_max: None, + limits: Default::default(), + client, + publish: None, + subscribe: None, + peer_hop: None, + cost: None, + version: VERSION, + path: None, + authority: None, + peer_setup_stream: None, + peer_declared: Some(peer::Peer::default()), + auth, + early_unis: Vec::new(), + extensions, + }) + .expect("start the session"); + + // The driver parks once SETUP is out. + moq_net_sim::timeout(std::time::Duration::from_secs(10), driver) + .await + .expect_err("the session ended instead of parking after SETUP"); + + let writes = log.writes.lock().unwrap().clone(); + let (setup, _) = setup::Setup::decode_slice(&writes, crate::Version::Ietf(VERSION)).expect("our SETUP"); + let (params, _) = ietf::Parameters::decode_slice(&setup.parameters, VERSION).expect("its options"); + ( + auth::from_setup(¶ms, VERSION) == Some(true), + solicit::from_setup(¶ms, VERSION).unwrap() == Some(true), + ) + } + + /// Each extension reaches our SETUP only while this side offers it. + #[moq_net_sim::test] + async fn a_declined_extension_stays_out_of_setup() { + let all = crate::setup::Extensions::default(); + let mut no_auth = all; + no_auth.auth = false; + let mut no_solicit = all; + no_solicit.solicit = false; + + // A client's handle supports AUTH exactly when it offers it. + assert_eq!(our_setup(true, crate::auth::Handle::new(true), all).await, (true, true)); + assert_eq!( + our_setup(true, crate::auth::Handle::new(false), no_auth).await, + (false, true) + ); + assert_eq!( + our_setup(true, crate::auth::Handle::new(true), no_solicit).await, + (true, false) + ); + } + + /// A server's handle supports AUTH only when the client offered it too, so a server + /// answering a client that did not offer it leaves the option out, whatever it offers. + #[moq_net_sim::test] + async fn a_server_offers_auth_only_to_a_client_that_did() { + let all = crate::setup::Extensions::default(); + assert_eq!( + our_setup(false, crate::auth::Handle::new(true), all).await, + (true, true) + ); + assert_eq!( + our_setup(false, crate::auth::Handle::new(false), all).await, + (false, true) + ); + } + + /// The bytes of a peer's Auth request presenting its connection credential. + async fn auth_request(version: Version) -> Vec { + let log = crate::lite::test_transport::Log::default(); + let mut writer = crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); + writer + .encode_message(&auth::Auth { + request_id: RequestId(0), + token: bytes::Bytes::new(), + }) + .await + .unwrap(); + let writes = log.writes.lock().unwrap(); + writes.clone() + } + + /// Run a draft-18 server whose client offered MoQ Auth and then sent an Auth request. + /// `offered` is whether the server offered it too. Returns how the session ended, if it + /// did, and what closed it. + async fn serve_auth_request(offered: bool) -> (Option>, Vec<(u32, String)>) { + const VERSION: Version = Version::Draft18; + + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()) + .with_incoming_bidis(vec![auth_request(VERSION).await]); + let log = session.log.clone(); + let (driver, _goaway, _) = start(Config { + runtime: crate::time::Clock::sim(), + session, + setup: None, + request_id_max: None, + limits: Default::default(), + client: false, + publish: None, + subscribe: None, + peer_hop: None, + cost: None, + version: VERSION, + path: None, + authority: None, + peer_setup_stream: None, + peer_declared: Some(peer::Peer { + auth: true, + ..Default::default() + }), + // As `Server` builds it: offered by the client, so supported iff we offer it. + auth: crate::auth::Handle::new(offered), + early_unis: Vec::new(), + extensions: crate::setup::Extensions::default(), + }) + .expect("start the session"); + + let res = moq_net_sim::timeout(std::time::Duration::from_secs(10), driver) + .await + .ok(); + (res, log.closes()) + } + + /// The Auth draft negotiates the extension only when both sides offer it, so an Auth + /// request reaching a side that declined it is a protocol violation, not a token to grant. + #[moq_net_sim::test] + async fn an_auth_request_to_a_side_that_declined_it_is_refused() { + let (res, closes) = serve_auth_request(true).await; + assert!(res.is_none(), "both offered: the request is served, {res:?}"); + assert_eq!(closes, vec![]); + + let (res, closes) = serve_auth_request(false).await; + let Some(Err(err)) = res else { + panic!("declined: the Auth request did not end the session: {res:?}"); + }; + assert_eq!(SessionError::from(&err), SessionError::ProtocolViolation); + assert_eq!( + closes, + vec![(SessionError::ProtocolViolation.to_code(), err.to_string())] + ); + } + /// The declared Hop ID must be the caller's own origin, whichever half carries it. /// /// A subscribe-only session (an ingest that publishes nothing) still routes, so @@ -1896,6 +2083,7 @@ mod tests { peer_declared: Some(peer::Peer::default()), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); @@ -1936,6 +2124,7 @@ mod tests { peer_declared: Some(peer::Peer::default()), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); @@ -2196,6 +2385,7 @@ mod tests { peer_declared: Some(peer::Peer::default()), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .unwrap() .0 @@ -2338,6 +2528,7 @@ mod tests { peer_declared: Some(peer::Peer::default()), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .unwrap(); @@ -2404,6 +2595,7 @@ mod tests { peer_declared: Some(peer::Peer::default()), auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), limits: Default::default(), }) .unwrap(); @@ -2505,6 +2697,7 @@ mod tests { peer_declared: None, auth: crate::auth::Handle::new(false), early_unis: Vec::new(), + extensions: Default::default(), }) .expect("start the session"); let driver = moq_net_sim::spawn(driver); diff --git a/rs/moq-net/src/ietf/solicit.rs b/rs/moq-net/src/ietf/solicit.rs index e02e826d05..470577094e 100644 --- a/rs/moq-net/src/ietf/solicit.rs +++ b/rs/moq-net/src/ietf/solicit.rs @@ -39,7 +39,7 @@ pub fn from_setup(params: &super::Parameters, _version: Version) -> Result { self_origin: crate::Hop, // What the peer declared in its SETUP. peer_setup: peer::PeerSetup, + // Whether our SETUP declared MoQ Solicit, which is what makes an unsolicited + // PUBLISH_NAMESPACE the peer's fault. + solicit: bool, // Local policy for what pulling from this peer costs, overriding whatever it // declared. See `cluster::link_cost`. cost: Option, @@ -688,6 +691,7 @@ where session_origin: peer_hop.unwrap_or(crate::Hop::UNKNOWN), self_origin, peer_setup, + solicit: true, cost, state: Default::default(), tasks, @@ -709,6 +713,13 @@ where self } + /// Whether our SETUP declared MoQ Solicit. Without it an unsolicited PUBLISH_NAMESPACE + /// is what we invited, even from a peer that implements the extension. + pub fn with_solicit(mut self, solicit: bool) -> Self { + self.solicit = solicit; + self + } + /// Follow the session's limit, attaching every withheld namespace a new limit /// covers. Each route's own serve task holds itself back when a limit no longer /// covers it. @@ -1188,16 +1199,20 @@ where /// Whether an incoming PUBLISH_NAMESPACE means the peer ignored our SETUP. /// - /// We always declare that advertisements to us must be solicited (MoQ Solicit), and a - /// peer that wrote the option at all proves it implements the extension, whichever - /// value it chose. It also cannot have advertised before reading our SETUP, since our - /// SETUP is what says whether advertising unasked is allowed. So this is a bug in the - /// peer, and a silent one on both sides if we tolerate it. + /// Unless [`with_solicit`](Self::with_solicit) turned it off, we declare that + /// advertisements to us must be solicited (MoQ Solicit), and a peer that wrote the + /// option at all proves it implements the extension, whichever value it chose. It also + /// cannot have advertised before reading our SETUP, since our SETUP is what says whether + /// advertising unasked is allowed. So this is a bug in the peer, and a silent one on + /// both sides if we tolerate it. /// /// Draft-14/15 are exempt: they have no inline NAMESPACE, so a PUBLISH_NAMESPACE /// request is also how a peer answers our SUBSCRIBE_NAMESPACE there, and the message /// alone does not say which it is. fn unsolicited_is_a_violation(&self, declared: Option) -> bool { + if !self.solicit { + return false; + } match self.version { Version::Draft14 | Version::Draft15 => false, _ => declared.is_some(), @@ -4671,8 +4686,13 @@ mod tests { } /// What an unsolicited advertisement means to a subscriber on `version` whose peer - /// declared `solicit`. + /// declared `solicit`, when we declared MoQ Solicit ourselves. fn unsolicited_is_a_violation(solicit: Option, version: Version) -> bool { + unsolicited_is_a_violation_to(true, solicit, version) + } + + /// The same, for a subscriber whose own SETUP declared MoQ Solicit only if `ours`. + fn unsolicited_is_a_violation_to(ours: bool, solicit: Option, version: Version) -> bool { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let session = crate::lite::test_transport::SinkSession::new(Default::default()); let peer_setup = peer::PeerSetup::default(); @@ -4695,10 +4715,23 @@ mod tests { tasks, Default::default(), ) + .with_solicit(ours) .unsolicited_is_a_violation(solicit) } - /// We always declare that advertisements to us must be solicited, so a peer that + /// A side that declined MoQ Solicit invited unsolicited advertisements, so one is + /// expected even from a peer that implements the extension. + #[moq_net_sim::test] + async fn an_announce_to_a_side_that_declined_solicit_is_fine() { + for solicit in [Some(true), Some(false), None] { + assert!( + !unsolicited_is_a_violation_to(false, solicit, Version::Draft17), + "{solicit:?}" + ); + } + } + + /// By default we declare that advertisements to us must be solicited, so a peer that /// implements the extension and announces anyway has a bug. Tolerating it is what /// keeps that bug invisible on both sides, so the session goes. /// diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 95d2651ab2..99173d5da7 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -21,6 +21,7 @@ pub struct Server { stats: stats::Session, versions: Versions, limits: crate::session::Limits, + extensions: setup::Extensions, } impl Server { @@ -72,6 +73,12 @@ impl Server { self } + /// Choose which moq-transport extensions to offer in SETUP. Defaults to all of them. + pub fn with_extensions(mut self, extensions: setup::Extensions) -> Self { + self.extensions = extensions; + self + } + /// The configured origin pair, each tagged with the stats context so the /// model attributes reads (egress) and writes (ingress) for this session. /// One shared context across both halves keeps presence and viewer counts @@ -438,8 +445,9 @@ impl Server { origin: peer_setup.declared.cluster.hop.filter(|h| *h != crate::Hop::UNKNOWN), token: peer_setup.token.clone(), assigned_hop: crate::Hop::random(), - // The client's SETUP already settled whether MoQ Auth is negotiated. - auth: crate::auth::Handle::new(peer_setup.declared.auth), + // The client's SETUP already settled whether it offered MoQ Auth, so it is + // negotiated exactly when we offer it too. + auth: crate::auth::Handle::new(peer_setup.declared.auth && self.extensions.auth), inner: Some(RequestInner { server: self.clone(), runtime, @@ -579,6 +587,7 @@ where peer_declared: Some(peer_setup.declared), auth: auth.clone(), early_unis: peer_setup.early, + extensions: server.extensions, })?; tracing::debug!(?version, "connected"); Ok(Session::new( @@ -644,7 +653,9 @@ where ietf::initial_max_request_id(server.limits.requests(), true), ); parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); - ietf::solicit::into_setup(&mut parameters, v); + if server.extensions.solicit { + ietf::solicit::into_setup(&mut parameters, v); + } ietf::hidden::into_setup(&mut parameters, v); ietf::active_count::into_setup(&mut parameters, v); parameters.encode_bytes(v)? @@ -705,6 +716,7 @@ where peer_declared: Some(peer_declared), auth: auth.clone(), early_unis: Vec::new(), + extensions: server.extensions, })?; (None, crate::driver::Protocol::Ietf(protocol), goaway, auth, setup) } diff --git a/rs/moq-net/src/setup.rs b/rs/moq-net/src/setup.rs index f8a12ef862..a84b5ee043 100644 --- a/rs/moq-net/src/setup.rs +++ b/rs/moq-net/src/setup.rs @@ -1,6 +1,8 @@ -//! The SETUP exchange, and the credential a peer may present in it. +//! The SETUP exchange, the extensions a side offers in it, and the credential a peer may +//! present in it. //! -//! Only [`Token`] is public; the SETUP messages themselves are wire internals. +//! Only [`Extensions`] and [`Token`] are public; the SETUP messages themselves are wire +//! internals. use bytes::Bytes; @@ -19,6 +21,30 @@ const SERVER_SETUP: u8 = 0x21; /// Draft-17 unified SETUP message type (varint 0x2F00) pub(crate) const SETUP_V17: u64 = 0x2F00; +/// The moq-transport Setup Options this side offers, each on by default. +/// +/// Turning one off connects as a peer that does not speak that extension. moq-lite carries +/// both in its core, so this only changes moq-transport sessions. +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(default, deny_unknown_fields)] +#[non_exhaustive] +pub struct Extensions { + /// The MoQ Auth extension: tokens presented and granted on their own stream (draft-17+). + pub auth: bool, + /// The MoQ Solicit extension: the peer answers our SUBSCRIBE_NAMESPACE rather than + /// announcing unasked. + pub solicit: bool, +} + +impl Default for Extensions { + fn default() -> Self { + Self { + auth: true, + solicit: true, + } + } +} + /// A credential a moq-transport peer presented in its SETUP's `AUTHORIZATION TOKEN` option. /// /// The transport never reads the bytes; verifying them is the application's job. diff --git a/rs/moq-net/tests/extensions.rs b/rs/moq-net/tests/extensions.rs new file mode 100644 index 0000000000..5817aa6f35 --- /dev/null +++ b/rs/moq-net/tests/extensions.rs @@ -0,0 +1,147 @@ +//! A side may decline a moq-transport extension with `setup::Extensions`, and then +//! connects as a peer that does not speak it. +//! +//! Deterministic: paused time over the mock transport. + +mod support; + +use std::time::Duration; + +use moq_net::{Error, Hop, Version, origin, setup::Extensions}; +use support::harness::{MockConnectOptions, MockPair, connect_mock}; + +/// Long enough, in virtual time, for anything in flight to reach the far side. +const SETTLE: Duration = Duration::from_secs(1); + +/// Every moq-transport draft, each of which carries the SOLICIT option. +const MOQT: [&str; 9] = [ + "moq-transport-14", + "moq-transport-15", + "moq-transport-16", + "moq-transport-17", + "moq-transport-18", + "moq-transport-19", + "moq-transport-20", + "moq-transport-21", + "moq-transport-22", +]; + +/// The first draft that negotiates MoQ Auth, and the newest. +const AUTH: [&str; 2] = ["moq-transport-17", "moq-transport-22"]; + +fn produce_origin(hop: u64) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +fn declined(f: impl FnOnce(&mut Extensions)) -> Extensions { + let mut extensions = Extensions::default(); + f(&mut extensions); + extensions +} + +fn assert_open(pair: &MockPair, context: &str) { + assert_eq!(pair.client_transport.close_reason(), None, "{context}: client closed"); + assert_eq!(pair.server_transport.close_reason(), None, "{context}: server closed"); +} + +/// Whether `path` is announced in `origin` right now. +fn announced(origin: &origin::Producer, path: &str) -> bool { + let mut announced = origin.consume().announced(); + let mut live = false; + while let Some(event) = announced.try_next() { + match event { + moq_net::announce::Event::Start(update) + | moq_net::announce::Event::Update(update) + | moq_net::announce::Event::Restart(update) => { + live |= update.prefix.as_str() == path; + } + moq_net::announce::Event::End(update) => { + live &= update.prefix.as_str() != path; + } + } + } + live +} + +/// A side that declines MoQ Solicit invites unsolicited advertisements, so its peer +/// announces unasked and that is not held against the peer, even though the peer +/// implements the extension itself. +#[moq_net_sim::test] +async fn a_side_that_declines_solicit_takes_unsolicited_announces() { + for name in MOQT { + let version: Version = name.parse().unwrap(); + for server_declines in [true, false] { + let context = format!("{name}, server declines: {server_declines}"); + let publisher = produce_origin(1); + let subscriber = produce_origin(2); + let no_solicit = declined(|e| e.solicit = false); + + let mut options = MockConnectOptions::new(version); + match server_declines { + true => { + options.client_publish = Some(publisher.consume()); + options.server_subscribe = Some(subscriber.clone()); + options.server_extensions = no_solicit; + } + false => { + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(subscriber.clone()); + options.client_extensions = no_solicit; + } + } + let pair = connect_mock(options).await; + + let broadcast = publisher.create_broadcast("room/cam").unwrap(); + broadcast.announce(Default::default()).unwrap(); + moq_net_sim::sleep(SETTLE).await; + + assert!(announced(&subscriber, "room/cam"), "{context}: never announced"); + assert_open(&pair, &context); + } + } +} + +/// A side that declines MoQ Auth leaves it un-negotiated whichever side declines: +/// neither side holds a grant, and a token is unsupported rather than presented. +#[moq_net_sim::test] +async fn a_declined_auth_extension_is_not_negotiated() { + for name in AUTH { + let version: Version = name.parse().unwrap(); + + // Control: offered by both, the client's connection credential earns a grant. + let pair = connect_mock(MockConnectOptions::new(version)).await; + let mut grant = pair.client.auth().grant(); + moq_net_sim::timeout(SETTLE, async { + while grant.peek().is_none() { + grant.changed().await.expect("grant watch ended"); + } + }) + .await + .unwrap_or_else(|_| panic!("{name}: no grant with both offering")); + + for server_declines in [true, false] { + let context = format!("{name}, server declines: {server_declines}"); + let no_auth = declined(|e| e.auth = false); + let mut options = MockConnectOptions::new(version); + match server_declines { + true => options.server_extensions = no_auth, + false => options.client_extensions = no_auth, + } + let pair = connect_mock(options).await; + moq_net_sim::sleep(SETTLE).await; + + for session in [&pair.client, &pair.server] { + assert_eq!(session.auth().grant().peek(), None, "{context}"); + let added = moq_net_sim::timeout(SETTLE, session.auth().add("token")).await; + assert!( + matches!(added, Ok(Err(Error::Unsupported))), + "{context}: {:?}", + added.map(|res| res.map(|_| ())) + ); + } + assert_open(&pair, &context); + } + } +} diff --git a/rs/moq-net/tests/support/harness.rs b/rs/moq-net/tests/support/harness.rs index 0fd6754a51..df4df72ad2 100644 --- a/rs/moq-net/tests/support/harness.rs +++ b/rs/moq-net/tests/support/harness.rs @@ -53,6 +53,10 @@ pub struct MockConnectOptions { pub latency: std::time::Duration, /// What the client may make the server's session hold, when not the default. pub server_limits: Option, + /// The extensions the client offers. + pub client_extensions: moq_net::setup::Extensions, + /// The extensions the server offers. + pub server_extensions: moq_net::setup::Extensions, /// The price the client puts on the link, when not the default. pub cost: Option, } @@ -68,6 +72,8 @@ impl MockConnectOptions { server_subscribe: None, latency: std::time::Duration::ZERO, server_limits: None, + client_extensions: Default::default(), + server_extensions: Default::default(), cost: None, } } @@ -99,7 +105,9 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { client_transport.set_latency(opts.latency); let transports = (client_transport.clone(), server_transport.clone()); - let mut client = Client::new().with_versions(opts.version.into()); + let mut client = Client::new() + .with_versions(opts.version.into()) + .with_extensions(opts.client_extensions); if let Some(cost) = opts.cost { client = client.with_cost(cost); } @@ -110,7 +118,9 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { client = client.with_subscriber(subscribe); } - let mut server = Server::new().with_versions(opts.version.into()); + let mut server = Server::new() + .with_versions(opts.version.into()) + .with_extensions(opts.server_extensions); if let Some(publish) = &opts.server_publish { server = server.with_publisher(publish); } diff --git a/rs/moq-relay/src/relay.rs b/rs/moq-relay/src/relay.rs index 97ba99c575..1049da6f30 100644 --- a/rs/moq-relay/src/relay.rs +++ b/rs/moq-relay/src/relay.rs @@ -146,6 +146,7 @@ impl Relay { let server_versions = config.listen.versions(); let handshake_timeout = config.listen.resolved_timeout(); + let server_extensions = config.listen.extensions; // Bind the QUIC workers first: they own the listen address when configured, // so the server below must not also try to bind it. @@ -280,6 +281,7 @@ impl Relay { let web = web::Web::new(auth.clone(), cluster.clone(), certificates, config.web) .with_shutdown(shutdown.clone()) .with_versions(server_versions) + .with_extensions(server_extensions) .with_timeout(handshake_timeout) .with_sessions(sessions.clone()) .bind()?; diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 86ac76a856..12d9be09de 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -194,6 +194,7 @@ pub struct Web { state: Arc, config: Config, versions: moq_net::Versions, + extensions: moq_net::setup::Extensions, health: moq_tokio::accept::Health, http_listener: Option, https_listener: Option, @@ -224,6 +225,7 @@ impl Web { state, config, versions: moq_net::Versions::all(), + extensions: Default::default(), health: moq_tokio::accept::Health::new("web"), http_listener: None, https_listener: None, @@ -286,6 +288,12 @@ impl Web { self } + /// Choose which moq-transport extensions WebSocket sessions offer in SETUP. + pub fn with_extensions(mut self, extensions: moq_net::setup::Extensions) -> Self { + self.extensions = extensions; + self + } + /// A live handle to the accept-loop health of the HTTP/HTTPS listeners, for an /// embedder that publishes it (see [`moq_tokio::accept`]). /// @@ -375,6 +383,7 @@ impl Web { }; app.layer(Extension(self.versions.clone())) + .layer(Extension(self.extensions)) .with_state(self.state.clone()) } diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index bb8d922c67..846a9ad7eb 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -29,6 +29,7 @@ pub(crate) async fn serve_ws( ConnectInfo(remote): ConnectInfo, socket_stats: Option>, Extension(versions): Extension, + Extension(extensions): Extension, State(state): State>, ) -> axum::response::Result { // If this isn't a WebSocket upgrade (e.g. a plain browser visit), serve @@ -99,6 +100,7 @@ pub(crate) async fn serve_ws( remote: remote.0, alpn, versions, + extensions, publish, subscribe, stats, @@ -117,6 +119,8 @@ struct SessionInputs { remote: SocketAddr, alpn: Option, versions: moq_net::Versions, + /// The moq-transport extensions this side offers in SETUP. + extensions: moq_net::setup::Extensions, publish: Option, subscribe: Option, stats: Session, @@ -153,6 +157,7 @@ where remote: _, alpn, versions, + extensions, publish, subscribe, stats, @@ -187,7 +192,10 @@ where // Only set the side the token actually grants. moq-net defaults the // unset side to a fresh no-op origin, which is fine for a // publish-only or subscribe-only token. - let mut server = moq_net::Server::new().with_versions(versions).with_stats(stats); + let mut server = moq_net::Server::new() + .with_versions(versions) + .with_extensions(extensions) + .with_stats(stats); if let Some(subscribe) = subscribe { server = server.with_publisher(subscribe); } @@ -976,6 +984,7 @@ mod tests { remote: "127.0.0.1:0".parse().unwrap(), alpn: Some(alpn.clone()), versions: moq_net::Versions::all(), + extensions: Default::default(), publish: None, subscribe: None, stats: Session::default(), @@ -1042,6 +1051,7 @@ mod tests { remote: "127.0.0.1:0".parse().unwrap(), alpn: Some(alpn.clone()), versions: moq_net::Versions::all(), + extensions: Default::default(), publish: None, subscribe: None, stats: Session::default(), diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 2c217564de..362e15788f 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -143,7 +143,9 @@ impl Client { let timeout = resolved.timeout; Ok(Self { - moq: moq_net::Client::new().with_versions(versions.clone()), + moq: moq_net::Client::new() + .with_versions(versions.clone()) + .with_extensions(config.extensions), #[cfg(any( feature = "noq", feature = "iroh", diff --git a/rs/moq-tokio/src/connect.rs b/rs/moq-tokio/src/connect.rs index 6589087eb8..9689ddb654 100644 --- a/rs/moq-tokio/src/connect.rs +++ b/rs/moq-tokio/src/connect.rs @@ -361,6 +361,21 @@ impl ConnectError { mod tests { use super::*; + /// The dial config offers every extension unless told otherwise, and leaves the + /// default out when serialized. + #[test] + fn extensions_default_on_and_parse_off() { + let config: Config = toml::from_str("[extensions]\nsolicit = false\n").expect("parse"); + assert!(!config.extensions.solicit); + assert!(config.extensions.auth); + assert!(toml::to_string(&config).expect("serialize").contains("solicit = false")); + assert!( + !toml::to_string(&Config::default()) + .expect("serialize") + .contains("extensions") + ); + } + #[test] fn auth_statuses_are_terminal() { assert_eq!(ConnectError::from_status_u16(401), Some(ConnectError::Unauthorized)); @@ -464,6 +479,11 @@ failover_delay = "1s" } } +/// Whether `extensions` offers every extension, the default a config leaves out. +pub(crate) fn all_extensions(extensions: &moq_net::setup::Extensions) -> bool { + *extensions == moq_net::setup::Extensions::default() +} + /// The dial side of an endpoint: where to connect and how to get there. /// /// Derives [`usage::Args`], so flatten it into a binary's own parser with @@ -609,6 +629,11 @@ pub struct Config { )] pub version: Vec, + /// The moq-transport extensions to offer in SETUP, all by default. + #[serde(default, skip_serializing_if = "crate::connect::all_extensions")] + #[usage(skip)] + pub extensions: moq_net::setup::Extensions, + /// TLS trust and client-certificate settings (`--connect-tls-*`). #[usage(flatten)] #[serde(default)] @@ -687,6 +712,7 @@ impl Default for Config { timeout: DEFAULT_TIMEOUT, timeout_arg: None, version: Vec::new(), + extensions: Default::default(), tls: Default::default(), once: None, reconnect: None, diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index 6f5c2a7fba..ee2d5fb791 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -146,6 +146,11 @@ pub struct Config { )] pub version: Vec, + /// The moq-transport extensions to offer in SETUP, all by default. + #[serde(default, skip_serializing_if = "crate::connect::all_extensions")] + #[usage(skip)] + pub extensions: moq_net::setup::Extensions, + /// Maximum time for one accepted connection to finish its handshake: the /// QUIC, WebTransport, WebSocket, or qmux one, then the MoQ SETUP, through /// [`crate::server::Request::ok`]. Defaults to 10 seconds; set to 0 to wait forever. @@ -268,6 +273,7 @@ impl Default for Config { #[cfg(all(feature = "uds", unix))] unix: Default::default(), version: Vec::new(), + extensions: Default::default(), timeout: DEFAULT_TIMEOUT, timeout_arg: None, tls: Default::default(), @@ -529,6 +535,21 @@ impl Config { #[cfg(test)] mod tests { use super::*; + + /// The listen config offers every extension unless told otherwise, and leaves the + /// default out when serialized. + #[test] + fn extensions_default_on_and_parse_off() { + let config: Config = toml::from_str("[extensions]\nauth = false\n").expect("parse"); + assert!(!config.extensions.auth); + assert!(config.extensions.solicit); + assert!(toml::to_string(&config).expect("serialize").contains("auth = false")); + assert!( + !toml::to_string(&Config::default()) + .expect("serialize") + .contains("extensions") + ); + } /// A parser wrapping the config, since it derives `Args` (see the note in /// [`crate::connect`]). #[derive(usage::Cli)] diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 8879dc2a80..1e34c6ceec 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -362,7 +362,10 @@ impl Server { iroh::listen(endpoint, &versions)?; } - let mut moq = moq_net::Server::new().with_versions(versions.clone()).with_stats(stats); + let mut moq = moq_net::Server::new() + .with_versions(versions.clone()) + .with_extensions(config.extensions) + .with_stats(stats); if let Some(publisher) = publisher { moq = moq.with_publisher(publisher); }