moqtap_codec/draft22/message.rs
1//! Draft-22 control message encoding and decoding.
2//!
3//! The one difference from draft-21 is `LOCATION_FILTER` (0x21). Its value
4//! opens with a `Location Filter Type` that names which `vi64` fields follow,
5//! and nothing on the wire states its length: the parameter is
6//! self-delimiting, like a Location, and a receiver that does not understand
7//! the type cannot find the end of it (Section 9.20.9, Table 6). See
8//! [`decode_location_filter`].
9//!
10//! Everything else is draft-21's wire, and so draft-20's:
11//! - **FETCH (0x16)** has no `Fetch Type`. Track Namespace and Track Name are
12//! inline fields of FETCH, and the range travels in the `LOCATION_FILTER`
13//! parameter (Section 9.11, Figure 16).
14//! - **PUBLISH_STATE_NOTIFY (0x22)** carries no Request ID (Section 9.10,
15//! Figure 15).
16//! - **`FILL_PARAMETERS` (0x23)** is a length-prefixed parameter carrying a
17//! nested parameter block in its own scope (Section 9.20.15).
18//! - **`INCLUDE_PROPERTIES` (0x35)** is a uint8 restricted to 0 and 1
19//! (Section 9.20.21).
20//! - **`EXPIRES` is the only parameter whose definition names `PUBLISH_OK`.**
21//! - `PUBLISH_DONE` status 0x3 and `REQUEST_ERROR` code 0x32 are unassigned.
22//!
23//! Every section, table and figure reference in this module is draft-22's own.
24
25use crate::auth_token::{AuthorizationToken, AUTH_TOKEN_PARAMETER};
26use crate::error::MAX_FULL_TRACK_NAME_LENGTH;
27pub use crate::error::{
28 CodecError, MAX_GOAWAY_URI_LENGTH, MAX_MESSAGE_LENGTH, MAX_NAMESPACE_TUPLE_SIZE,
29 MAX_REASON_PHRASE_LENGTH,
30};
31use crate::kvp::{KeyValuePair, KvpError, KvpValue, MAX_KVP_VALUE_LEN};
32use crate::types::*;
33use crate::varint::{Moqt18 as Wire, VarInt};
34use bytes::{Buf, BufMut};
35
36// ============================================================
37// Parameter encoding helpers for draft-22
38// ============================================================
39
40/// How a parameter value is encoded on the wire.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42enum ParamEncoding {
43 /// Bare varint.
44 Varint,
45 /// Single byte (uint8).
46 Uint8,
47 /// Two consecutive varints (group, object).
48 Location,
49 /// Length-prefixed bytes.
50 LengthPrefixed,
51 /// A Track Namespace as defined in draft-22 Section 8.7: a varint field
52 /// count followed by that many length-prefixed fields.
53 ///
54 /// The fifth value encoding draft-22 Section 9.20 lists: "Track Namespace:
55 /// A sequence of length-prefixed fields (Section 8.7)". TRACK_NAMESPACE_PREFIX
56 /// is the parameter that uses it; the field count is the only length the
57 /// wire carries.
58 TrackNamespaceValue,
59 /// A `Location Filter Type` followed by the zero to four `vi64` fields that
60 /// type names (draft-22 Section 9.20.9).
61 ///
62 /// None of the five encodings Section 9.20 lists fits it. The parameter
63 /// definition states its own structure, and that structure has no length:
64 /// the type is the only thing that says where the value ends.
65 LocationFilter,
66}
67
68fn param_encoding(key: u64) -> Option<ParamEncoding> {
69 match key {
70 // 0x02 = OBJECT_DELIVERY_TIMEOUT (Section 9.20.4)
71 // 0x04 = RENDEZVOUS_TIMEOUT (Section 9.20.6). Not MAX_CACHE_DURATION:
72 // that is Property Type 0x04 in the separate Properties
73 // registry (Section 16.8), a different namespace that happens
74 // to reuse the number.
75 // 0x06 = SUBGROUP_DELIVERY_TIMEOUT (Section 9.20.3)
76 // 0x08 = EXPIRES (Section 9.20.16)
77 // 0x0A = FILL_TIMEOUT (Section 9.20.5)
78 // 0x32 = NEW_GROUP_REQUEST (Section 9.20.19)
79 0x02 | 0x04 | 0x06 | 0x08 | 0x0A | 0x32 => Some(ParamEncoding::Varint),
80 // 0x10 = FORWARD (Section 9.20.18), 0x20 = SUBSCRIBER_PRIORITY,
81 // 0x22 = GROUP_ORDER, and 0x35 = INCLUDE_PROPERTIES.
82 // Section 9.20.21: "The INCLUDE_PROPERTIES parameter (Parameter Type
83 // 0x35) is a uint8."
84 0x10 | 0x20 | 0x22 | 0x35 => Some(ParamEncoding::Uint8),
85 // 0x09 = LARGEST_OBJECT. Draft-22 Section 9.20.17: "The LARGEST_OBJECT
86 // parameter (Parameter Type 0x9) is a Location." A Location is
87 // two consecutive varints (Section 9.20), with no length ahead
88 // of them. The type is odd, which on a Key-Value-Pair would mean
89 // length-prefixed; Message Parameters are not Key-Value-Pairs
90 // and the odd/even rule does not reach them.
91 0x09 => Some(ParamEncoding::Location),
92 // 0x34 = TRACK_NAMESPACE_PREFIX. Section 9.20.20: it "uses the Track
93 // Namespace encoding described in Section 8.7".
94 0x34 => Some(ParamEncoding::TrackNamespaceValue),
95 // 0x21 = LOCATION_FILTER. Section 9.20.9: "The Location Filter Type
96 // dictates which optional variable-length integer fields follow,
97 // and how they are interpreted." No length precedes the type,
98 // and the type is odd for the same reason 0x09's is: the
99 // odd/even rule is a Key-Value-Pair rule and does not reach
100 // Message Parameters.
101 0x21 => Some(ParamEncoding::LocationFilter),
102 // 0x03 = AUTHORIZATION_TOKEN
103 // 0x23 = FILL_PARAMETERS. Section 9.20.15: it "uses length-prefixed
104 // encoding", and the value is a nested parameter block rather
105 // than opaque bytes
106 // 0x25 = SUBGROUP_FILTER, 0x26 = OBJECTID_FILTER, 0x27 = PRIORITY_FILTER,
107 // 0x28 = OBJECT_PROPERTY_FILTER, 0x29 = TRACK_PROPERTY_FILTER
108 // (Range Filters, Section 8.6)
109 0x03 | 0x23 | 0x25 | 0x26 | 0x27 | 0x28 | 0x29 => Some(ParamEncoding::LengthPrefixed),
110 _ => None,
111 }
112}
113
114/// Whether `value` is inside the range draft-22 allows for a uint8-valued
115/// parameter.
116///
117/// Three of the four uint8 parameters restrict their range and say the receiver
118/// MUST close the session with PROTOCOL_VIOLATION on anything outside it:
119/// GROUP_ORDER allows only Ascending (0x1) and Descending (0x2) (Section
120/// 9.20.8), FORWARD allows only 0 and 1 (Section 9.20.18), and
121/// INCLUDE_PROPERTIES allows only 0 and 1 (Section 9.20.21).
122/// SUBSCRIBER_PRIORITY (Section 9.20.7) uses the whole 0-255 range,
123/// so it has no entry here.
124///
125/// Range-checking on decode is what makes the values usable: an application
126/// that tests `group_order == 2` for descending would otherwise treat 7 as
127/// neither ascending nor descending and carry on.
128fn uint8_value_in_range(key: u64, value: u8) -> bool {
129 match key {
130 // FORWARD (0x10)
131 0x10 => value <= 1,
132 // GROUP_ORDER (0x22)
133 0x22 => value == 1 || value == 2,
134 // INCLUDE_PROPERTIES (0x35), Section 9.20.21: "The allowed values are
135 // 0 (do not send Properties) or 1 (send Properties), and the default
136 // is 1. If an endpoint receives a value outside this range, it MUST
137 // close the session with PROTOCOL_VIOLATION."
138 0x35 => value <= 1,
139 _ => true,
140 }
141}
142
143/// AUTHORIZATION TOKEN, Parameter Type 0x03.
144const AUTHORIZATION_TOKEN: u64 = 0x03;
145
146/// Whether draft-22 lets a message carry parameter type `key` more than once.
147///
148/// Section 9.20 states the default: "Senders MUST NOT repeat the same Parameter
149/// Type in a message unless the parameter definition explicitly allows multiple
150/// instances of that type to be sent in a single message." Two definitions do.
151///
152/// * `AUTHORIZATION_TOKEN` (0x03), Section 8.9: "An Authorization Token MAY
153/// be repeated within a message as long as the combination of Token Type and
154/// Token Value are unique after resolving any aliases."
155/// * The five Range Filters (0x25 through 0x29), Section 3.3.2: "The Track Property filter parameter MAY appear multiple times in a
156/// SUBSCRIBE_TRACKS message or REQUEST_UPDATE for it. All other filter
157/// parameters MAY appear multiple times in a FETCH, SUBSCRIBE,
158/// SUBSCRIBE_TRACKS, or REQUEST_UPDATE (on a subscription, from the subscriber
159/// only) message."
160///
161/// # A zero `Type Delta` is "the same type again", not an error
162///
163/// The two rules interact, and **the draft does not say how**. Section 9.20 also
164/// requires that "Parameters MUST be serialized in ascending order by Type", so
165/// a second instance of a repeatable type produces a `Type Delta` of 0 — well
166/// formed only if the decoder reads a zero delta as a repeat rather than as a
167/// malformation. That reading is the one taken here; the alternative makes
168/// Section 3.3.2's permission unusable, because there is no other encoding for
169/// a second filter of the same type.
170///
171/// What the filters may **not** do is repeat the same (Parameter Type, SetID,
172/// Property Type) triple, and Section 3.3.2 answers that with a REQUEST_ERROR
173/// carrying INVALID_FILTER rather than with a session close. A reply an endpoint
174/// sends is not a frame a decoder refuses, so nothing here enforces it — see
175/// [`crate::range_filter`] for the reader an endpoint uses to decide.
176fn parameter_may_repeat(key: u64) -> bool {
177 matches!(key, AUTHORIZATION_TOKEN | 0x25..=0x29)
178}
179
180/// Add a delta to the previous delta-encoded key.
181///
182/// Draft-22 Section 8.3: "The previous Type value plus the Delta Type MUST NOT
183/// be greater than 2^64 - 1. If a Delta Type is received that would be too
184/// large, the Session MUST be closed with a PROTOCOL_VIOLATION." MoQT varints
185/// span the whole 64-bit range, so a peer can drive the sum past the end: a
186/// debug build panicked on the addition and a release build wrapped the key and
187/// reported the parameter under a type its sender never wrote.
188fn add_delta(prev_key: u64, delta: u64) -> Result<u64, CodecError> {
189 prev_key.checked_add(delta).ok_or(CodecError::KeyDeltaOverflow(prev_key, delta))
190}
191
192/// Hold a namespace-plus-name pair to the Full Track Name cap.
193///
194/// Draft-22 Section 8.7: "The maximum total length of a Full Track Name is
195/// 4,096 bytes. The length of a Full Track Name is computed as the sum of the
196/// Track Namespace Field Length fields and the Track Name Length field... If an
197/// endpoint receives a Track Namespace or a Full Track Name exceeding 4,096
198/// bytes, it MUST close the session with a PROTOCOL_VIOLATION."
199///
200/// The namespace half of that sentence is enforced inside the namespace decoder,
201/// which is the only place that sees a namespace with no name beside it. This is
202/// the other half, and it has to live where the two are decoded together: a
203/// namespace at 4,000 bytes and a name at 500 are each legal alone.
204fn check_full_track_name(namespace: &TrackNamespace, track_name: &[u8]) -> Result<(), CodecError> {
205 let total = namespace.field_bytes_len().saturating_add(track_name.len());
206 if total > MAX_FULL_TRACK_NAME_LENGTH {
207 return Err(CodecError::TrackNameTooLong);
208 }
209 Ok(())
210}
211
212/// Hold every AUTHORIZATION TOKEN parameter to the Token structure it names.
213///
214/// Section 8.9: "If the Token structure cannot be decoded, the receiver
215/// MUST close the Session with KEY_VALUE_FORMATTING_ERROR." That is the answer
216/// Section 8.3 gives for any Type whose value does not match the
217/// serialization that Type defines; the Token is the one structure this draft
218/// spells out, and the only parameter value in it that is more than opaque
219/// bytes.
220///
221/// Both namespaces carry the type on this draft, and both reach here.
222///
223/// A type this draft cannot name is left alone. The rule is conditional on the
224/// receiver understanding the Type, and an extension's parameter carries bytes
225/// no rule here describes.
226fn check_authorization_tokens(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
227 for parameter in parameters {
228 let key = parameter.key.into_inner();
229 if key != AUTH_TOKEN_PARAMETER {
230 continue;
231 }
232 match ¶meter.value {
233 KvpValue::Bytes(value) => {
234 AuthorizationToken::decode_moqt::<Wire>(key, value)?;
235 }
236 // Unreachable from the decoder, which picks the shape from the
237 // type and finds this one length-prefixed. A caller that built the
238 // pair in memory can still get here, and it is the same rule: the
239 // value is not the serialization the type defines.
240 KvpValue::Varint(_) => {
241 return Err(CodecError::KeyValueFormatting {
242 key,
243 detail: "its value is a bare varint where the type defines a Token structure",
244 });
245 }
246 }
247 }
248 Ok(())
249}
250
251/// The parameter types draft-22 Section 9.20.15, Table 7 permits inside a
252/// `FILL_PARAMETERS` value, in ascending order.
253///
254/// `TRACK_PROPERTY_FILTER` (0x29) is deliberately absent: a fill applies to one
255/// already-selected track, so a filter that selects tracks has nothing to do
256/// there. The draft does not say that in as many words, but it does say what
257/// happens to one that arrives anyway — "An endpoint that receives a parameter
258/// inside FILL_PARAMETERS that is not listed above MUST close the session with
259/// PROTOCOL_VIOLATION" — which is what makes the omission load-bearing rather
260/// than editorial. A relay forwarding a downstream `FILL_PARAMETERS` upstream
261/// has to strip it rather than pass it on.
262const FILL_PARAMETERS_ALLOWED: &[u64] = &[0x0A, 0x20, 0x21, 0x22, 0x25, 0x26, 0x27, 0x28];
263
264/// The LOCATION_FILTER parameter type, draft-22 Section 9.20.9.
265pub const LOCATION_FILTER: u64 = 0x21;
266
267/// The FILL_PARAMETERS parameter type, draft-22 Section 9.20.15.
268pub const FILL_PARAMETERS: u64 = 0x23;
269
270/// The `Location Filter Type` values draft-22 Section 9.20.9 assigns, named as
271/// Table 6 names them.
272pub mod location_filter_types {
273 /// 0x00, None: "no fields follow and there is no Location Filter".
274 pub const NONE: u64 = 0x00;
275 /// 0x01, Relative Start: "a relative StartGroup follows".
276 pub const RELATIVE_START: u64 = 0x01;
277 /// 0x02, Absolute Start: "StartGroup and StartObject follow".
278 pub const ABSOLUTE_START: u64 = 0x02;
279 /// 0x03, Absolute Start, Group End: "StartGroup, StartObject, and
280 /// EndGroupDelta follow".
281 pub const ABSOLUTE_START_GROUP_END: u64 = 0x03;
282 /// 0x04, Absolute Range: "StartGroup, StartObject, EndGroupDelta, and
283 /// EndObject follow".
284 pub const ABSOLUTE_RANGE: u64 = 0x04;
285 /// 0x05, Next Object: "no fields follow and it specifies the Next Object".
286 pub const NEXT_OBJECT: u64 = 0x05;
287}
288
289/// How many `vi64` fields follow a `Location Filter Type`, or `None` for a type
290/// draft-22 does not assign.
291///
292/// Section 9.20.9 gives one sentence per type and closes the list: "Any other
293/// Location Filter Type is a PROTOCOL_VIOLATION." The fields are always a
294/// prefix of `StartGroup`, `StartObject`, `EndGroupDelta`, `EndObject`, so the
295/// count is all a reader needs to know which ones arrived.
296///
297/// Types 0x00 and 0x05 both carry nothing and mean opposite things — no filter
298/// at all, and a filter that starts at the Next Object — so a caller that wants
299/// the meaning switches on the type, never on the count.
300pub fn location_filter_field_count(filter_type: u64) -> Option<usize> {
301 use location_filter_types as T;
302 match filter_type {
303 T::NONE | T::NEXT_OBJECT => Some(0),
304 T::RELATIVE_START => Some(1),
305 T::ABSOLUTE_START => Some(2),
306 T::ABSOLUTE_START_GROUP_END => Some(3),
307 T::ABSOLUTE_RANGE => Some(4),
308 _ => None,
309 }
310}
311
312/// Read one `LOCATION_FILTER` value off `buf`: the `Location Filter Type`, then
313/// exactly the fields that type names.
314///
315/// `buf` is bounded by the enclosing message body, or by the enclosing
316/// `FILL_PARAMETERS` value, and nothing narrower: the parameter carries no
317/// length of its own. A field that would run past that bound is a parameter
318/// that is not the structure its type defines, which is
319/// [`CodecError::SubscriptionFilterMalformed`] rather than a short read — the
320/// frame around it arrived whole.
321///
322/// A type outside Section 9.20.9's list is [`CodecError::InvalidFilterType`],
323/// and it is refused before any field is read: the type is the only thing that
324/// says where the value ends, so past an unknown one there is nothing a reader
325/// can do but stop.
326fn read_location_filter(buf: &mut impl Buf) -> Result<(u64, Vec<u64>), CodecError> {
327 let filter_type = VarInt::decode_moqt::<Wire>(buf)
328 .map_err(|_| CodecError::SubscriptionFilterMalformed {
329 detail: "its Location Filter Type runs past the end of the message",
330 })?
331 .into_inner();
332 let count = location_filter_field_count(filter_type)
333 .ok_or(CodecError::InvalidFilterType(filter_type))?;
334 let mut fields = Vec::with_capacity(count);
335 for _ in 0..count {
336 let field = VarInt::decode_moqt::<Wire>(buf).map_err(|_| {
337 CodecError::SubscriptionFilterMalformed {
338 detail: "a field its Location Filter Type names runs past the end of the message",
339 }
340 })?;
341 fields.push(field.into_inner());
342 }
343 Ok((filter_type, fields))
344}
345
346/// Write a `LOCATION_FILTER` value in its minimal encoding: the type, then the
347/// fields, each the shortest `vi64` that holds it.
348///
349/// The decoder stores what it read through this, so a value that arrived with
350/// a non-minimal varint anywhere in it — the type included, which Section 8.1
351/// permits like any other — re-encodes to different bytes. That is the same
352/// treatment a Location (0x09) gets, and it is what makes such a frame a
353/// non-canonical one rather than one this codec silently rewrites on relay.
354fn write_location_filter(filter_type: u64, fields: &[u64]) -> Vec<u8> {
355 let mut encoded = Vec::new();
356 VarInt::from_u64_moqt(filter_type).encode_moqt::<Wire>(&mut encoded);
357 for field in fields {
358 VarInt::from_u64_moqt(*field).encode_moqt::<Wire>(&mut encoded);
359 }
360 encoded
361}
362
363/// Decode a draft-22 `LOCATION_FILTER` value into its `Location Filter Type`
364/// and the fields that type names.
365///
366/// The value is the bytes a decoded [`KeyValuePair`] holds under type 0x21: the
367/// type and its fields, with no length ahead of them. The fields come back in
368/// wire order, `StartGroup`, `StartObject`, `EndGroupDelta`, `EndObject`, as
369/// many as [`location_filter_field_count`] gives for the type; Section 9.20.9,
370/// Table 6 says what each type means.
371///
372/// Refuses a type draft-22 does not assign with
373/// [`CodecError::InvalidFilterType`], and a value that holds fewer or more
374/// bytes than its type's fields with [`CodecError::SubscriptionFilterMalformed`].
375/// Every value the decoder stores passes; the refusals are for a value built in
376/// memory, which the encoder holds to the same rule before writing it.
377pub fn decode_location_filter(value: &[u8]) -> Result<(u64, Vec<u64>), CodecError> {
378 let mut cursor = value;
379 let (filter_type, fields) = read_location_filter(&mut cursor)?;
380 if cursor.has_remaining() {
381 return Err(CodecError::SubscriptionFilterMalformed {
382 detail: "bytes follow the fields its Location Filter Type names",
383 });
384 }
385 Ok((filter_type, fields))
386}
387
388/// Hold a decoded `LOCATION_FILTER` to the one arithmetic rule draft-22 states
389/// about it.
390///
391/// Section 9.20.9: "EndGroupDelta is delta encoded from StartGroup, but both the
392/// start and end groups are absolute, not relative to Largest Object. If
393/// StartGroup + EndGroupDelta exceeds 2^64 - 1, the endpoint MUST close the
394/// session with a PROTOCOL_VIOLATION." The sum only exists on types 0x03 and
395/// 0x04, the two that carry an `EndGroupDelta`, which are the two whose field
396/// count reaches three.
397///
398/// There is deliberately no check that the end is at or after the start.
399/// `EndGroupDelta` is unsigned and added to `StartGroup`, so the end group can
400/// never precede the start group; and within one group draft-22 states no rule
401/// about `EndObject` being below `StartObject`. Section 3.3.1 says the opposite
402/// for subscriptions — "A Location Filter on a subscription is always valid,
403/// even if it specifies a range entirely before Largest Object" — so refusing
404/// one would close sessions over a sentence that is not there.
405fn check_location_filter_fields(fields: &[u64]) -> Result<(), CodecError> {
406 if fields.len() < 3 {
407 return Ok(());
408 }
409 let start_group = fields[0];
410 let delta = fields[2];
411 if start_group.checked_add(delta).is_none() {
412 return Err(CodecError::FilterEndGroupOverflow { start_group, delta });
413 }
414 Ok(())
415}
416
417/// Decode the nested parameter block a `FILL_PARAMETERS` value carries.
418///
419/// # The value begins with a `Number of Parameters` count
420///
421/// Section 9.20.15 says the value is "a sequence of Parameters that apply to
422/// the fill fetch stream" and is "encoded as if they were Parameters for a
423/// separate message", and stops there. **This codec reads that as the
424/// full block, count included.** The draft does not state it either way, and the
425/// wrong choice desynchronises the whole outer parameter list rather than
426/// producing a recognisable error, so it is worth naming: Section 9.20 defines a
427/// parameter block as count-bounded — "Because unknown parameters cannot be
428/// skipped, the block is bounded by a parameter count rather than a length" —
429/// and the phrase *Parameters for a message* denotes that block everywhere else
430/// in Section 9. The outer length prefix is the generic length-prefixed value
431/// encoding every such parameter gets and says nothing about the value's
432/// internal structure.
433///
434/// So an empty `FILL_PARAMETERS` — the common case, a fill with every setting
435/// inherited from the subscription — is `Length = 1` carrying the single byte
436/// `0x00`, and **not** `Length = 0`.
437///
438/// # The `Type Delta` chain restarts here, and the outer chain is unaffected
439///
440/// Section 9.20.15: "The value of FILL_PARAMETERS is a separate parameter
441/// scope. Parameters inside it are not considered to appear in the enclosing
442/// message for the purposes of Section 9.20, so a Parameter Type MAY appear
443/// both in the message and inside FILL_PARAMETERS." Section 9.20 defines
444/// `Type Delta` as the difference from "the previous Parameter Type in the
445/// message", and a separate scope is not the message — so the inner chain
446/// starts from 0, and the outer parameter after `FILL_PARAMETERS` deltas from
447/// `0x23` rather than from the last inner type. **The draft states neither
448/// half explicitly**; both are decided here.
449///
450/// # What it refuses
451///
452/// A type outside Table 7 is [`CodecError::ParameterOutOfScope`], reported
453/// against `FILL_PARAMETERS` itself because the nested scope is the "message"
454/// the parameter appeared in. A type draft-22 does not define at all is
455/// [`CodecError::UnknownMessageParameter`], checked first so an unknown type is
456/// not reported as a known one in the wrong place.
457pub fn decode_fill_parameters(value: &[u8]) -> Result<Vec<KeyValuePair>, CodecError> {
458 let mut cursor = value;
459 let count = VarInt::decode_moqt::<Wire>(&mut cursor)?.into_inner() as usize;
460 let mut params = crate::types::reserve_bounded(count, &cursor);
461 // A separate scope, so the chain starts from 0 exactly as a message's does.
462 let mut prev_key: u64 = 0;
463
464 for i in 0..count {
465 let delta = VarInt::decode_moqt::<Wire>(&mut cursor)?.into_inner();
466 let abs_key = add_delta(prev_key, delta)?;
467 if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
468 return Err(CodecError::DuplicateParameter(abs_key));
469 }
470 prev_key = abs_key;
471
472 let encoding =
473 param_encoding(abs_key).ok_or(CodecError::UnknownMessageParameter(abs_key))?;
474 if !FILL_PARAMETERS_ALLOWED.contains(&abs_key) {
475 return Err(CodecError::ParameterOutOfScope {
476 key: abs_key,
477 message_type: FILL_PARAMETERS,
478 });
479 }
480
481 let value = decode_parameter_value(encoding, abs_key, &mut cursor)?;
482 params.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
483 }
484
485 // The block is count-bounded and the parameter is length-bounded, and the
486 // two have to agree. Bytes left over mean the count was short of what the
487 // sender wrote, which is the same disagreement a control message's Length
488 // field can have with its body.
489 if cursor.has_remaining() {
490 return Err(CodecError::SubscriptionFilterMalformed {
491 detail: "its parameter count leaves bytes unread inside FILL_PARAMETERS",
492 });
493 }
494 check_location_filters(¶ms)?;
495 Ok(params)
496}
497
498/// Hold every LOCATION_FILTER and FILL_PARAMETERS parameter to the structure
499/// its own type names.
500///
501/// Applied on both sides. A value the decoder refuses is one the peer must
502/// close the session over, so writing it is a way to end a session rather than
503/// a way to ask for anything.
504///
505/// The two are checked together because `FILL_PARAMETERS` can carry a
506/// `LOCATION_FILTER` of its own, and a filter that is malformed one level down
507/// is exactly as malformed. [`decode_fill_parameters`] recurses back into this
508/// function for that reason; the nesting bottoms out because Table 7 does not
509/// list `FILL_PARAMETERS` inside itself.
510///
511/// The values are otherwise decoded and discarded. What is kept is the refusal
512/// — each stays on its parameter as bytes, so a caller reads one through
513/// [`decode_location_filter`] or [`decode_fill_parameters`] when it wants the
514/// structure rather than the frame.
515///
516/// On decode the `LOCATION_FILTER` half can only find the overflow: the
517/// decoder read the value by its type, so its shape is already right. On
518/// encode it finds everything, because a value built in memory can name one
519/// type and carry another type's fields, and the encoder writes the value with
520/// no length that would let a receiver notice.
521fn check_location_filters(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
522 for parameter in parameters {
523 let key = parameter.key.into_inner();
524 if key != LOCATION_FILTER && key != FILL_PARAMETERS {
525 continue;
526 }
527 match ¶meter.value {
528 KvpValue::Bytes(value) if key == LOCATION_FILTER => {
529 check_location_filter_fields(&decode_location_filter(value)?.1)?;
530 }
531 KvpValue::Bytes(value) => {
532 decode_fill_parameters(value)?;
533 }
534 // Unreachable from the decoder, which stores both of these as
535 // bytes. A caller that built the pair in memory can still get here,
536 // and it is the same rule.
537 KvpValue::Varint(_) => {
538 return Err(CodecError::SubscriptionFilterMalformed {
539 detail: "its value is a bare varint where the type defines a structure",
540 });
541 }
542 }
543 }
544 Ok(())
545}
546
547/// Read one parameter's value in the shape its type names.
548///
549/// Shared by the message's own parameter block and by the nested block inside
550/// `FILL_PARAMETERS`, so a type cannot be read one way in a message and another
551/// way in a fill.
552fn decode_parameter_value(
553 encoding: ParamEncoding,
554 abs_key: u64,
555 buf: &mut impl Buf,
556) -> Result<KvpValue, CodecError> {
557 Ok(match encoding {
558 ParamEncoding::Varint => KvpValue::Varint(VarInt::decode_moqt::<Wire>(buf)?),
559 ParamEncoding::Uint8 => {
560 if buf.remaining() < 1 {
561 return Err(CodecError::UnexpectedEnd);
562 }
563 let byte = buf.get_u8();
564 if !uint8_value_in_range(abs_key, byte) {
565 return Err(CodecError::ParameterValueOutOfRange {
566 key: abs_key,
567 value: byte as u64,
568 });
569 }
570 KvpValue::Varint(VarInt::from_u64_moqt(byte as u64))
571 }
572 ParamEncoding::Location => {
573 let group = VarInt::decode_moqt::<Wire>(buf)?;
574 let object = VarInt::decode_moqt::<Wire>(buf)?;
575 let mut encoded = Vec::new();
576 group.encode_moqt::<Wire>(&mut encoded);
577 object.encode_moqt::<Wire>(&mut encoded);
578 KvpValue::Bytes(encoded)
579 }
580 ParamEncoding::LengthPrefixed => {
581 let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
582 KvpValue::Bytes(read_bytes(buf, len)?)
583 }
584 ParamEncoding::LocationFilter => {
585 let (filter_type, fields) = read_location_filter(buf)?;
586 KvpValue::Bytes(write_location_filter(filter_type, &fields))
587 }
588 ParamEncoding::TrackNamespaceValue => {
589 // A prefix of zero fields is legal: Section 2.4.1 puts a Track
590 // Namespace at "between 0 and 32 Track Namespace Fields", and
591 // an empty prefix matches every namespace.
592 let ns = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
593 let mut encoded = Vec::new();
594 ns.encode_moqt::<Wire>(&mut encoded);
595 KvpValue::Bytes(encoded)
596 }
597 })
598}
599
600/// Decode a count-prefixed list of parameters with delta-encoded types.
601fn decode_parameters(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
602 let count = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
603 let mut params = crate::types::reserve_bounded(count, buf);
604 let mut prev_key: u64 = 0;
605
606 for i in 0..count {
607 let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
608 let abs_key = add_delta(prev_key, delta)?;
609 // Types ascend, so a repeat is always a zero delta against the
610 // parameter before it. Draft-22 Section 9.20: "Receivers SHOULD check
611 // that there are no unexpected duplicate parameters and close the
612 // session with PROTOCOL_VIOLATION if found." Downstream code that scans
613 // the list for a key takes whichever copy it meets first, so two
614 // implementations reading one frame can pick opposite values.
615 //
616 // "Unexpected" is what `parameter_may_repeat` reads: a zero delta on a
617 // type whose own definition permits repeats is the second instance,
618 // which is the only encoding such an instance has.
619 if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
620 return Err(CodecError::DuplicateParameter(abs_key));
621 }
622 prev_key = abs_key;
623
624 // Section 9.20: "All Message Parameters MUST be defined in the
625 // negotiated version of MOQT or negotiated via Setup Options. An
626 // endpoint that receives an unknown Message Parameter MUST close the
627 // session with PROTOCOL_VIOLATION. Because the receiver has to
628 // understand every Message Parameter, there is no need for a mechanism
629 // to skip unknown parameters." Because unknown parameters
630 // cannot be skipped, the block is bounded by a parameter count rather
631 // than a length.
632 //
633 // The table this consults is the registry's, so a type it cannot name
634 // is one this draft does not define. Reporting it as an ordinary
635 // malformation would leave the rule enforced against the frame and
636 // invisible to the session.
637 let encoding =
638 param_encoding(abs_key).ok_or(CodecError::UnknownMessageParameter(abs_key))?;
639
640 let value = decode_parameter_value(encoding, abs_key, buf)?;
641
642 params.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
643 }
644 check_authorization_tokens(¶ms)?;
645 check_location_filters(¶ms)?;
646 Ok(params)
647}
648
649/// Whether `bytes` is exactly the wire form of a Location — two consecutive
650/// varints and nothing after them.
651///
652/// `decode_parameters` builds this value by reading two varints and
653/// re-serialising them, so every value it produces satisfies this. A value
654/// built in memory need not, and the encode arm writes these bytes verbatim
655/// because a Location carries no length of its own. Without this check a
656/// caller could hand over one varint, or three, and the codec would put a
657/// frame on the wire that its own decoder answers with an error.
658fn is_location_value(bytes: &[u8]) -> bool {
659 let mut buf = bytes;
660 VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
661 && VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
662 && !buf.has_remaining()
663}
664
665/// Whether `bytes` is exactly the wire form of a Track Namespace, with
666/// nothing after it. The same reasoning as [`is_location_value`]: the value
667/// goes out verbatim, so it has to be something this draft can read back.
668fn is_track_namespace_value(bytes: &[u8]) -> bool {
669 let mut buf = bytes;
670 TrackNamespace::decode_allow_empty_moqt::<Wire>(&mut buf).is_ok() && !buf.has_remaining()
671}
672
673/// Encode a count-prefixed list of parameters with delta-encoded types.
674///
675/// Errors with [`CodecError::InvalidField`] on a uint8-valued parameter whose
676/// value [`decode_parameters`] would refuse, so the two directions accept the
677/// same set of frames.
678///
679/// The check is not a mirror added for tidiness. A uint8 parameter's value is
680/// written as one octet, and a value that does not fit one is otherwise
681/// truncated to its low byte: GROUP_ORDER 258 becomes the byte 0x02, which is
682/// Descending — a well-formed frame carrying a value the caller never asked
683/// for, and one no receiver could tell from a genuine Descending. Refusing is
684/// the only outcome that does not silently rewrite the message.
685///
686/// The two structure rules are here for a plainer reason. A value under a type
687/// that defines a structure and is not that structure — a Token, a filter — is
688/// one the receiver must close the session over, so writing it is not a way to
689/// send it; the sender's first sign of trouble would be the session going.
690fn encode_parameters(params: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
691 check_authorization_tokens(params)?;
692 check_location_filters(params)?;
693 VarInt::from_usize(params.len()).encode_moqt::<Wire>(buf);
694 let mut prev_key: u64 = 0;
695
696 for (i, p) in params.iter().enumerate() {
697 let abs_key = p.key.into_inner();
698 // The delta is a difference, so a descending pair wraps the subtraction
699 // into a nine-byte delta the peer resolves to an unrelated key, and a
700 // repeated type is a frame `decode_parameters` refuses. Both are
701 // refused here so the two directions accept the same set of frames.
702 let delta = abs_key
703 .checked_sub(prev_key)
704 .ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
705 if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
706 return Err(CodecError::DuplicateParameter(abs_key));
707 }
708 prev_key = abs_key;
709 VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
710
711 // The same maximum the decoder below applies, and the same one this
712 // draft's Setup Option encoder has always applied: "The maximum length
713 // of a value is 2^16-1 bytes. If an endpoint receives a length larger
714 // than the maximum, it MUST close the session with a PROTOCOL_VIOLATION."
715 // A value past it is one the peer must end the session over, so writing
716 // it is not a way to send it.
717 //
718 // Hoisted above the shape table rather than repeated inside it: a
719 // Location is bytes as well, and one past the maximum is not a Location.
720 if let KvpValue::Bytes(b) = &p.value {
721 if b.len() > MAX_KVP_VALUE_LEN {
722 return Err(KvpError::ValueTooLong(b.len()).into());
723 }
724 }
725
726 let encoding = param_encoding(abs_key);
727 match (&p.value, encoding) {
728 (KvpValue::Varint(v), Some(ParamEncoding::Varint)) => {
729 v.encode_moqt::<Wire>(buf);
730 }
731 (KvpValue::Varint(v), Some(ParamEncoding::Uint8)) => {
732 let raw = v.into_inner();
733 let byte = u8::try_from(raw).map_err(|_| CodecError::InvalidField)?;
734 if !uint8_value_in_range(abs_key, byte) {
735 return Err(CodecError::ParameterValueOutOfRange {
736 key: abs_key,
737 value: byte as u64,
738 });
739 }
740 buf.put_u8(byte);
741 }
742 // These values are already stored in their own wire form — two
743 // varints for a Location, a field count and its fields for a Track
744 // Namespace, a type and its fields for a Location Filter — so they
745 // go out as they are. Adding a length here is the bug these arms
746 // exist to avoid.
747 (KvpValue::Bytes(b), Some(ParamEncoding::Location)) => {
748 if !is_location_value(b) {
749 return Err(CodecError::InvalidField);
750 }
751 buf.put_slice(b);
752 }
753 (KvpValue::Bytes(b), Some(ParamEncoding::TrackNamespaceValue)) => {
754 if !is_track_namespace_value(b) {
755 return Err(CodecError::InvalidField);
756 }
757 buf.put_slice(b);
758 }
759 // `check_location_filters` above has already held the value to
760 // its own type, which is the only check there is room for: the
761 // receiver finds the end of these bytes by reading the type, so a
762 // value whose fields disagree with its type desynchronises every
763 // parameter after it rather than failing on its own.
764 (KvpValue::Bytes(b), Some(ParamEncoding::LocationFilter)) => {
765 buf.put_slice(b);
766 }
767 (KvpValue::Bytes(b), Some(ParamEncoding::LengthPrefixed)) => {
768 VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
769 buf.put_slice(b);
770 }
771 _ => {
772 // Fallback: encode as KVP even/odd
773 match &p.value {
774 KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
775 KvpValue::Bytes(b) => {
776 VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
777 buf.put_slice(b);
778 }
779 }
780 }
781 }
782 }
783 Ok(())
784}
785
786/// Decode delta-encoded KVPs with even/odd convention (for setup options
787/// and track properties). Read until buffer is exhausted.
788fn decode_kvp_delta(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
789 let mut pairs = Vec::new();
790 let mut prev_key: u64 = 0;
791
792 while buf.has_remaining() {
793 let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
794 let abs_key = add_delta(prev_key, delta)?;
795 prev_key = abs_key;
796
797 let value = if abs_key.is_multiple_of(2) {
798 let v = VarInt::decode_moqt::<Wire>(buf)?;
799 KvpValue::Varint(v)
800 } else {
801 let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
802 // Draft-22 Section 8.3: "The maximum length of a value is 2^16-1
803 // bytes. If an endpoint receives a length larger than the maximum,
804 // it MUST close the session with a PROTOCOL_VIOLATION." The
805 // standalone `KeyValuePair::decode` already enforces this; stating
806 // it here too means the two readers of the same wire shape answer
807 // the same way, rather than this one leaning on the caller having
808 // clipped the buffer to a control message first.
809 if len > MAX_KVP_VALUE_LEN {
810 return Err(KvpError::ValueTooLong(len).into());
811 }
812 let data = read_bytes(buf, len)?;
813 KvpValue::Bytes(data)
814 };
815
816 pairs.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
817 }
818 Ok(pairs)
819}
820
821/// Encode delta-encoded KVPs with even/odd convention.
822///
823/// Refuses a list that is not in ascending order by type, for the same reason
824/// [`encode_parameters`] does: the delta is a difference, and a descending pair
825/// wraps it into a nine-byte delta the peer resolves to an unrelated key.
826fn encode_kvp_delta(pairs: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
827 let mut prev_key: u64 = 0;
828 for p in pairs {
829 let abs_key = p.key.into_inner();
830 let delta = abs_key
831 .checked_sub(prev_key)
832 .ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
833 prev_key = abs_key;
834 VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
835 match &p.value {
836 KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
837 KvpValue::Bytes(b) => {
838 if b.len() > MAX_KVP_VALUE_LEN {
839 return Err(KvpError::ValueTooLong(b.len()).into());
840 }
841 VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
842 buf.put_slice(b);
843 }
844 }
845 }
846 Ok(())
847}
848
849/// Immutable Properties, Property Type 0xB.
850///
851/// Section 10.7: Immutable Properties are "a Track or Object Property that
852/// contains a sequence of Key-Value-Pairs (see Figure 3) that are themselves
853/// Track or Object Properties, respectively". The Type is odd, so its value is
854/// length-prefixed bytes, and those bytes are another delta-typed run starting
855/// from 0.
856const IMMUTABLE_PROPERTIES: u64 = 0x0B;
857
858/// Whether `value` is inside the range draft-22 allows for a Track Property
859/// type that restricts one.
860///
861/// Two types do, and each answers anything outside its range with a session
862/// close. DEFAULT_PUBLISHER_GROUP_ORDER (0x22), Section 10.5: "The allowed
863/// values are Ascending (0x1) or Descending (0x2). If an endpoint receives a
864/// value outside this range, it MUST close the session with
865/// PROTOCOL_VIOLATION." DYNAMIC_GROUPS (0x30), Section 10.6: "The allowed
866/// values are 0 or 1... If an endpoint receives a value larger than 1, it MUST
867/// close the session with PROTOCOL_VIOLATION."
868///
869/// Both are Track Properties, so the list they arrive in is the one carried by
870/// a control message rather than the properties on an object.
871///
872/// DEFAULT_PUBLISHER_PRIORITY (0x0E) is not here. Section 10.4 says
873/// "Priorities above 255 are invalid" and stops, where the two above name a
874/// consequence in the next clause. A range stated without one is not a close.
875///
876/// The numbers belong to the Property registry and not the Message Parameter
877/// one. Type 0x22 is GROUP_ORDER as a parameter and
878/// DEFAULT_PUBLISHER_GROUP_ORDER as a property, and the two happen to permit the
879/// same pair of values while meaning different things — one subscriber's
880/// preference against a property of the track. Reading either table for the
881/// other's types would be right by accident here and wrong at the next entry.
882fn track_property_value_in_range(key: u64, value: u64) -> bool {
883 match key {
884 // DEFAULT_PUBLISHER_GROUP_ORDER (0x22)
885 0x22 => value == 1 || value == 2,
886 // DYNAMIC_GROUPS (0x30)
887 0x30 => value <= 1,
888 _ => true,
889 }
890}
891
892/// Refuse a Track Property whose value falls outside the range its type allows,
893/// wherever in the list it is carried.
894///
895/// # Inside Immutable Properties as well as beside them
896///
897/// The list is walked one level down through Immutable Properties, whose
898/// contents Section 10.7 defines as properties themselves. The draft asks for
899/// this in as many words: "When looking for the value of a property, processors
900/// MUST search both the mutable properties and the contents of Immutable
901/// Properties." A check applied only to the outer list is one a peer opts out of
902/// by moving a pair inside the block, and the block is where an Original
903/// Publisher puts what a relay must not rewrite — which is where a track's group
904/// order and dynamic-group support belong.
905///
906/// Bytes under 0xB that do not parse as a Key-Value-Pair run are left alone
907/// rather than refused. Section 10.7 says relays "MAY decode and view the
908/// Properties in the Key-Value-Pairs", which is a permission and not a
909/// requirement, so a block this codec cannot read is carried to the caller
910/// intact instead of ending the session.
911fn check_track_property_values(properties: &[KeyValuePair]) -> Result<(), CodecError> {
912 for property in properties {
913 let key = property.key.into_inner();
914 match &property.value {
915 KvpValue::Varint(value) => {
916 let value = value.into_inner();
917 if !track_property_value_in_range(key, value) {
918 return Err(CodecError::TrackPropertyValueOutOfRange { key, value });
919 }
920 }
921 KvpValue::Bytes(bytes) if key == IMMUTABLE_PROPERTIES => {
922 let mut inner = &bytes[..];
923 // A block that is not a Key-Value-Pair run is skipped rather
924 // than refused. See the note above: reading inside it is a
925 // permission, so one that cannot be read is carried.
926 //
927 // Skipped means this block and only this block. The rule the
928 // draft states here is about the block whose pairs will not
929 // parse, and says nothing about its neighbours; ending the
930 // whole walk would let a peer keep an out-of-range property
931 // from being looked at by putting an unparseable block in
932 // front of it.
933 if let Ok(nested) = decode_kvp_delta(&mut inner) {
934 check_track_property_values(&nested)?;
935 }
936 }
937 KvpValue::Bytes(_) => {}
938 }
939 }
940 Ok(())
941}
942
943/// Decode the Track Properties that fill the tail of a control message.
944///
945/// [`decode_kvp_delta`] with the Property registry's value rules applied. The
946/// two are separate because that function also reads Setup Options, which are a
947/// third namespace numbering its entries independently of this one.
948fn decode_track_properties(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
949 let properties = decode_kvp_delta(buf)?;
950 check_track_property_values(&properties)?;
951 Ok(properties)
952}
953
954/// Encode a control message's Track Properties.
955///
956/// Held to the same value ranges as the decoder. A value this codec refuses to
957/// read is one it must not write: the peer that receives it is required to close
958/// the session, so the sender's first sign of trouble would be the session
959/// going.
960fn encode_track_properties(
961 properties: &[KeyValuePair],
962 buf: &mut impl BufMut,
963) -> Result<(), CodecError> {
964 check_track_property_values(properties)?;
965 encode_kvp_delta(properties, buf)
966}
967
968/// The Setup Option types this draft defines.
969///
970/// Section 9.1 assigns PATH, AUTHORIZATION TOKEN, MAX_AUTH_TOKEN_CACHE_SIZE, AUTHORITY,
971/// MAX_FILTER_RANGES, MOQT_IMPLEMENTATION and MAX_REQUEST_UPDATES.
972///
973/// The list exists for one rule and one direction. Section 9.1: "Receivers
974/// MUST allow duplicates of unknown Setup Options." A receiver may therefore
975/// refuse a repeat only of a type it can name, and an option outside this list
976/// is one an extension defined and this codec has no business closing a session
977/// over. Nothing else reads it - unknown options are still decoded and carried,
978/// as "Receivers MUST ignore unrecognized Setup Options" requires.
979const KNOWN_SETUP_OPTIONS: &[u64] = &[0x01, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08];
980
981/// The one Setup Option whose definition allows more than one instance.
982///
983/// Section 9.1.4: "The AUTHORIZATION TOKEN Setup Option (Option Type 0x03)
984/// is functionally equivalent to the AUTHORIZATION TOKEN message parameter...
985/// The endpoint can specify one or more tokens in SETUP that the peer can use to
986/// authorize MOQT session establishment." That is the "unless the option
987/// definition explicitly allows multiple instances" carve-out, and it is the
988/// only one on this draft.
989const REPEATABLE_SETUP_OPTION: u64 = 0x03;
990
991/// Decode the Setup Options of a SETUP message.
992///
993/// Section 9.1: "Senders MUST NOT repeat the same Option Type in a message
994/// unless the option definition explicitly allows multiple instances. Receivers
995/// MUST allow duplicates of unknown Setup Options."
996///
997/// The second sentence is why this is not the mirror of
998/// [`encode_setup_options`]: a repeat of a type this draft names is refused, and
999/// a repeat of any other type is carried. Types ascend and are delta-encoded, so
1000/// a repeat is always a zero delta against the option before it.
1001fn decode_setup_options(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
1002 let options = decode_kvp_delta(buf)?;
1003 for (i, option) in options.iter().enumerate() {
1004 let key = option.key.into_inner();
1005 if key == REPEATABLE_SETUP_OPTION || !KNOWN_SETUP_OPTIONS.contains(&key) {
1006 continue;
1007 }
1008 if options[..i].iter().any(|earlier| earlier.key == option.key) {
1009 return Err(CodecError::DuplicateParameter(key));
1010 }
1011 }
1012 check_authorization_tokens(&options)?;
1013 Ok(options)
1014}
1015
1016/// Encode the Setup Options of a SETUP message.
1017///
1018/// The sender's half of the same sentence, and it is the wider half: "Senders
1019/// MUST NOT repeat the same Option Type in a message" names no exception for
1020/// types the sender does not recognise, so every repeat is refused here except
1021/// the one the draft allows. A caller holding an option this codec has never
1022/// heard of still may not send it twice.
1023///
1024/// The token is in this namespace as well, and is held to its structure here for
1025/// the reason [`encode_parameters`] gives.
1026fn encode_setup_options(options: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
1027 check_authorization_tokens(options)?;
1028 for (i, option) in options.iter().enumerate() {
1029 if option.key.into_inner() == REPEATABLE_SETUP_OPTION {
1030 continue;
1031 }
1032 if options[..i].iter().any(|earlier| earlier.key == option.key) {
1033 return Err(CodecError::DuplicateParameter(option.key.into_inner()));
1034 }
1035 }
1036 encode_kvp_delta(options, buf)
1037}
1038
1039// ============================================================
1040// Message Types
1041// ============================================================
1042
1043#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1044#[repr(u64)]
1045pub enum MessageType {
1046 RequestUpdate = 0x02,
1047 Subscribe = 0x03,
1048 SubscribeOk = 0x04,
1049 RequestError = 0x05,
1050 PublishNamespace = 0x06,
1051 /// REQUEST_OK (0x07). Draft-22 Section 1.5 uses PUBLISH_OK as a shorthand
1052 /// for a REQUEST_OK sent in response to a PUBLISH.
1053 RequestOk = 0x07,
1054 Namespace = 0x08,
1055 PublishDone = 0x0B,
1056 TrackStatus = 0x0D,
1057 NamespaceDone = 0x0E,
1058 PublishSkipped = 0x0F,
1059 GoAway = 0x10,
1060 Fetch = 0x16,
1061 FetchOk = 0x18,
1062 Publish = 0x1D,
1063 /// PUBLISH_STATE_NOTIFY (0x22), new in draft-20 (Section 9.10).
1064 PublishStateNotify = 0x22,
1065 /// SUBSCRIBE_NAMESPACE (renumbered to 0x50 in draft-18).
1066 SubscribeNamespace = 0x50,
1067 /// SUBSCRIBE_TRACKS (new message in draft-18).
1068 SubscribeTracks = 0x51,
1069 Setup = 0x2F00,
1070}
1071
1072impl MessageType {
1073 pub fn from_id(id: u64) -> Option<Self> {
1074 match id {
1075 0x02 => Some(MessageType::RequestUpdate),
1076 0x03 => Some(MessageType::Subscribe),
1077 0x04 => Some(MessageType::SubscribeOk),
1078 0x05 => Some(MessageType::RequestError),
1079 0x06 => Some(MessageType::PublishNamespace),
1080 0x07 => Some(MessageType::RequestOk),
1081 0x08 => Some(MessageType::Namespace),
1082 0x0B => Some(MessageType::PublishDone),
1083 0x0D => Some(MessageType::TrackStatus),
1084 0x0E => Some(MessageType::NamespaceDone),
1085 0x0F => Some(MessageType::PublishSkipped),
1086 0x10 => Some(MessageType::GoAway),
1087 0x16 => Some(MessageType::Fetch),
1088 0x18 => Some(MessageType::FetchOk),
1089 0x1D => Some(MessageType::Publish),
1090 0x22 => Some(MessageType::PublishStateNotify),
1091 0x50 => Some(MessageType::SubscribeNamespace),
1092 0x51 => Some(MessageType::SubscribeTracks),
1093 0x2F00 => Some(MessageType::Setup),
1094 _ => None,
1095 }
1096 }
1097
1098 pub fn id(&self) -> u64 {
1099 *self as u64
1100 }
1101
1102 /// This type's name in the shared vector corpus: the `message_type` its
1103 /// draft's `codec/messages/*.json` files carry, in `snake_case`.
1104 pub fn name(&self) -> &'static str {
1105 match self {
1106 MessageType::RequestUpdate => "request_update",
1107 MessageType::Subscribe => "subscribe",
1108 MessageType::SubscribeOk => "subscribe_ok",
1109 MessageType::RequestError => "request_error",
1110 MessageType::PublishNamespace => "publish_namespace",
1111 MessageType::RequestOk => "request_ok",
1112 MessageType::Namespace => "namespace",
1113 MessageType::PublishDone => "publish_done",
1114 MessageType::TrackStatus => "track_status",
1115 MessageType::NamespaceDone => "namespace_done",
1116 MessageType::PublishSkipped => "publish_skipped",
1117 MessageType::GoAway => "goaway",
1118 MessageType::Fetch => "fetch",
1119 MessageType::FetchOk => "fetch_ok",
1120 MessageType::Publish => "publish",
1121 MessageType::PublishStateNotify => "publish_state_notify",
1122 MessageType::SubscribeNamespace => "subscribe_namespace",
1123 MessageType::SubscribeTracks => "subscribe_tracks",
1124 MessageType::Setup => "setup",
1125 }
1126 }
1127}
1128
1129// ============================================================
1130// Session Lifecycle Messages
1131// ============================================================
1132
1133/// Unified SETUP (0x2F00).
1134#[derive(Debug, Clone, PartialEq, Eq)]
1135pub struct Setup {
1136 pub options: Vec<KeyValuePair>,
1137}
1138
1139/// GOAWAY (0x10). Draft-22's GOAWAY has no Request ID field, so the
1140/// control-stream and request-stream forms are identical on the wire.
1141/// Draft-18 is the one draft that carries the field, and only when the
1142/// message is sent on the control stream.
1143#[derive(Debug, Clone, PartialEq, Eq)]
1144pub struct GoAway {
1145 pub new_session_uri: Vec<u8>,
1146 pub timeout: VarInt,
1147}
1148
1149// ============================================================
1150// Consolidated Response Messages
1151// ============================================================
1152
1153/// REQUEST_OK (0x07). Used as a generic OK response and as the alias for
1154/// PUBLISH_OK / REQUEST_UPDATE_OK / TRACK_STATUS_OK / SUBSCRIBE_NAMESPACE_OK
1155/// / PUBLISH_NAMESPACE_OK.
1156///
1157/// `track_properties` is only populated for TRACK_STATUS_OK; for every
1158/// other shape it MUST be empty (length implicit from the message length).
1159#[derive(Debug, Clone, PartialEq, Eq)]
1160pub struct RequestOk {
1161 pub parameters: Vec<KeyValuePair>,
1162 pub track_properties: Vec<KeyValuePair>,
1163}
1164
1165/// Optional Redirect structure carried in REQUEST_ERROR with code 0x34.
1166#[derive(Debug, Clone, PartialEq, Eq)]
1167pub struct Redirect {
1168 pub connect_uri: Vec<u8>,
1169 pub track_namespace: TrackNamespace,
1170 pub track_name: Vec<u8>,
1171}
1172
1173/// REQUEST_ERROR (0x05). Adds an optional Redirect structure when
1174/// `error_code` is REDIRECT (0x34).
1175#[derive(Debug, Clone, PartialEq, Eq)]
1176pub struct RequestError {
1177 pub error_code: VarInt,
1178 pub retry_interval: VarInt,
1179 pub reason_phrase: Vec<u8>,
1180 pub redirect: Option<Redirect>,
1181}
1182
1183/// REQUEST_ERROR error codes with dedicated meaning.
1184///
1185/// Drafts 16 through 18 define DUPLICATE_SUBSCRIPTION (0x19) among these
1186/// codes; drafts 19 and 20 do not, because Section 3.1 lets an endpoint hold
1187/// multiple concurrent subscriptions to the same Track, each under its own
1188/// Request ID.
1189pub mod request_error_codes {
1190 /// A Mandatory Track Property the receiver does not understand.
1191 pub const UNSUPPORTED_EXTENSION: u64 = 0x33;
1192 /// Response carries a [`super::Redirect`] structure.
1193 pub const REDIRECT: u64 = 0x34;
1194 /// New in draft-20: SUBSCRIBE_TRACKS filter parameters conflict among too
1195 /// many subscribers to aggregate the subscription upstream.
1196 pub const CONFLICTING_FILTERS: u64 = 0x35;
1197 /// New in draft-20: a Range Filter parameter is invalid or exceeds
1198 /// MAX_FILTER_RANGES.
1199 pub const INVALID_FILTER: u64 = 0x36;
1200}
1201
1202// ============================================================
1203// Subscribe Messages
1204// ============================================================
1205
1206#[derive(Debug, Clone, PartialEq, Eq)]
1207pub struct Subscribe {
1208 pub request_id: VarInt,
1209 pub track_namespace: TrackNamespace,
1210 pub track_name: Vec<u8>,
1211 pub parameters: Vec<KeyValuePair>,
1212}
1213
1214/// SUBSCRIBE_OK (0x04).
1215#[derive(Debug, Clone, PartialEq, Eq)]
1216pub struct SubscribeOk {
1217 pub track_alias: VarInt,
1218 pub parameters: Vec<KeyValuePair>,
1219 pub track_properties: Vec<KeyValuePair>,
1220}
1221
1222#[derive(Debug, Clone, PartialEq, Eq)]
1223pub struct RequestUpdate {
1224 pub request_id: VarInt,
1225 pub parameters: Vec<KeyValuePair>,
1226}
1227
1228// ============================================================
1229// Publish Messages
1230// ============================================================
1231
1232#[derive(Debug, Clone, PartialEq, Eq)]
1233pub struct Publish {
1234 pub request_id: VarInt,
1235 pub track_namespace: TrackNamespace,
1236 pub track_name: Vec<u8>,
1237 pub track_alias: VarInt,
1238 pub parameters: Vec<KeyValuePair>,
1239 pub track_properties: Vec<KeyValuePair>,
1240}
1241
1242/// PUBLISH_DONE (0x0B). The wire layout is unchanged from draft-19; what
1243/// draft-20 changed is the `Stream Count` sentinel, what the count includes,
1244/// and the removal of status code 0x3.
1245#[derive(Debug, Clone, PartialEq, Eq)]
1246pub struct PublishDone {
1247 /// The reason the publisher is ending the subscription.
1248 ///
1249 /// Not validated against the registry, on decode or on encode, and that is
1250 /// the draft's instruction rather than an omission. Draft-22 Section 13:
1251 /// "Receipt of an unknown error code in any error context (Session
1252 /// Termination, REQUEST_ERROR, PUBLISH_DONE, or Data Stream Reset) MUST be
1253 /// treated as equivalent to INTERNAL_ERROR for that context. An endpoint
1254 /// MUST NOT close the session because it received an unknown error code in
1255 /// a REQUEST_ERROR or PUBLISH_DONE." Refusing the frame would take that
1256 /// choice away from the caller, so an unassigned code is carried up and
1257 /// [`super::error_codes::PublishDoneStatusCode::from_u64`] answers `None`
1258 /// for it.
1259 ///
1260 /// That reaches 0x3 in particular. Draft-19 assigned it to
1261 /// `SUBSCRIPTION_ENDED`; draft-20 removed the row and the behaviour behind
1262 /// it together (Section 3.3.1: "A publisher does not end a subscription
1263 /// solely because the Largest Object advances past the end of the current
1264 /// Location Filter"). A draft-22 receiver reads a 0x3 as INTERNAL_ERROR.
1265 pub status_code: VarInt,
1266 /// The number of streams the publisher opened for this subscription.
1267 ///
1268 /// Draft-22 Section 9.9 widens what is counted: the total now includes
1269 /// "streams that contained no Objects (e.g., an empty Subgroup) and
1270 /// including any fill fetch streams (see Section 3.4)". Draft-19 counted
1271 /// no fill streams because it had none.
1272 ///
1273 /// See [`publish_done_codes::STREAM_COUNT_UNKNOWN`] for the sentinel.
1274 pub stream_count: VarInt,
1275 pub reason_phrase: Vec<u8>,
1276}
1277
1278/// Numeric values for the [`PublishDone`] fields.
1279pub mod publish_done_codes {
1280 /// Draft-18 onwards: TOO_FAR_BEHIND is 0x05 (was 0x06 in draft-17).
1281 pub const TOO_FAR_BEHIND: u64 = 0x05;
1282 /// Draft-18 onwards: EXPIRED is 0x06 (was 0x05 in draft-17).
1283 pub const EXPIRED: u64 = 0x06;
1284
1285 /// The value [`super::PublishDone::stream_count`] carries when the
1286 /// publisher cannot state an exact count.
1287 ///
1288 /// Draft-22 Section 9.9: "If the publisher is unable to set Stream Count
1289 /// to the exact number of streams opened for the subscription, it MUST set
1290 /// Stream Count to 2^64 - 1." Draft-19 said `2^62 - 1`, which is where the
1291 /// QUIC varint tops out; MoQT's own varint (Section 8.1) is a
1292 /// leading-ones-length prefix reaching a full 64 bits in nine bytes, so
1293 /// this value is encodable at all — as `ff` followed by eight `ff` bytes.
1294 ///
1295 /// **In draft-22 the sentinel is indistinguishable from a well-formed
1296 /// exact count.** Draft-19's `2^62 - 1` leaves headroom above the marker;
1297 /// there is none above this one, so a publisher that really opened
1298 /// `2^64 - 1` streams cannot say so and a receiver cannot tell the two
1299 /// apart. The draft does not remark on it. Not a practical problem, and
1300 /// worth knowing before writing a comparison against this constant.
1301 pub const STREAM_COUNT_UNKNOWN: u64 = u64::MAX;
1302}
1303
1304// ============================================================
1305// Publish Namespace Messages
1306// ============================================================
1307
1308#[derive(Debug, Clone, PartialEq, Eq)]
1309pub struct PublishNamespace {
1310 pub request_id: VarInt,
1311 pub track_namespace: TrackNamespace,
1312 pub parameters: Vec<KeyValuePair>,
1313}
1314
1315// ============================================================
1316// Namespace Messages
1317// ============================================================
1318
1319#[derive(Debug, Clone, PartialEq, Eq)]
1320pub struct Namespace {
1321 pub namespace_suffix: TrackNamespace,
1322}
1323
1324#[derive(Debug, Clone, PartialEq, Eq)]
1325pub struct NamespaceDone {
1326 pub namespace_suffix: TrackNamespace,
1327}
1328
1329// ============================================================
1330// Subscribe Namespace / Tracks Messages
1331// ============================================================
1332
1333/// SUBSCRIBE_NAMESPACE (0x50). Subscribes to NAMESPACE / NAMESPACE_DONE
1334/// advertisements for namespaces matching `namespace_prefix`. Only drafts 16
1335/// and 17 carry a `subscribe_options` varint here; on draft-22 a namespace
1336/// subscription only produces NAMESPACE / NAMESPACE_DONE.
1337#[derive(Debug, Clone, PartialEq, Eq)]
1338pub struct SubscribeNamespace {
1339 pub request_id: VarInt,
1340 pub namespace_prefix: TrackNamespace,
1341 pub parameters: Vec<KeyValuePair>,
1342}
1343
1344/// SUBSCRIBE_TRACKS (0x51, new in draft-18). Subscribes to PUBLISH messages
1345/// for tracks whose namespace matches `namespace_prefix`. Carries the FORWARD
1346/// parameter (Section 9.20.18), which on drafts 15 through 17 may appear on
1347/// SUBSCRIBE_NAMESPACE instead.
1348#[derive(Debug, Clone, PartialEq, Eq)]
1349pub struct SubscribeTracks {
1350 pub request_id: VarInt,
1351 pub namespace_prefix: TrackNamespace,
1352 pub parameters: Vec<KeyValuePair>,
1353}
1354
1355// ============================================================
1356// Track Status Messages
1357// ============================================================
1358
1359#[derive(Debug, Clone, PartialEq, Eq)]
1360pub struct TrackStatus {
1361 pub request_id: VarInt,
1362 pub track_namespace: TrackNamespace,
1363 pub track_name: Vec<u8>,
1364 pub parameters: Vec<KeyValuePair>,
1365}
1366
1367// ============================================================
1368// Fetch Messages
1369// ============================================================
1370
1371/// FETCH (0x16), rebuilt in draft-20 (Section 9.11, Figure 16).
1372///
1373/// ```text
1374/// FETCH Message {
1375/// Type (vi64) = 0x16,
1376/// Length (16),
1377/// Request ID (vi64),
1378/// Track Namespace (..),
1379/// Track Name Length (vi64),
1380/// Track Name (..),
1381/// Number of Parameters (vi64),
1382/// Parameters (..) ...
1383/// }
1384/// ```
1385///
1386/// # What went, and why draft-19's shape cannot be ported
1387///
1388/// Draft-19's FETCH opened with a `Fetch Type` that chose between a Standalone
1389/// Fetch — a namespace, a name and an inline `Start Location` / `End Location`
1390/// pair — and a Joining Fetch of two varints. Draft-20 deleted the field, both
1391/// structures, the Fetch Type registry and the whole joining mechanism, and
1392/// promoted the namespace and the name to fields of FETCH itself, in the
1393/// positions they hold inside draft-19's Standalone Fetch. What is left is
1394/// byte-identical to [`Subscribe`] apart from the type code.
1395///
1396/// The range travels in the `LOCATION_FILTER` parameter (Section 9.20.9): "If
1397/// omitted from FETCH or SUBSCRIBE, the fetch or subscription is unfiltered."
1398/// Draft-22 places no restriction on which `Location Filter Type` a FETCH may
1399/// carry, so every type decodes here, Relative Start (0x01) and Next Object
1400/// (0x05) included.
1401///
1402/// # `INVALID_RANGE` and a start past Largest Object
1403///
1404/// Section 3.2: "If no Objects have been published for the track or Start
1405/// Location is greater than the Largest Object (Section 3.1.4) the publisher
1406/// MUST return FETCH_ERROR with error code INVALID_RANGE." A Relative Start of
1407/// `StartGroup = 0` resolves to the Next Group, and a Next Object filter to the
1408/// object after Largest Object; both are by construction greater than Largest
1409/// Object, so read literally the rule rejects every such FETCH, and the text
1410/// carves out nothing. Nothing here can apply either reading: the test needs
1411/// Largest Object, which is track state rather than anything in this frame, so
1412/// the decision belongs to the endpoint. It is recorded here because this is
1413/// where the filter arrives.
1414///
1415/// **The codepoint did not change.** A draft-19 decoder fed one of these reads
1416/// the `Number of Track Namespace Fields` count as a `Fetch Type` and
1417/// mis-parses without complaint; there is no in-band version signal to catch
1418/// it. That is why draft-22 has a FETCH decoder of its own rather than sharing
1419/// draft-19's.
1420#[derive(Debug, Clone, PartialEq, Eq)]
1421pub struct Fetch {
1422 pub request_id: VarInt,
1423 pub track_namespace: TrackNamespace,
1424 pub track_name: Vec<u8>,
1425 pub parameters: Vec<KeyValuePair>,
1426}
1427
1428/// FETCH_OK (0x18). `end_of_track` is uint8.
1429///
1430/// Byte-identical to draft-19; the field that changed meaning is
1431/// [`FetchOk::end_object`].
1432#[derive(Debug, Clone, PartialEq, Eq)]
1433pub struct FetchOk {
1434 pub end_of_track: u8,
1435 pub end_group: VarInt,
1436 /// The Object ID of the **last Object the response covers**, inclusive.
1437 ///
1438 /// This is the silent off-by-one of the revision, and it is absent from
1439 /// draft-20's own change log. Draft-19 Section 10.13 defined the pair as
1440 /// "the end of the range covered by the FETCH response, using the same
1441 /// encoding as the FETCH request End Location (the last Object, plus 1; or
1442 /// 0 to indicate the entire Group)". Draft-22 Section 9.12 says "The end of
1443 /// the range covered by the FETCH response, inclusive", and Section 3.3.1
1444 /// says a Location filter "specifies an inclusive range of Locations".
1445 ///
1446 /// So both draft-19 conventions are gone: there is no plus one, and an
1447 /// Object of 0 now means object 0 rather than the whole group. The bytes
1448 /// are identical between the two drafts and the meaning is not, and nothing
1449 /// on the wire distinguishes them — an encoder ported forward with its
1450 /// arithmetic intact fetches one object too many, and one whose end lands
1451 /// on object 0 fetches a single object where the same bytes cover a whole
1452 /// group on draft-19.
1453 ///
1454 /// **Ambiguous, and the draft leaves it so.** When the request's filter
1455 /// omitted `EndObject`, so the
1456 /// filter meant "all objects in the End Group", Section 9.12 does not say
1457 /// what to report — and with the `0`-means-whole-group encoding gone there
1458 /// is no way left to spell it. The same applies to a relative start. The
1459 /// reading this codec's corpus takes, and the only sane one, is the Object
1460 /// ID of the last Object actually covered; the draft does not say so, and
1461 /// nothing here can enforce it, because resolving it needs the track rather
1462 /// than the frame.
1463 pub end_object: VarInt,
1464 pub parameters: Vec<KeyValuePair>,
1465 pub track_properties: Vec<KeyValuePair>,
1466}
1467
1468// ============================================================
1469// Publish State Notify
1470// ============================================================
1471
1472/// PUBLISH_STATE_NOTIFY (0x22), new in draft-20 (Section 9.10, Figure 15).
1473///
1474/// ```text
1475/// PUBLISH_STATE_NOTIFY Message {
1476/// Type (vi64) = 0x22,
1477/// Length (16),
1478/// Number of Parameters (vi64),
1479/// Parameters (..) ...
1480/// }
1481/// ```
1482///
1483/// **There is no Request ID field.** The message is identified by the
1484/// subscription's bidirectional request stream it arrives on, the way
1485/// SUBSCRIBE_OK, PUBLISH_DONE and FETCH_OK are.
1486///
1487/// The rules a codec cannot check, because they are about the session rather
1488/// than the frame, and which the caller therefore owns (Section 9.10):
1489///
1490/// * it is sent by the **publisher only**, on a **subscription's** stream.
1491/// Receiving one for any other request type, or from the subscriber, MUST
1492/// close the session with a PROTOCOL_VIOLATION;
1493/// * it is **unilateral** — the receiver does not answer with REQUEST_OK or
1494/// REQUEST_ERROR — and it is not counted against the `MAX_REQUEST_UPDATES`
1495/// Setup Option;
1496/// * it carries only the parameters whose values changed, and an absent
1497/// parameter is unchanged;
1498/// * a publisher MUST NOT use it to change a subscriber-controlled parameter
1499/// unless the subscriber asked for the change;
1500/// * the publisher **MUST** include `LARGEST_OBJECT` (0x09) if known, so the
1501/// subscriber can locate the point in the track where the change took
1502/// effect. Nothing here enforces that: a missing required parameter is not
1503/// something this decoder detects, and a message that omits it is still a
1504/// well-formed frame.
1505#[derive(Debug, Clone, PartialEq, Eq)]
1506pub struct PublishStateNotify {
1507 pub parameters: Vec<KeyValuePair>,
1508}
1509
1510// ============================================================
1511// Publish Skipped
1512// ============================================================
1513
1514/// PUBLISH_SKIPPED (0x0F). Drafts 17 and 18 name this codepoint
1515/// PUBLISH_BLOCKED; the wire layout — track namespace suffix then track name —
1516/// is the same on all four drafts that carry the message.
1517#[derive(Debug, Clone, PartialEq, Eq)]
1518pub struct PublishSkipped {
1519 pub namespace_suffix: TrackNamespace,
1520 pub track_name: Vec<u8>,
1521}
1522
1523// ============================================================
1524// Unified Message Enum
1525// ============================================================
1526
1527#[derive(Debug, Clone, PartialEq, Eq)]
1528pub enum ControlMessage {
1529 Setup(Setup),
1530 GoAway(GoAway),
1531 RequestOk(RequestOk),
1532 RequestError(RequestError),
1533 Subscribe(Subscribe),
1534 SubscribeOk(SubscribeOk),
1535 RequestUpdate(RequestUpdate),
1536 Publish(Publish),
1537 PublishStateNotify(PublishStateNotify),
1538 PublishDone(PublishDone),
1539 PublishNamespace(PublishNamespace),
1540 Namespace(Namespace),
1541 NamespaceDone(NamespaceDone),
1542 SubscribeNamespace(SubscribeNamespace),
1543 SubscribeTracks(SubscribeTracks),
1544 TrackStatus(TrackStatus),
1545 Fetch(Fetch),
1546 FetchOk(FetchOk),
1547 PublishSkipped(PublishSkipped),
1548}
1549
1550// Draft-22 has no range check on a control message, and the absence is the
1551// draft's rather than an omission here.
1552//
1553// Draft-19 carried one: its FETCH held a `Start Location` and an `End Location`
1554// inline, and draft-19 Section 10.12.3 said "End Location MUST specify the
1555// same or a larger Location than Start Location for Standalone and Absolute
1556// Joining
1557// Fetches". Draft-20 deleted both fields with the rest of the FETCH rewrite
1558// (Section 9.11), so there is no pair left in any message to compare.
1559//
1560// What replaced them cannot express the malformation either. A `LOCATION_FILTER`
1561// (Section 3.3.1) states its end as an unsigned `EndGroupDelta` added to
1562// `StartGroup`, so the end group can never precede the start group; and within
1563// one group draft-22 states no rule about an `EndObject` below `StartObject`.
1564// Section 3.3.1 says the opposite for subscriptions — "A Location Filter on a
1565// subscription is always valid, even if it specifies a range entirely before
1566// Largest Object" — so refusing one would close sessions over a sentence that
1567// is not there.
1568//
1569// The one backwards-range rule draft-22 does keep is in Section 9.12: "If End
1570// Location is smaller than the Start Location in the corresponding FETCH the
1571// receiver MUST close the session with a PROTOCOL_VIOLATION." It compares a
1572// FETCH_OK against the FETCH it answers, which is session state and not
1573// anything one frame holds, so it belongs to the caller.
1574
1575/// Refuse a message whose discriminator disagrees with the fields beside it.
1576///
1577/// One draft-22 message carries a field that says which of the following fields
1578/// are on the wire: REQUEST_ERROR's Error Code, whose REDIRECT value (0x34) is
1579/// what puts the Redirect structure on the wire. This codec holds the
1580/// alternative in an `Option`, so a value can say one thing in its
1581/// discriminator and another in its body, and the two sides of the codec
1582/// resolve that differently — the encoder writes whatever the body holds, and
1583/// the decoder reads whatever the discriminator announces.
1584///
1585/// A REQUEST_ERROR with code REDIRECT and no Redirect body encodes to a message
1586/// that ends where the decoder expects a Connect URI length, so the peer reads
1587/// the redirect out of whatever follows or runs off the end. The mirror case is
1588/// quieter and no better: a Redirect body under any other error code is written
1589/// out and then skipped by a decoder that was never told to look for it, so the
1590/// sender believes it redirected a peer that never saw a redirect.
1591///
1592/// Draft-19 had a second discriminator here, FETCH's `Fetch Type`, choosing
1593/// between a standalone body and a joining pair. Draft-20 deleted the field and
1594/// both bodies (Section 9.11), so FETCH has nothing left to disagree with
1595/// itself about.
1596///
1597/// Refusing at the encoder keeps the two readings from ever diverging on the
1598/// wire.
1599fn check_discriminators(message: &ControlMessage) -> Result<(), CodecError> {
1600 // One arm, where drafts 17 to 19 have two, so an `if let` rather than a
1601 // one-armed `match`; a second discriminator would make it a `match`.
1602 if let ControlMessage::RequestError(m) = message {
1603 let code_is_redirect = m.error_code.into_inner() == request_error_codes::REDIRECT;
1604 if code_is_redirect != m.redirect.is_some() {
1605 return Err(CodecError::InvalidField);
1606 }
1607 }
1608 Ok(())
1609}
1610
1611/// Whether draft-22 lets Message Parameter `key` appear in `message`.
1612///
1613/// Section 9.20.1: "Each Message Parameter definition indicates the message
1614/// types in which it can appear, and each control message definition lists the
1615/// parameters it allows. If a parameter appears in some other type of message,
1616/// the receiving endpoint MUST close the connection with a PROTOCOL_VIOLATION."
1617/// One arm per entry in the Message Parameters registry (Section 16.7),
1618/// carrying the message types that entry's own definition names. The lists in
1619/// the message definitions say the same thing from the other side.
1620///
1621/// Four things about this draft the arms below fold in:
1622///
1623/// * Six of the names are one wire type. Section 1.5: "This document uses the
1624/// shorthand PUBLISH_OK, REQUEST_UPDATE_OK, TRACK_STATUS_OK,
1625/// SUBSCRIBE_NAMESPACE_OK, SUBSCRIBE_TRACKS_OK and PUBLISH_NAMESPACE_OK to
1626/// refer to a REQUEST_OK sent in response to the corresponding request type."
1627/// Section 9.3 gives each of the six its own list, and which list applies
1628/// depends on the request the REQUEST_OK answers, which is session state and
1629/// not in the frame. So a REQUEST_OK is held to the union of the six:
1630/// `EXPIRES` (0x08) and `LARGEST_OBJECT` (0x09).
1631/// * REQUEST_UPDATE is the same shape. Section 9.5 lists its parameters per
1632/// kind of request being updated, and the request is named by a Request ID
1633/// this frame cannot resolve, so REQUEST_UPDATE is held to the union of the
1634/// five lists. A session that knows the request can hold an update to the
1635/// narrower list; this table cannot, and it does not guess.
1636/// * The five Range Filters state their scope in Section 3.3.2 rather than in
1637/// their own subsections: the Track Property filter "MAY appear multiple
1638/// times in a SUBSCRIBE_TRACKS message or REQUEST_UPDATE for it", and "All
1639/// other filter parameters MAY appear multiple times in a FETCH, SUBSCRIBE,
1640/// SUBSCRIBE_TRACKS, or REQUEST_UPDATE (on a subscription, from the
1641/// subscriber only) message".
1642/// * **SUBSCRIBE_TRACKS takes the parameters Section 9.18 lists, plus
1643/// `LOCATION_FILTER` (0x21) and `FILL_PARAMETERS` (0x23), and the two extra
1644/// are a decision this codec makes.** Section 9.18: "The parameters that can
1645/// appear in a SUBSCRIBE_TRACKS are AUTHORIZATION_TOKEN, FORWARD,
1646/// GROUP_ORDER, SUBGROUP_FILTER, OBJECTID_FILTER, PRIORITY_FILTER,
1647/// OBJECT_PROPERTY_FILTER, TRACK_PROPERTY_FILTER and INCLUDE_PROPERTIES."
1648/// That is exactly the set whose own definitions name SUBSCRIBE_TRACKS, and
1649/// it is the "unless otherwise specified" in Section 3.6.2's "Any Parameter
1650/// that can be specified on a Subscription (ie: in SUBSCRIBE) is valid in
1651/// SUBSCRIBE_TRACKS, unless otherwise specified", so
1652/// `OBJECT_DELIVERY_TIMEOUT` (0x02), `RENDEZVOUS_TIMEOUT` (0x04),
1653/// `SUBGROUP_DELIVERY_TIMEOUT` (0x06), `SUBSCRIBER_PRIORITY` (0x20) and
1654/// `NEW_GROUP_REQUEST` (0x32) are refused there under Section 9.20.1.
1655/// Section 3.6.2 then closes: "To join Tracks initiated via the resulting
1656/// PUBLISHes, the subscriber can specify a Location Filter and optionally
1657/// include FILL_PARAMETERS in the SUBSCRIBE_TRACKS" — two parameters that
1658/// neither the list nor their own definitions place in this message. On
1659/// those two alone the section conflicts with the list, the definitions and
1660/// Section 9.20.1's rule, and the conflict is the one drafts 20 and 21
1661/// carry. This codec follows Section 3.6.2 for them: it is the one sentence
1662/// telling a subscriber what to put in this message, and a scope miss ends
1663/// the session.
1664///
1665/// FETCH_OK has no arm in the table below, and that is the draft's doing rather
1666/// than an omission here: Section 9.12 gives it a Parameters field and says "No
1667/// parameters are currently defined for FETCH_OK", so every type this draft
1668/// defines is "some other type of message" there.
1669///
1670/// PUBLISH_STATE_NOTIFY is admitted by exactly three parameters. Section 9.10
1671/// closes the list itself: "The parameters that can appear in a
1672/// PUBLISH_STATE_NOTIFY are LARGEST_OBJECT, FORWARD and LOCATION_FILTER."
1673///
1674/// The table decides scope only. A type this draft does not define has no scope
1675/// to be outside of and is answered by [`CodecError::UnknownMessageParameter`],
1676/// which is why the final arm carries rather than refuses.
1677pub fn parameter_in_scope(key: u64, message: MessageType) -> bool {
1678 use MessageType as M;
1679 // SUBSCRIBE_TRACKS appears only in the arms Section 9.18 lists, and in 0x21
1680 // and 0x23, which Section 3.6.2 names; see the doc comment above.
1681 match key {
1682 // Section 9.20.4 OBJECT_DELIVERY_TIMEOUT: "It MAY appear in a
1683 // SUBSCRIBE, PUBLISH, or REQUEST_UPDATE message."
1684 0x02 => matches!(message, M::Subscribe | M::Publish | M::RequestUpdate),
1685 // Section 9.20.2 AUTHORIZATION TOKEN: "It MAY appear in a PUBLISH,
1686 // SUBSCRIBE, REQUEST_UPDATE, SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS,
1687 // PUBLISH_NAMESPACE, TRACK_STATUS or FETCH message." The sentence after
1688 // it, "This Parameter MUST NOT be copied from a SUBSCRIBE_TRACKS to the
1689 // resulting PUBLISH message Parameters", is a rule about two messages
1690 // and so belongs to the caller rather than to this table.
1691 0x03 => matches!(
1692 message,
1693 M::Publish
1694 | M::Subscribe
1695 | M::RequestUpdate
1696 | M::SubscribeNamespace
1697 | M::SubscribeTracks
1698 | M::PublishNamespace
1699 | M::TrackStatus
1700 | M::Fetch
1701 ),
1702 // Section 9.20.6 RENDEZVOUS TIMEOUT: it "MAY appear in a SUBSCRIBE
1703 // message".
1704 0x04 => matches!(message, M::Subscribe),
1705 // Section 9.20.3 SUBGROUP_DELIVERY_TIMEOUT: "It MAY appear in a
1706 // SUBSCRIBE, PUBLISH, or REQUEST_UPDATE message."
1707 0x06 => matches!(message, M::Subscribe | M::Publish | M::RequestUpdate),
1708 // Section 9.20.16 EXPIRES: "It MAY appear in SUBSCRIBE_OK, PUBLISH,
1709 // PUBLISH_OK, SUBSCRIBE_NAMESPACE_OK, SUBSCRIBE_TRACKS_OK,
1710 // PUBLISH_NAMESPACE_OK, or REQUEST_UPDATE_OK." Five of those seven are
1711 // a REQUEST_OK, and it is the one parameter whose definition names
1712 // PUBLISH_OK.
1713 0x08 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
1714 // Section 9.20.17 LARGEST OBJECT: "It MAY appear in SUBSCRIBE_OK,
1715 // PUBLISH, REQUEST_UPDATE_OK, TRACK_STATUS_OK, or
1716 // PUBLISH_STATE_NOTIFY."
1717 0x09 => {
1718 matches!(message, M::SubscribeOk | M::Publish | M::RequestOk | M::PublishStateNotify)
1719 }
1720 // Section 9.20.5 FILL TIMEOUT: it "MAY appear in a FETCH message, or
1721 // inside a FILL_PARAMETERS parameter (see Section 9.20.15) in a
1722 // SUBSCRIBE or REQUEST_UPDATE (for a subscription), where it applies to
1723 // the fill fetch stream." The nested half is not a scope this table can
1724 // answer — a nested parameter is not in the enclosing message at all,
1725 // per Section 9.20.15 — so it is enforced by
1726 // [`decode_fill_parameters`] against Table 7 instead.
1727 0x0A => matches!(message, M::Fetch),
1728 // Section 9.20.18 FORWARD: "It MAY appear in SUBSCRIBE, REQUEST_UPDATE
1729 // (for a subscription or a SUBSCRIBE_TRACKS request), PUBLISH,
1730 // SUBSCRIBE_TRACKS and PUBLISH_STATE_NOTIFY."
1731 0x10 => matches!(
1732 message,
1733 M::Subscribe
1734 | M::RequestUpdate
1735 | M::Publish
1736 | M::SubscribeTracks
1737 | M::PublishStateNotify
1738 ),
1739 // Section 9.20.7 SUBSCRIBER PRIORITY: "It MAY appear in a SUBSCRIBE,
1740 // PUBLISH, FETCH, or REQUEST_UPDATE (for a subscription or FETCH)."
1741 0x20 => matches!(message, M::Subscribe | M::Publish | M::Fetch | M::RequestUpdate),
1742 // Section 9.20.9 LOCATION FILTER: "The LOCATION_FILTER parameter
1743 // (Parameter Type 0x21) MAY appear in a FETCH, SUBSCRIBE, PUBLISH,
1744 // REQUEST_UPDATE (for a subscription) or PUBLISH_STATE_NOTIFY
1745 // message." SUBSCRIBE_TRACKS is admitted on top of that list by
1746 // Section 3.6.2's closing sentence, as for FILL_PARAMETERS below.
1747 0x21 => matches!(
1748 message,
1749 M::Fetch
1750 | M::Subscribe
1751 | M::Publish
1752 | M::RequestUpdate
1753 | M::PublishStateNotify
1754 | M::SubscribeTracks
1755 ),
1756 // Section 9.20.8 GROUP ORDER: "It MAY appear in a SUBSCRIBE, PUBLISH,
1757 // SUBSCRIBE_TRACKS, or FETCH, or inside a FILL_PARAMETERS parameter".
1758 // The nested half is Table 7's, as for FILL_TIMEOUT above.
1759 0x22 => matches!(message, M::Subscribe | M::Publish | M::SubscribeTracks | M::Fetch),
1760 // Section 9.20.15 FILL_PARAMETERS: it "MAY appear in a SUBSCRIBE or
1761 // REQUEST_UPDATE (for a subscription) message."
1762 //
1763 // SUBSCRIBE_TRACKS is admitted on top of those two by Section 3.6.2's
1764 // closing sentence, which names FILL_PARAMETERS "in the
1765 // SUBSCRIBE_TRACKS" outright; see the doc comment above for why that
1766 // sentence wins over this list.
1767 //
1768 // FETCH is refused, and that is the case the corpus pins: a fill fetch
1769 // stream is something a subscription opens, and a FETCH already is one.
1770 0x23 => matches!(message, M::Subscribe | M::RequestUpdate | M::SubscribeTracks),
1771 // Section 3.3.2: "All other filter parameters MAY appear multiple times
1772 // in a FETCH, SUBSCRIBE, SUBSCRIBE_TRACKS, or REQUEST_UPDATE (on a
1773 // subscription, from the subscriber only) message." SUBGROUP_FILTER
1774 // (Section 9.20.10), OBJECTID_FILTER (Section 9.20.11), PRIORITY_FILTER
1775 // (Section 9.20.12) and OBJECT_PROPERTY_FILTER (Section 9.20.13) are
1776 // those four.
1777 0x25..=0x28 => {
1778 matches!(message, M::Fetch | M::Subscribe | M::SubscribeTracks | M::RequestUpdate)
1779 }
1780 // Section 3.3.2, of TRACK_PROPERTY_FILTER (Section 9.20.14) alone: it
1781 // "MAY appear multiple times in a SUBSCRIBE_TRACKS message or
1782 // REQUEST_UPDATE for it". It selects tracks rather than objects, which
1783 // is why it is the one filter a SUBSCRIBE may not carry and the one
1784 // Table 7 keeps out of FILL_PARAMETERS.
1785 0x29 => matches!(message, M::SubscribeTracks | M::RequestUpdate),
1786 // Section 9.20.19 NEW GROUP REQUEST: "It MAY appear in SUBSCRIBE or
1787 // REQUEST_UPDATE for a subscription."
1788 0x32 => matches!(message, M::Subscribe | M::RequestUpdate),
1789 // Section 9.20.20 TRACK_NAMESPACE_PREFIX: "It MAY appear in
1790 // REQUEST_UPDATE for a SUBSCRIBE_NAMESPACE or SUBSCRIBE_TRACKS
1791 // request." The two named there are the request being updated, not two
1792 // more places the parameter may be written.
1793 0x34 => matches!(message, M::RequestUpdate),
1794 // Section 9.20.21 INCLUDE_PROPERTIES: "It MAY appear in SUBSCRIBE,
1795 // TRACK_STATUS, FETCH or SUBSCRIBE_TRACKS."
1796 0x35 => matches!(message, M::Subscribe | M::TrackStatus | M::Fetch | M::SubscribeTracks),
1797 _ => true,
1798 }
1799}
1800
1801/// Refuse a message carrying a Message Parameter its own definition does not
1802/// place there.
1803///
1804/// Section 9.20.1 answers this with a close, which the drafts below do not.
1805/// Draft-16 Section 9.2.2, and drafts 07 through 15 under the older name
1806/// Version Specific Parameters, end the same sentence "it MUST be ignored" —
1807/// so this check belongs to drafts 17 and later and to no draft before them.
1808///
1809/// Applied on both sides. A parameter outside its scope is one the peer must
1810/// close the session over, so writing one is a way to end a session rather than
1811/// a way to ask for anything.
1812fn check_parameter_scope(message: &ControlMessage) -> Result<(), CodecError> {
1813 let parameters = match message {
1814 ControlMessage::RequestOk(m) => &m.parameters,
1815 ControlMessage::Subscribe(m) => &m.parameters,
1816 ControlMessage::SubscribeOk(m) => &m.parameters,
1817 ControlMessage::RequestUpdate(m) => &m.parameters,
1818 ControlMessage::Publish(m) => &m.parameters,
1819 ControlMessage::PublishStateNotify(m) => &m.parameters,
1820 ControlMessage::PublishNamespace(m) => &m.parameters,
1821 ControlMessage::SubscribeNamespace(m) => &m.parameters,
1822 ControlMessage::SubscribeTracks(m) => &m.parameters,
1823 ControlMessage::TrackStatus(m) => &m.parameters,
1824 ControlMessage::Fetch(m) => &m.parameters,
1825 ControlMessage::FetchOk(m) => &m.parameters,
1826 // No Message Parameters field. SETUP is named here rather than left to
1827 // a wildcard because the draft says why it can never have one: Section
1828 // 9.20.1 notes that "since Setup Options use a separate namespace, it
1829 // is impossible for Message Parameters to appear in Setup messages",
1830 // and this codec keeps the two namespaces in separate fields.
1831 ControlMessage::Setup(_)
1832 | ControlMessage::GoAway(_)
1833 | ControlMessage::RequestError(_)
1834 | ControlMessage::PublishDone(_)
1835 | ControlMessage::Namespace(_)
1836 | ControlMessage::NamespaceDone(_)
1837 | ControlMessage::PublishSkipped(_) => return Ok(()),
1838 };
1839
1840 let message_type = message.message_type();
1841 for parameter in parameters {
1842 let key = parameter.key.into_inner();
1843 if !parameter_in_scope(key, message_type) {
1844 return Err(CodecError::ParameterOutOfScope { key, message_type: message_type.id() });
1845 }
1846 }
1847 Ok(())
1848}
1849
1850impl ControlMessage {
1851 pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1852 check_discriminators(self)?;
1853 check_parameter_scope(self)?;
1854 let mut body = Vec::with_capacity(256);
1855 self.encode_body(&mut body)?;
1856
1857 if body.len() > MAX_MESSAGE_LENGTH {
1858 return Err(CodecError::MessageTooLong(body.len()));
1859 }
1860
1861 let msg_type = self.message_type();
1862 VarInt::from_usize(msg_type.id() as usize).encode_moqt::<Wire>(buf);
1863 // Draft-22: 16-bit length (big-endian)
1864 buf.put_u16(body.len() as u16);
1865 buf.put_slice(&body);
1866 Ok(())
1867 }
1868
1869 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1870 let type_id = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1871 let msg_type =
1872 MessageType::from_id(type_id).ok_or(CodecError::UnknownMessageType(type_id))?;
1873 // Draft-22: 16-bit length (big-endian)
1874 if buf.remaining() < 2 {
1875 return Err(CodecError::UnexpectedEnd);
1876 }
1877 let body_len = buf.get_u16() as usize;
1878 if buf.remaining() < body_len {
1879 return Err(CodecError::UnexpectedEnd);
1880 }
1881 let body_bytes = buf.copy_to_bytes(body_len);
1882 let mut body = &body_bytes[..];
1883 let msg = match Self::decode_body(msg_type, &mut body) {
1884 Ok(msg) => msg,
1885 // The fields wanted more bytes than the Length allowed. This buffer
1886 // is already bounded by that Length, so running out inside it cannot
1887 // mean the message is still arriving - which is what the same error
1888 // means everywhere else, and why a reader loops on it rather than
1889 // closing. Here there is nothing left to arrive.
1890 Err(
1891 CodecError::UnexpectedEnd
1892 | CodecError::Kvp(crate::kvp::KvpError::UnexpectedEnd)
1893 | CodecError::Kvp(crate::kvp::KvpError::VarInt(
1894 crate::varint::VarIntError::UnexpectedEnd,
1895 ))
1896 | CodecError::VarInt(crate::varint::VarIntError::UnexpectedEnd),
1897 ) => {
1898 return Err(CodecError::ControlMessageLengthMismatch {
1899 declared: body_len,
1900 detail: "its fields ran past the end",
1901 });
1902 }
1903 Err(e) => return Err(e),
1904 };
1905 check_parameter_scope(&msg)?;
1906 // Draft-22 Section 9: "If the length does not match the length of the
1907 // Message Body, the receiver MUST close the session with a
1908 // PROTOCOL_VIOLATION." A body parser that stops short leaves bytes
1909 // here; without this the surplus is discarded and a truncated or
1910 // mis-framed field looks like a well-formed message.
1911 if body.has_remaining() {
1912 return Err(CodecError::ControlMessageLengthMismatch {
1913 declared: body_len,
1914 detail: "its fields left bytes unread",
1915 });
1916 }
1917 Ok(msg)
1918 }
1919
1920 fn encode_body(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1921 match self {
1922 ControlMessage::Setup(m) => {
1923 encode_setup_options(&m.options, buf)?;
1924 }
1925 ControlMessage::GoAway(m) => {
1926 if m.new_session_uri.len() > MAX_GOAWAY_URI_LENGTH {
1927 return Err(CodecError::GoAwayUriTooLong);
1928 }
1929 VarInt::from_usize(m.new_session_uri.len()).encode_moqt::<Wire>(buf);
1930 buf.put_slice(&m.new_session_uri);
1931 m.timeout.encode_moqt::<Wire>(buf);
1932 }
1933 ControlMessage::RequestOk(m) => {
1934 encode_parameters(&m.parameters, buf)?;
1935 encode_track_properties(&m.track_properties, buf)?;
1936 }
1937 ControlMessage::RequestError(m) => {
1938 if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1939 return Err(CodecError::ReasonPhraseTooLong);
1940 }
1941 m.error_code.encode_moqt::<Wire>(buf);
1942 m.retry_interval.encode_moqt::<Wire>(buf);
1943 VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
1944 buf.put_slice(&m.reason_phrase);
1945 if let Some(r) = &m.redirect {
1946 r.track_namespace.validate_moqt()?;
1947 check_full_track_name(&r.track_namespace, &r.track_name)?;
1948 VarInt::from_usize(r.connect_uri.len()).encode_moqt::<Wire>(buf);
1949 buf.put_slice(&r.connect_uri);
1950 r.track_namespace.encode_moqt::<Wire>(buf);
1951 VarInt::from_usize(r.track_name.len()).encode_moqt::<Wire>(buf);
1952 buf.put_slice(&r.track_name);
1953 }
1954 }
1955 ControlMessage::Subscribe(m) => {
1956 m.track_namespace.validate_moqt()?;
1957 check_full_track_name(&m.track_namespace, &m.track_name)?;
1958 m.request_id.encode_moqt::<Wire>(buf);
1959 m.track_namespace.encode_moqt::<Wire>(buf);
1960 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1961 buf.put_slice(&m.track_name);
1962 encode_parameters(&m.parameters, buf)?;
1963 }
1964 ControlMessage::SubscribeOk(m) => {
1965 m.track_alias.encode_moqt::<Wire>(buf);
1966 encode_parameters(&m.parameters, buf)?;
1967 encode_track_properties(&m.track_properties, buf)?;
1968 }
1969 ControlMessage::RequestUpdate(m) => {
1970 m.request_id.encode_moqt::<Wire>(buf);
1971 encode_parameters(&m.parameters, buf)?;
1972 }
1973 ControlMessage::Publish(m) => {
1974 m.track_namespace.validate_moqt()?;
1975 check_full_track_name(&m.track_namespace, &m.track_name)?;
1976 m.request_id.encode_moqt::<Wire>(buf);
1977 m.track_namespace.encode_moqt::<Wire>(buf);
1978 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1979 buf.put_slice(&m.track_name);
1980 m.track_alias.encode_moqt::<Wire>(buf);
1981 encode_parameters(&m.parameters, buf)?;
1982 encode_track_properties(&m.track_properties, buf)?;
1983 }
1984 // Section 9.10, Figure 15: parameter count, then parameters,
1985 // and nothing else. No Request ID — the subscription's stream is
1986 // what identifies the message.
1987 ControlMessage::PublishStateNotify(m) => {
1988 encode_parameters(&m.parameters, buf)?;
1989 }
1990 ControlMessage::PublishDone(m) => {
1991 if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1992 return Err(CodecError::ReasonPhraseTooLong);
1993 }
1994 m.status_code.encode_moqt::<Wire>(buf);
1995 m.stream_count.encode_moqt::<Wire>(buf);
1996 VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
1997 buf.put_slice(&m.reason_phrase);
1998 }
1999 ControlMessage::PublishNamespace(m) => {
2000 m.track_namespace.validate_moqt()?;
2001 m.request_id.encode_moqt::<Wire>(buf);
2002 m.track_namespace.encode_moqt::<Wire>(buf);
2003 encode_parameters(&m.parameters, buf)?;
2004 }
2005 ControlMessage::Namespace(m) => {
2006 m.namespace_suffix.validate_moqt()?;
2007 m.namespace_suffix.encode_moqt::<Wire>(buf);
2008 }
2009 ControlMessage::NamespaceDone(m) => {
2010 m.namespace_suffix.validate_moqt()?;
2011 m.namespace_suffix.encode_moqt::<Wire>(buf);
2012 }
2013 ControlMessage::SubscribeNamespace(m) => {
2014 m.namespace_prefix.validate_moqt()?;
2015 m.request_id.encode_moqt::<Wire>(buf);
2016 m.namespace_prefix.encode_moqt::<Wire>(buf);
2017 encode_parameters(&m.parameters, buf)?;
2018 }
2019 ControlMessage::SubscribeTracks(m) => {
2020 m.namespace_prefix.validate_moqt()?;
2021 m.request_id.encode_moqt::<Wire>(buf);
2022 m.namespace_prefix.encode_moqt::<Wire>(buf);
2023 encode_parameters(&m.parameters, buf)?;
2024 }
2025 ControlMessage::TrackStatus(m) => {
2026 m.track_namespace.validate_moqt()?;
2027 check_full_track_name(&m.track_namespace, &m.track_name)?;
2028 m.request_id.encode_moqt::<Wire>(buf);
2029 m.track_namespace.encode_moqt::<Wire>(buf);
2030 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
2031 buf.put_slice(&m.track_name);
2032 encode_parameters(&m.parameters, buf)?;
2033 }
2034 // Section 9.11, Figure 16. Byte-identical to SUBSCRIBE above:
2035 // no Fetch Type, no inline range, and the namespace and name where
2036 // draft-19 put them inside its Standalone Fetch.
2037 ControlMessage::Fetch(m) => {
2038 m.track_namespace.validate_moqt()?;
2039 check_full_track_name(&m.track_namespace, &m.track_name)?;
2040 m.request_id.encode_moqt::<Wire>(buf);
2041 m.track_namespace.encode_moqt::<Wire>(buf);
2042 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
2043 buf.put_slice(&m.track_name);
2044 encode_parameters(&m.parameters, buf)?;
2045 }
2046 ControlMessage::FetchOk(m) => {
2047 buf.put_u8(m.end_of_track);
2048 m.end_group.encode_moqt::<Wire>(buf);
2049 m.end_object.encode_moqt::<Wire>(buf);
2050 encode_parameters(&m.parameters, buf)?;
2051 encode_track_properties(&m.track_properties, buf)?;
2052 }
2053 ControlMessage::PublishSkipped(m) => {
2054 m.namespace_suffix.validate_moqt()?;
2055 check_full_track_name(&m.namespace_suffix, &m.track_name)?;
2056 m.namespace_suffix.encode_moqt::<Wire>(buf);
2057 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
2058 buf.put_slice(&m.track_name);
2059 }
2060 }
2061 Ok(())
2062 }
2063
2064 fn decode_body(msg_type: MessageType, buf: &mut impl Buf) -> Result<Self, CodecError> {
2065 match msg_type {
2066 MessageType::Setup => {
2067 let options = decode_setup_options(buf)?;
2068 Ok(ControlMessage::Setup(Setup { options }))
2069 }
2070 MessageType::GoAway => {
2071 let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2072 // Draft-22 Section 9.2: an endpoint that receives a New
2073 // Session URI Length above the maximum MUST close the session
2074 // with a PROTOCOL_VIOLATION. Checked here as well as on encode
2075 // so an oversized URI never reaches the application.
2076 if uri_len > MAX_GOAWAY_URI_LENGTH {
2077 return Err(CodecError::GoAwayUriTooLong);
2078 }
2079 let uri = read_bytes(buf, uri_len)?;
2080 let timeout = VarInt::decode_moqt::<Wire>(buf)?;
2081 Ok(ControlMessage::GoAway(GoAway { new_session_uri: uri, timeout }))
2082 }
2083 MessageType::RequestOk => {
2084 let parameters = decode_parameters(buf)?;
2085 let track_properties = decode_track_properties(buf)?;
2086 Ok(ControlMessage::RequestOk(RequestOk { parameters, track_properties }))
2087 }
2088 MessageType::RequestError => {
2089 let error_code = VarInt::decode_moqt::<Wire>(buf)?;
2090 let retry_interval = VarInt::decode_moqt::<Wire>(buf)?;
2091 let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2092 // Draft-22 Section 8.5: a received reason phrase length above
2093 // the maximum MUST close the session with a PROTOCOL_VIOLATION.
2094 if reason_len > MAX_REASON_PHRASE_LENGTH {
2095 return Err(CodecError::ReasonPhraseTooLong);
2096 }
2097 let reason_phrase = read_bytes(buf, reason_len)?;
2098 let redirect = if error_code.into_inner() == request_error_codes::REDIRECT {
2099 let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2100 let connect_uri = read_bytes(buf, uri_len)?;
2101 let track_namespace = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
2102 let name_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2103 let track_name = read_bytes(buf, name_len)?;
2104 check_full_track_name(&track_namespace, &track_name)?;
2105 Some(Redirect { connect_uri, track_namespace, track_name })
2106 } else {
2107 None
2108 };
2109 Ok(ControlMessage::RequestError(RequestError {
2110 error_code,
2111 retry_interval,
2112 reason_phrase,
2113 redirect,
2114 }))
2115 }
2116 MessageType::Subscribe => {
2117 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2118 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
2119 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2120 let track_name = read_bytes(buf, tn_len)?;
2121 check_full_track_name(&track_namespace, &track_name)?;
2122 let parameters = decode_parameters(buf)?;
2123 Ok(ControlMessage::Subscribe(Subscribe {
2124 request_id,
2125 track_namespace,
2126 track_name,
2127 parameters,
2128 }))
2129 }
2130 MessageType::SubscribeOk => {
2131 let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
2132 let parameters = decode_parameters(buf)?;
2133 let track_properties = decode_track_properties(buf)?;
2134 Ok(ControlMessage::SubscribeOk(SubscribeOk {
2135 track_alias,
2136 parameters,
2137 track_properties,
2138 }))
2139 }
2140 MessageType::RequestUpdate => {
2141 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2142 let parameters = decode_parameters(buf)?;
2143 Ok(ControlMessage::RequestUpdate(RequestUpdate { request_id, parameters }))
2144 }
2145 MessageType::Publish => {
2146 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2147 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
2148 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2149 let track_name = read_bytes(buf, tn_len)?;
2150 let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
2151 check_full_track_name(&track_namespace, &track_name)?;
2152 let parameters = decode_parameters(buf)?;
2153 let track_properties = decode_track_properties(buf)?;
2154 Ok(ControlMessage::Publish(Publish {
2155 request_id,
2156 track_namespace,
2157 track_name,
2158 track_alias,
2159 parameters,
2160 track_properties,
2161 }))
2162 }
2163 MessageType::PublishStateNotify => {
2164 let parameters = decode_parameters(buf)?;
2165 Ok(ControlMessage::PublishStateNotify(PublishStateNotify { parameters }))
2166 }
2167 MessageType::PublishDone => {
2168 // The Status Code is carried up whatever it is. Section 13
2169 // forbids closing the session over an unknown one and requires
2170 // it to be read as INTERNAL_ERROR, so refusing the frame here
2171 // would take a choice away from the caller that the draft gives
2172 // it. See [`PublishDone::status_code`].
2173 let status_code = VarInt::decode_moqt::<Wire>(buf)?;
2174 let stream_count = VarInt::decode_moqt::<Wire>(buf)?;
2175 let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2176 // Draft-22 Section 8.5, same bound as REQUEST_ERROR above.
2177 if reason_len > MAX_REASON_PHRASE_LENGTH {
2178 return Err(CodecError::ReasonPhraseTooLong);
2179 }
2180 let reason_phrase = read_bytes(buf, reason_len)?;
2181 Ok(ControlMessage::PublishDone(PublishDone {
2182 status_code,
2183 stream_count,
2184 reason_phrase,
2185 }))
2186 }
2187 MessageType::PublishNamespace => {
2188 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2189 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
2190 let parameters = decode_parameters(buf)?;
2191 Ok(ControlMessage::PublishNamespace(PublishNamespace {
2192 request_id,
2193 track_namespace,
2194 parameters,
2195 }))
2196 }
2197 MessageType::Namespace => {
2198 let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
2199 Ok(ControlMessage::Namespace(Namespace { namespace_suffix }))
2200 }
2201 MessageType::NamespaceDone => {
2202 let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
2203 Ok(ControlMessage::NamespaceDone(NamespaceDone { namespace_suffix }))
2204 }
2205 MessageType::SubscribeNamespace => {
2206 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2207 let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
2208 let parameters = decode_parameters(buf)?;
2209 Ok(ControlMessage::SubscribeNamespace(SubscribeNamespace {
2210 request_id,
2211 namespace_prefix,
2212 parameters,
2213 }))
2214 }
2215 MessageType::SubscribeTracks => {
2216 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2217 let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
2218 let parameters = decode_parameters(buf)?;
2219 Ok(ControlMessage::SubscribeTracks(SubscribeTracks {
2220 request_id,
2221 namespace_prefix,
2222 parameters,
2223 }))
2224 }
2225 MessageType::TrackStatus => {
2226 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2227 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
2228 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2229 let track_name = read_bytes(buf, tn_len)?;
2230 check_full_track_name(&track_namespace, &track_name)?;
2231 let parameters = decode_parameters(buf)?;
2232 Ok(ControlMessage::TrackStatus(TrackStatus {
2233 request_id,
2234 track_namespace,
2235 track_name,
2236 parameters,
2237 }))
2238 }
2239 MessageType::Fetch => {
2240 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
2241 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
2242 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2243 let track_name = read_bytes(buf, tn_len)?;
2244 check_full_track_name(&track_namespace, &track_name)?;
2245 let parameters = decode_parameters(buf)?;
2246 Ok(ControlMessage::Fetch(Fetch {
2247 request_id,
2248 track_namespace,
2249 track_name,
2250 parameters,
2251 }))
2252 }
2253 MessageType::FetchOk => {
2254 if buf.remaining() < 1 {
2255 return Err(CodecError::UnexpectedEnd);
2256 }
2257 let end_of_track = buf.get_u8();
2258 let end_group = VarInt::decode_moqt::<Wire>(buf)?;
2259 let end_object = VarInt::decode_moqt::<Wire>(buf)?;
2260 let parameters = decode_parameters(buf)?;
2261 let track_properties = decode_track_properties(buf)?;
2262 Ok(ControlMessage::FetchOk(FetchOk {
2263 end_of_track,
2264 end_group,
2265 end_object,
2266 parameters,
2267 track_properties,
2268 }))
2269 }
2270 MessageType::PublishSkipped => {
2271 let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
2272 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
2273 let track_name = read_bytes(buf, tn_len)?;
2274 check_full_track_name(&namespace_suffix, &track_name)?;
2275 Ok(ControlMessage::PublishSkipped(PublishSkipped { namespace_suffix, track_name }))
2276 }
2277 }
2278 }
2279
2280 pub fn message_type(&self) -> MessageType {
2281 match self {
2282 ControlMessage::Setup(_) => MessageType::Setup,
2283 ControlMessage::GoAway(_) => MessageType::GoAway,
2284 ControlMessage::RequestOk(_) => MessageType::RequestOk,
2285 ControlMessage::RequestError(_) => MessageType::RequestError,
2286 ControlMessage::Subscribe(_) => MessageType::Subscribe,
2287 ControlMessage::SubscribeOk(_) => MessageType::SubscribeOk,
2288 ControlMessage::RequestUpdate(_) => MessageType::RequestUpdate,
2289 ControlMessage::Publish(_) => MessageType::Publish,
2290 ControlMessage::PublishStateNotify(_) => MessageType::PublishStateNotify,
2291 ControlMessage::PublishDone(_) => MessageType::PublishDone,
2292 ControlMessage::PublishNamespace(_) => MessageType::PublishNamespace,
2293 ControlMessage::Namespace(_) => MessageType::Namespace,
2294 ControlMessage::NamespaceDone(_) => MessageType::NamespaceDone,
2295 ControlMessage::SubscribeNamespace(_) => MessageType::SubscribeNamespace,
2296 ControlMessage::SubscribeTracks(_) => MessageType::SubscribeTracks,
2297 ControlMessage::TrackStatus(_) => MessageType::TrackStatus,
2298 ControlMessage::Fetch(_) => MessageType::Fetch,
2299 ControlMessage::FetchOk(_) => MessageType::FetchOk,
2300 ControlMessage::PublishSkipped(_) => MessageType::PublishSkipped,
2301 }
2302 }
2303}
2304
2305#[cfg(test)]
2306mod tests {
2307 use super::*;
2308
2309 /// Frame `body` as a draft-22 control message of `type_id`, declaring
2310 /// `declared_len` rather than the body's real length. Used to build the
2311 /// mismatched frame the length rule is about.
2312 fn frame_with_declared_len(type_id: u64, declared_len: u16, body: &[u8]) -> Vec<u8> {
2313 let mut out = Vec::new();
2314 VarInt::from_u64_moqt(type_id).encode_moqt::<Wire>(&mut out);
2315 out.put_u16(declared_len);
2316 out.put_slice(body);
2317 out
2318 }
2319
2320 fn frame(type_id: u64, body: &[u8]) -> Vec<u8> {
2321 frame_with_declared_len(type_id, body.len() as u16, body)
2322 }
2323
2324 /// A SUBSCRIBE body: request id 1, namespace ("a"), track name "b", and
2325 /// `params` already encoded.
2326 fn subscribe_body(params: &[u8]) -> Vec<u8> {
2327 let mut body = vec![0x01, 0x01, 0x01, b'a', 0x01, b'b'];
2328 body.extend_from_slice(params);
2329 body
2330 }
2331
2332 /// Draft-22 Section 9: "If the length does not match the length of the
2333 /// Message Body, the receiver MUST close the session with a
2334 /// PROTOCOL_VIOLATION."
2335 ///
2336 /// Without the trailing-byte check in `decode` this SUBSCRIBE parses and
2337 /// the two surplus bytes vanish:
2338 ///
2339 /// ```text
2340 /// assertion `left == right` failed
2341 /// left: Ok(Subscribe(Subscribe { request_id: VarInt(1), track_namespace:
2342 /// TrackNamespace([[97]]), track_name: [98], parameters: [] }))
2343 /// right: Err(InvalidField)
2344 /// ```
2345 #[test]
2346 fn a_message_body_shorter_than_the_declared_length_is_refused() {
2347 let body = subscribe_body(&[0x00]);
2348 let mut junked = body.clone();
2349 junked.extend_from_slice(&[0xff, 0xff]);
2350 let bytes = frame_with_declared_len(0x03, (body.len() + 2) as u16, &junked);
2351
2352 let mut buf = &bytes[..];
2353 assert_eq!(
2354 ControlMessage::decode(&mut buf),
2355 Err(CodecError::ControlMessageLengthMismatch {
2356 declared: (body.len() + 2),
2357 detail: "its fields left bytes unread",
2358 })
2359 );
2360
2361 // The same body with an honest length still decodes, so the guard
2362 // rejects the mismatch and not the message.
2363 let honest = frame(0x03, &body);
2364 let mut buf = &honest[..];
2365 assert!(ControlMessage::decode(&mut buf).is_ok());
2366 }
2367
2368 /// Draft-22 Section 8.5: "The reason phrase length has a maximum value of
2369 /// 1024 bytes. If an endpoint receives a length exceeding the maximum, it
2370 /// MUST close the session with a PROTOCOL_VIOLATION".
2371 ///
2372 /// Without the decode-side bound the 2000-byte phrase is handed to the
2373 /// application:
2374 ///
2375 /// ```text
2376 /// assertion `left == right` failed
2377 /// left: Ok(RequestError(RequestError { error_code: VarInt(1),
2378 /// retry_interval: VarInt(0), reason_phrase: [120, 120, ...],
2379 /// redirect: None }))
2380 /// right: Err(ReasonPhraseTooLong)
2381 /// ```
2382 ///
2383 /// (The 2000 repeated bytes of the phrase are elided from that transcript.)
2384 #[test]
2385 fn an_over_long_reason_phrase_is_refused_on_decode() {
2386 for (type_id, prefix) in [(0x05u64, vec![0x01, 0x00]), (0x0B, vec![0x01, 0x00])] {
2387 let mut body = prefix;
2388 let over = MAX_REASON_PHRASE_LENGTH + 976;
2389 VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
2390 body.extend(std::iter::repeat_n(b'x', over));
2391 let bytes = frame(type_id, &body);
2392
2393 let mut buf = &bytes[..];
2394 assert_eq!(
2395 ControlMessage::decode(&mut buf),
2396 Err(CodecError::ReasonPhraseTooLong),
2397 "message type 0x{type_id:x}"
2398 );
2399 }
2400 }
2401
2402 /// Draft-22 Section 9.2: "The maximum length of the New Session URI is
2403 /// 8,192 bytes. If an endpoint receives a length exceeding the maximum, it
2404 /// MUST close the session with a PROTOCOL_VIOLATION."
2405 ///
2406 /// Without the decode-side bound the oversized URI reaches the application
2407 /// and a migrating endpoint follows it:
2408 ///
2409 /// ```text
2410 /// assertion `left == right` failed
2411 /// left: Ok(GoAway(GoAway { new_session_uri: [117, 117, ...],
2412 /// timeout: VarInt(0) }))
2413 /// right: Err(GoAwayUriTooLong)
2414 /// ```
2415 ///
2416 /// (The 9000 repeated bytes of the URI are elided from that transcript.)
2417 #[test]
2418 fn an_over_long_goaway_uri_is_refused_on_decode() {
2419 let over = MAX_GOAWAY_URI_LENGTH + 808;
2420 let mut body = Vec::new();
2421 VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
2422 body.extend(std::iter::repeat_n(b'u', over));
2423 body.push(0x00); // timeout
2424 let bytes = frame(0x10, &body);
2425
2426 let mut buf = &bytes[..];
2427 assert_eq!(ControlMessage::decode(&mut buf), Err(CodecError::GoAwayUriTooLong));
2428 }
2429
2430 /// Draft-22 Section 9.20.8 (GROUP_ORDER): "The allowed values are Ascending
2431 /// (0x1) or Descending (0x2). If an endpoint receives a value outside this
2432 /// range, it MUST close the session with PROTOCOL_VIOLATION." Section
2433 /// 9.20.18 says the same of FORWARD with the values 0 and 1.
2434 ///
2435 /// Without `uint8_value_in_range` the out-of-range byte is handed up as an
2436 /// ordinary parameter:
2437 ///
2438 /// ```text
2439 /// assertion `left == right` failed
2440 /// left: Ok(Subscribe(Subscribe { request_id: VarInt(1), track_namespace:
2441 /// TrackNamespace([[97]]), track_name: [98], parameters:
2442 /// [KeyValuePair { key: VarInt(34), value: Varint(VarInt(7)) }] }))
2443 /// right: Err(InvalidField)
2444 /// ```
2445 #[test]
2446 fn a_uint8_parameter_outside_its_range_is_refused() {
2447 // key, rejected value, accepted value
2448 let cases = [(0x22u8, 7u8, 2u8), (0x10, 9, 1)];
2449 for (key, bad, good) in cases {
2450 let bytes = frame(0x03, &subscribe_body(&[0x01, key, bad]));
2451 let mut buf = &bytes[..];
2452 assert_eq!(
2453 ControlMessage::decode(&mut buf),
2454 Err(CodecError::ParameterValueOutOfRange { key: key as u64, value: bad as u64 }),
2455 "parameter 0x{key:x} value {bad}"
2456 );
2457
2458 let bytes = frame(0x03, &subscribe_body(&[0x01, key, good]));
2459 let mut buf = &bytes[..];
2460 assert!(
2461 ControlMessage::decode(&mut buf).is_ok(),
2462 "parameter 0x{key:x} value {good} should still decode"
2463 );
2464 }
2465 }
2466
2467 /// SUBSCRIBER_PRIORITY (0x20) is a uint8 with no restricted range, so it
2468 /// must keep accepting the whole 0-255 span. This is the negative half of
2469 /// the range check: a table that over-reached would fail here.
2470 #[test]
2471 fn subscriber_priority_still_accepts_the_whole_byte_range() {
2472 for value in [0u8, 1, 2, 128, 255] {
2473 let bytes = frame(0x03, &subscribe_body(&[0x01, 0x20, value]));
2474 let mut buf = &bytes[..];
2475 assert!(ControlMessage::decode(&mut buf).is_ok(), "priority {value}");
2476 }
2477 }
2478
2479 fn param(key: u64, value: &[u8]) -> KeyValuePair {
2480 KeyValuePair { key: VarInt::from_u64_moqt(key), value: KvpValue::Bytes(value.to_vec()) }
2481 }
2482
2483 /// Draft-22 Section 9.20.17: "The LARGEST_OBJECT parameter (Parameter Type
2484 /// 0x9) is a Location." Section 9.20 defines Location as "Two consecutive
2485 /// varints (Group, Object)" — the value carries no length of its own.
2486 ///
2487 /// The frame below is built from the draft rather than from this encoder:
2488 /// REQUEST_OK, four body bytes, one parameter, type delta `0x09`, then the
2489 /// two varints `0x0a` and `0x03` for Location (10, 3). A length-prefixed
2490 /// spelling would need a fifth byte.
2491 ///
2492 /// With `0x09` back in the Length-prefixed arm, the decoder reads the
2493 /// group varint `0x0a` as a value length of 10 and runs off the end of a
2494 /// four-byte body. Both this test and
2495 /// [`a_location_does_not_eat_the_block_that_follows_it`] fail with:
2496 ///
2497 /// ```text
2498 /// spec-correct frame must decode: UnexpectedEnd
2499 /// ```
2500 #[test]
2501 fn largest_object_is_two_bare_varints() {
2502 let body = [0x01, 0x09, 0x0a, 0x03];
2503 let bytes = frame(0x07, &body);
2504
2505 let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2506 let ControlMessage::RequestOk(ok) = &msg else {
2507 panic!("expected REQUEST_OK, got {msg:?}")
2508 };
2509 assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
2510 assert!(ok.track_properties.is_empty(), "the four body bytes are all parameter");
2511
2512 let mut out = Vec::new();
2513 msg.encode(&mut out).expect("re-encode");
2514 assert_eq!(out, bytes, "the value must go back out as the two bare varints it came in as");
2515 }
2516
2517 /// A Location value the encoder was handed but the decoder could not read
2518 /// back is refused on the way out, not written.
2519 ///
2520 /// LARGEST_OBJECT carries no length of its own — that is the whole point
2521 /// of the encoding — so `encode_parameters` writes its bytes verbatim. A
2522 /// value built in memory rather than decoded is under no obligation to be
2523 /// two varints, and without this check the codec answers `Ok(())` and puts
2524 /// a frame on the wire that `ControlMessage::decode` then refuses. One
2525 /// varint short and one varint long are the two ways to get it wrong.
2526 ///
2527 /// # What it catches
2528 ///
2529 /// Dropping the `is_location_value` guard from this draft's encode arm,
2530 /// run:
2531 ///
2532 /// ```text
2533 /// panicked at crates\moqtap-codec\src\draft22\message.rs:
2534 /// LARGEST_OBJECT of one varint must not encode: the decoder cannot read it back
2535 ///
2536 /// test result: FAILED. 1 passed; 1 failed; 0 ignored; 0 measured; 108 filtered out
2537 /// ```
2538 ///
2539 /// The sibling draft kept its guard and kept passing, which is what shows
2540 /// the check is per-draft and not inherited from somewhere shared.
2541 #[test]
2542 fn a_location_value_that_is_not_two_varints_is_refused_on_encode() {
2543 for (label, value) in
2544 [("one varint", vec![0x0a]), ("three varints", vec![0x0a, 0x03, 0x05])]
2545 {
2546 let msg = ControlMessage::RequestOk(RequestOk {
2547 parameters: vec![param(0x09, &value)],
2548 track_properties: Vec::new(),
2549 });
2550 let mut out = Vec::new();
2551 assert!(
2552 msg.encode(&mut out).is_err(),
2553 "LARGEST_OBJECT of {label} must not encode: the decoder cannot read it back"
2554 );
2555 }
2556
2557 // The well-formed value still goes out, so the check refuses the
2558 // malformed case and not the encoding itself.
2559 let msg = ControlMessage::RequestOk(RequestOk {
2560 parameters: vec![param(0x09, &[0x0a, 0x03])],
2561 track_properties: Vec::new(),
2562 });
2563 let mut out = Vec::new();
2564 msg.encode(&mut out).expect("a Location of exactly two varints must still encode");
2565 ControlMessage::decode(&mut &out[..]).expect("and must decode back");
2566 }
2567
2568 /// The same Location read through a message that carries other fields
2569 /// after it, so a stray length byte cannot hide in a trailing block.
2570 ///
2571 /// SUBSCRIBE_OK is track alias `0x05`, then the parameters, then the track
2572 /// properties. With LARGEST_OBJECT (10, 3) and one property
2573 /// (OBJECT_DELIVERY_TIMEOUT, type `0x02`, 5000ms as the two-byte varint
2574 /// `0x93 0x88`), the body is `05 01 09 0a 03 02 93 88`.
2575 #[test]
2576 fn a_location_does_not_eat_the_block_that_follows_it() {
2577 let body = [0x05, 0x01, 0x09, 0x0a, 0x03, 0x02, 0x93, 0x88];
2578 let bytes = frame(0x04, &body);
2579
2580 let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2581 let ControlMessage::SubscribeOk(ok) = &msg else {
2582 panic!("expected SUBSCRIBE_OK, got {msg:?}")
2583 };
2584 assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
2585 assert_eq!(
2586 ok.track_properties,
2587 vec![KeyValuePair {
2588 key: VarInt::from_u64_moqt(0x02),
2589 value: KvpValue::Varint(VarInt::from_u64_moqt(5000)),
2590 }]
2591 );
2592
2593 let mut out = Vec::new();
2594 msg.encode(&mut out).expect("re-encode");
2595 assert_eq!(out, bytes);
2596 }
2597
2598 /// Draft-22 Section 9.20.20: the TRACK_NAMESPACE_PREFIX parameter
2599 /// (Parameter Type 0x34) "uses the Track Namespace encoding described in
2600 /// Section 8.7" — a varint field count followed by that many
2601 /// length-prefixed fields, and nothing in front of it. That encoding is not
2602 /// one of the four Section 9.20 lists, so it cannot be assumed to be
2603 /// Length-prefixed by default.
2604 ///
2605 /// The frame below is built from Section 8.7: REQUEST_UPDATE for request
2606 /// `7`, one parameter, type delta `0x34`, then the namespace ("live",
2607 /// "sports") as `02 04 "live" 06 "sports"`. Sixteen body bytes; a
2608 /// length-prefixed spelling would need a seventeenth for the outer length.
2609 ///
2610 /// With `0x34` back in the Length-prefixed arm the field count `0x02` is
2611 /// read as an outer length of two bytes, leaving eleven bytes of namespace
2612 /// unread. Draft-22's Section 9 body-length check turns that into a
2613 /// refusal rather than a truncated value:
2614 ///
2615 /// ```text
2616 /// spec-correct frame must decode: InvalidField
2617 /// ```
2618 ///
2619 /// That check is not a safety net here. Where the surplus lands inside the
2620 /// declared body — as in
2621 /// [`an_empty_track_namespace_prefix_is_one_zero_byte`], whose namespace is
2622 /// one byte long — the misread is silent, and that test fails instead with:
2623 ///
2624 /// ```text
2625 /// assertion `left == right` failed
2626 /// left: [KeyValuePair { key: VarInt(52), value: Bytes([]) }]
2627 /// right: [KeyValuePair { key: VarInt(52), value: Bytes([0]) }]
2628 /// ```
2629 #[test]
2630 fn track_namespace_prefix_is_a_bare_track_namespace() {
2631 let namespace: Vec<u8> = [&[0x02, 0x04][..], b"live", &[0x06][..], b"sports"].concat();
2632 assert_eq!(namespace.len(), 13);
2633
2634 let body: Vec<u8> = [&[0x07, 0x01, 0x34][..], &namespace].concat();
2635 assert_eq!(body.len(), 16);
2636 let bytes = frame(0x02, &body);
2637
2638 let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2639 let ControlMessage::RequestUpdate(update) = &msg else {
2640 panic!("expected REQUEST_UPDATE, got {msg:?}")
2641 };
2642 assert_eq!(update.request_id.into_inner(), 7);
2643 assert_eq!(update.parameters, vec![param(0x34, &namespace)]);
2644
2645 let mut out = Vec::new();
2646 msg.encode(&mut out).expect("re-encode");
2647 assert_eq!(out, bytes, "no outer length may appear in front of the Track Namespace");
2648 }
2649
2650 /// An empty prefix is a legal Track Namespace: Section 2.4.1 puts one at
2651 /// "between 0 and 32 Track Namespace Fields". On the wire that is the
2652 /// single byte `0x00`, and it must not be confused with a length-prefixed
2653 /// value of zero bytes.
2654 #[test]
2655 fn an_empty_track_namespace_prefix_is_one_zero_byte() {
2656 let body = [0x07, 0x01, 0x34, 0x00];
2657 let bytes = frame(0x02, &body);
2658
2659 let msg = ControlMessage::decode(&mut &bytes[..]).expect("empty prefix must decode");
2660 let ControlMessage::RequestUpdate(update) = &msg else {
2661 panic!("expected REQUEST_UPDATE, got {msg:?}")
2662 };
2663 assert_eq!(update.parameters, vec![param(0x34, &[0x00])]);
2664
2665 let mut out = Vec::new();
2666 msg.encode(&mut out).expect("re-encode");
2667 assert_eq!(out, bytes);
2668 }
2669}