moqtap_codec/draft19/message.rs
1//! Draft-19 control message encoding and decoding.
2//!
3//! Key differences from draft-18:
4//! - `Request ID` field removed from GOAWAY entirely; the control-stream and
5//! request-stream forms are now identical.
6//! - New Range Filter parameters (length-prefixed): SUBGROUP_FILTER (0x25),
7//! OBJECTID_FILTER (0x26), PRIORITY_FILTER (0x27), OBJECT_PROPERTY_FILTER
8//! (0x28) and TRACK_PROPERTY_FILTER (0x29).
9//! - New Setup Options MAX_FILTER_RANGES (0x06) and MAX_REQUEST_UPDATES (0x08);
10//! both are even KVP types carrying a varint value.
11//! - GROUP_ORDER (0x22) moves from PUBLISH_OK to SUBSCRIBE_TRACKS (the wire
12//! encoding of the parameter is unchanged).
13//! - PUBLISH_BLOCKED renamed to PUBLISH_SKIPPED (still type 0x0F; wire
14//! identical).
15//! - SUBSCRIPTION_FILTER renamed to LOCATION_FILTER (still parameter 0x21).
16//! - REQUEST_ERROR adds CONFLICTING_FILTERS (0x35) and INVALID_FILTER (0x36);
17//! DUPLICATE_SUBSCRIPTION (0x19) is removed.
18//! - The framing field after Message Length is named Message Body (draft-19
19//! Section 10, Figure 3); earlier drafts called it Message Payload. The
20//! change is editorial, so the bytes are unchanged, but this module uses the
21//! new name.
22
23use crate::auth_token::{AuthorizationToken, AUTH_TOKEN_PARAMETER};
24use crate::error::MAX_FULL_TRACK_NAME_LENGTH;
25pub use crate::error::{
26 CodecError, MAX_GOAWAY_URI_LENGTH, MAX_MESSAGE_LENGTH, MAX_NAMESPACE_TUPLE_SIZE,
27 MAX_REASON_PHRASE_LENGTH,
28};
29use crate::kvp::{KeyValuePair, KvpError, KvpValue, MAX_KVP_VALUE_LEN};
30use crate::subscription_filter::{SubscriptionFilter, SUBSCRIPTION_FILTER_PARAMETER};
31use crate::types::check_location_range;
32use crate::types::*;
33use crate::varint::{Moqt18 as Wire, VarInt};
34use bytes::{Buf, BufMut};
35
36// ============================================================
37// Parameter encoding helpers for draft-19
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-19 Section 2.4.1: a varint field
52 /// count followed by that many length-prefixed fields.
53 ///
54 /// Not one of the four value encodings draft-19 Section 10.2 lists. A
55 /// parameter definition is free to name an encoding from elsewhere in the
56 /// document, and TRACK_NAMESPACE_PREFIX does exactly that; the field count
57 /// is the only length the wire carries.
58 TrackNamespaceValue,
59}
60
61fn param_encoding(key: u64) -> Option<ParamEncoding> {
62 match key {
63 // 0x02 = OBJECT_DELIVERY_TIMEOUT (DELIVERY_TIMEOUT on drafts 11-17)
64 // 0x04 = RENDEZVOUS_TIMEOUT (draft-19 Section 10.2.6). Not
65 // MAX_CACHE_DURATION: that is Property Type 0x04 in the
66 // separate Properties registry (Section 15.8), a different
67 // namespace that happens to reuse the number.
68 // 0x06 = SUBGROUP_DELIVERY_TIMEOUT (new in draft-18)
69 // 0x08 = EXPIRES
70 // 0x0A = FILL_TIMEOUT (new in draft-18, FETCH only)
71 // 0x32 = NEW_GROUP_REQUEST
72 0x02 | 0x04 | 0x06 | 0x08 | 0x0A | 0x32 => Some(ParamEncoding::Varint),
73 // 0x10 = FORWARD, 0x20 = SUBSCRIBER_PRIORITY, 0x22 = GROUP_ORDER
74 0x10 | 0x20 | 0x22 => Some(ParamEncoding::Uint8),
75 // 0x09 = LARGEST_OBJECT. Draft-19 Section 10.2.16: "The LARGEST_OBJECT
76 // parameter (Parameter Type 0x9) is a Location." A Location is
77 // two consecutive varints (Section 10.2), with no length ahead
78 // of them.
79 0x09 => Some(ParamEncoding::Location),
80 // 0x34 = TRACK_NAMESPACE_PREFIX. Section 10.2.19: it "uses the Track
81 // Namespace encoding described in Section 2.4.1".
82 0x34 => Some(ParamEncoding::TrackNamespaceValue),
83 // 0x03 = AUTHORIZATION_TOKEN
84 // 0x21 = LOCATION_FILTER (SUBSCRIPTION_FILTER on drafts 15-18)
85 // 0x25 = SUBGROUP_FILTER, 0x26 = OBJECTID_FILTER, 0x27 = PRIORITY_FILTER,
86 // 0x28 = OBJECT_PROPERTY_FILTER, 0x29 = TRACK_PROPERTY_FILTER
87 // (Range Filters, new in draft-19)
88 0x03 | 0x21 | 0x25 | 0x26 | 0x27 | 0x28 | 0x29 => Some(ParamEncoding::LengthPrefixed),
89 _ => None,
90 }
91}
92
93/// Whether `value` is inside the range draft-19 allows for a uint8-valued
94/// parameter.
95///
96/// Two of the three uint8 parameters restrict their range and say the receiver
97/// MUST close the session with PROTOCOL_VIOLATION on anything outside it:
98/// GROUP_ORDER allows only Ascending (0x1) and Descending (0x2) (Section
99/// 10.2.8), and FORWARD allows only 0 and 1 (Section 10.2.17).
100/// SUBSCRIBER_PRIORITY (Section 10.2.7) uses the whole 0-255 range, so it has
101/// no entry here.
102///
103/// Range-checking on decode is what makes the values usable: an application
104/// that tests `group_order == 2` for descending would otherwise treat 7 as
105/// neither ascending nor descending and carry on.
106fn uint8_value_in_range(key: u64, value: u8) -> bool {
107 match key {
108 // FORWARD (0x10)
109 0x10 => value <= 1,
110 // GROUP_ORDER (0x22)
111 0x22 => value == 1 || value == 2,
112 _ => true,
113 }
114}
115
116const AUTHORIZATION_TOKEN: u64 = 0x03;
117
118/// Whether a message may carry `key` more than once.
119///
120/// Section 10.2 states the default: "Senders MUST NOT repeat the same Parameter
121/// Type in a message unless the parameter definition explicitly allows multiple
122/// instances of that type to be sent in a single message." Two definitions do.
123///
124/// * `AUTHORIZATION_TOKEN` (0x03), Section 10.2.2: "The AUTHORIZATION TOKEN
125/// parameter MAY be repeated within a message as long as the combination of
126/// Token Type and Token Value are unique after resolving any aliases."
127/// * The five Range Filters (0x25 through 0x29), Section 5.1.3: "The Track
128/// Property filter parameter MAY appear multiple times in a SUBSCRIBE_TRACKS
129/// message or REQUEST_UPDATE for it. All other filter parameters MAY appear
130/// multiple times in a FETCH, SUBSCRIBE, SUBSCRIBE_TRACKS, PUBLISH_OK, or
131/// REQUEST_UPDATE (on a subscription, from the subscriber only) message."
132///
133/// # A zero `Type Delta` is "the same type again", not an error
134///
135/// The two rules interact, and **the draft does not say how**. Section 10.2 also
136/// requires that "Parameters MUST be serialized in ascending order by Type", so
137/// a second instance of a repeatable type produces a `Type Delta` of 0 — well
138/// formed only if the decoder reads a zero delta as a repeat rather than as a
139/// malformation. That reading is the one taken here; the alternative makes
140/// Section 5.1.3's permission unusable, because there is no other encoding for
141/// a second filter of the same type.
142///
143/// # Why the permission is not decoration
144///
145/// "All filter parameters with the same SetID value are combined using logical
146/// 'AND' operations, then all the resulting sets are combined using logical
147/// 'OR' operations." One filter parameter carries one SetID, so a subscriber
148/// asking for two alternatives that each constrain the same field — Subgroup 1
149/// to 3 at low priority, or Subgroup 10 to 12 at high — has to send SUBGROUP_
150/// FILTER twice, once per set. Refusing the repeat does not narrow what a peer
151/// can express; it collapses the set lattice to one filter per type and turns
152/// conforming traffic into a session close.
153///
154/// What the filters may **not** do is repeat the same (Parameter Type, SetID,
155/// Property Type) triple, and Section 5.1.3 answers that with a REQUEST_ERROR
156/// carrying INVALID_FILTER rather than with a session close. A reply an endpoint
157/// sends is not a frame a decoder refuses, so nothing here enforces it — see
158/// [`crate::range_filter`] for the reader an endpoint uses to decide.
159fn parameter_may_repeat(key: u64) -> bool {
160 matches!(key, AUTHORIZATION_TOKEN | 0x25..=0x29)
161}
162
163/// Add a delta to the previous delta-encoded key.
164///
165/// Draft-19 Section 1.4.3: "The previous Type value plus the Delta Type MUST NOT
166/// be greater than 2^64 - 1. If a Delta Type is received that would be too
167/// large, the Session MUST be closed with a PROTOCOL_VIOLATION." MoQT varints
168/// span the whole 64-bit range, so a peer can drive the sum past the end: a
169/// debug build panicked on the addition and a release build wrapped the key and
170/// reported the parameter under a type its sender never wrote.
171fn add_delta(prev_key: u64, delta: u64) -> Result<u64, CodecError> {
172 prev_key.checked_add(delta).ok_or(CodecError::KeyDeltaOverflow(prev_key, delta))
173}
174
175/// Hold a namespace-plus-name pair to the Full Track Name cap.
176///
177/// Draft-19 Section 2.4.1: "The maximum total length of a Full Track Name is
178/// 4,096 bytes. The length of a Full Track Name is computed as the sum of the
179/// Track Namespace Field Length fields and the Track Name Length field... If an
180/// endpoint receives a Track Namespace or a Full Track Name exceeding 4,096
181/// bytes, it MUST close the session with a PROTOCOL_VIOLATION."
182///
183/// The namespace half of that sentence is enforced inside the namespace decoder,
184/// which is the only place that sees a namespace with no name beside it. This is
185/// the other half, and it has to live where the two are decoded together: a
186/// namespace at 4,000 bytes and a name at 500 are each legal alone.
187fn check_full_track_name(namespace: &TrackNamespace, track_name: &[u8]) -> Result<(), CodecError> {
188 let total = namespace.field_bytes_len().saturating_add(track_name.len());
189 if total > MAX_FULL_TRACK_NAME_LENGTH {
190 return Err(CodecError::TrackNameTooLong);
191 }
192 Ok(())
193}
194
195/// Hold every AUTHORIZATION TOKEN parameter to the Token structure it names.
196///
197/// Section 10.2.2: "If the Token structure cannot be decoded, the receiver
198/// MUST close the Session with KEY_VALUE_FORMATTING_ERROR." That is the answer
199/// Section 1.4.3 gives for any Type whose value does not match the
200/// serialization that Type defines; the Token is the one structure this draft
201/// spells out, and the only parameter value in it that is more than opaque
202/// bytes.
203///
204/// Both namespaces carry the type on this draft, and both reach here.
205///
206/// A type this draft cannot name is left alone. The rule is conditional on the
207/// receiver understanding the Type, and an extension's parameter carries bytes
208/// no rule here describes.
209fn check_authorization_tokens(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
210 for parameter in parameters {
211 let key = parameter.key.into_inner();
212 if key != AUTH_TOKEN_PARAMETER {
213 continue;
214 }
215 match ¶meter.value {
216 KvpValue::Bytes(value) => {
217 AuthorizationToken::decode_moqt::<Wire>(key, value)?;
218 }
219 // Unreachable from the decoder, which picks the shape from the
220 // type and finds this one length-prefixed. A caller that built the
221 // pair in memory can still get here, and it is the same rule: the
222 // value is not the serialization the type defines.
223 KvpValue::Varint(_) => {
224 return Err(CodecError::KeyValueFormatting {
225 key,
226 detail: "its value is a bare varint where the type defines a Token structure",
227 });
228 }
229 }
230 }
231 Ok(())
232}
233
234/// Hold every LOCATION_FILTER parameter to the filter structure it names.
235///
236/// Section 5.1.2: "An endpoint that receives a filter type other than the above
237/// MUST close the session with PROTOCOL_VIOLATION." Section 10.2.9 defines the
238/// parameter, which this draft renamed from SUBSCRIPTION_FILTER to
239/// LOCATION_FILTER when it added the Range Filters beside it. The number, 0x21,
240/// and the structure are the ones draft-18 had.
241///
242/// Drafts 15 and 16 stated the length rule of this parameter directly, at
243/// draft-16 Section 9.2.2.5 — "If the length of the Subscription Filter does
244/// not match the parameter length, the publisher MUST close the session with
245/// PROTOCOL_VIOLATION." Draft-17 dropped that sentence, and what answers the
246/// same malformation here is the general rule of Section 1.4.3, which names
247/// KEY_VALUE_FORMATTING_ERROR. Same malformation, different code, and the
248/// session table is where the two part.
249///
250/// The End Group is a delta, and this draft states what happens when resolving
251/// it leaves the number space: "the last Group ID to be delivered
252/// will be the Group ID in Start Location plus the End Group Delta. If the
253/// resulting Group ID would be greater than 2^64 - 1, the endpoint MUST close
254/// the session with a PROTOCOL_VIOLATION." That is why the sum is taken here and
255/// not left to the caller — draft-17, which introduced the delta and states no
256/// such sentence, does not take it.
257///
258/// The filter is otherwise decoded and discarded. What is kept is the refusal —
259/// the value stays on the parameter as the bytes that arrived, so a caller reads
260/// it through [`SubscriptionFilter::decode_moqt`] when it wants the filter
261/// rather than the frame.
262fn check_subscription_filters(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
263 for parameter in parameters {
264 if parameter.key.into_inner() != SUBSCRIPTION_FILTER_PARAMETER {
265 continue;
266 }
267 match ¶meter.value {
268 KvpValue::Bytes(value) => {
269 SubscriptionFilter::decode_moqt::<Wire>(value)?.last_group()?;
270 }
271 // Unreachable from the decoder, which picks the shape from the type
272 // and finds this one length-prefixed. A caller that built the pair
273 // in memory can still get here, and it is the same rule.
274 KvpValue::Varint(_) => {
275 return Err(CodecError::SubscriptionFilterMalformed {
276 detail: "its value is a bare varint where the type defines a filter",
277 });
278 }
279 }
280 }
281 Ok(())
282}
283
284/// Decode a count-prefixed list of parameters with delta-encoded types.
285fn decode_parameters(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
286 let count = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
287 let mut params = crate::types::reserve_bounded(count, buf);
288 let mut prev_key: u64 = 0;
289
290 for i in 0..count {
291 let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
292 let abs_key = add_delta(prev_key, delta)?;
293 // Types ascend, so a repeat is always a zero delta against the
294 // parameter before it. Draft-19 Section 10.2: "Receivers SHOULD check
295 // that there are no unexpected duplicate parameters and close the
296 // session with PROTOCOL_VIOLATION if found." Downstream code that scans
297 // the list for a key takes whichever copy it meets first, so two
298 // implementations reading one frame can pick opposite values.
299 //
300 // "Unexpected" is what `parameter_may_repeat` reads: a zero delta on a
301 // type whose own definition permits repeats is the second instance,
302 // which is the only encoding such an instance has.
303 if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
304 return Err(CodecError::DuplicateParameter(abs_key));
305 }
306 prev_key = abs_key;
307
308 // Section 10.2: "All Message Parameters MUST be defined in the
309 // negotiated version of MOQT or negotiated via Setup Options. An
310 // endpoint that receives an unknown Message Parameter MUST close the
311 // session with PROTOCOL_VIOLATION. Because the receiver has to
312 // understand every Message Parameter, there is no need for a mechanism
313 // to skip unknown parameters." Because unknown parameters
314 // cannot be skipped, the block is bounded by a parameter count rather
315 // than a length.
316 //
317 // The table this consults is the registry's, so a type it cannot name
318 // is one this draft does not define. Reporting it as an ordinary
319 // malformation, which is what it did before, left the rule enforced
320 // against the frame and invisible to the session.
321 let encoding =
322 param_encoding(abs_key).ok_or(CodecError::UnknownMessageParameter(abs_key))?;
323
324 let value = match encoding {
325 ParamEncoding::Varint => {
326 let v = VarInt::decode_moqt::<Wire>(buf)?;
327 KvpValue::Varint(v)
328 }
329 ParamEncoding::Uint8 => {
330 if buf.remaining() < 1 {
331 return Err(CodecError::UnexpectedEnd);
332 }
333 let byte = buf.get_u8();
334 if !uint8_value_in_range(abs_key, byte) {
335 return Err(CodecError::ParameterValueOutOfRange {
336 key: abs_key,
337 value: byte as u64,
338 });
339 }
340 KvpValue::Varint(VarInt::from_u64_moqt(byte as u64))
341 }
342 ParamEncoding::Location => {
343 let group = VarInt::decode_moqt::<Wire>(buf)?;
344 let object = VarInt::decode_moqt::<Wire>(buf)?;
345 let mut encoded = Vec::new();
346 group.encode_moqt::<Wire>(&mut encoded);
347 object.encode_moqt::<Wire>(&mut encoded);
348 KvpValue::Bytes(encoded)
349 }
350 ParamEncoding::LengthPrefixed => {
351 let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
352 let data = read_bytes(buf, len)?;
353 KvpValue::Bytes(data)
354 }
355 ParamEncoding::TrackNamespaceValue => {
356 // A prefix of zero fields is legal: Section 2.4.1 puts a Track
357 // Namespace at "between 0 and 32 Track Namespace Fields", and
358 // an empty prefix matches every namespace.
359 let ns = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
360 let mut encoded = Vec::new();
361 ns.encode_moqt::<Wire>(&mut encoded);
362 KvpValue::Bytes(encoded)
363 }
364 };
365
366 params.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
367 }
368 check_authorization_tokens(¶ms)?;
369 check_subscription_filters(¶ms)?;
370 Ok(params)
371}
372
373/// Whether `bytes` is exactly the wire form of a Location — two consecutive
374/// varints and nothing after them.
375///
376/// `decode_parameters` builds this value by reading two varints and
377/// re-serialising them, so every value it produces satisfies this. A value
378/// built in memory need not, and the encode arm writes these bytes verbatim
379/// because a Location carries no length of its own. Without this check a
380/// caller could hand over one varint, or three, and the codec would put a
381/// frame on the wire that its own decoder answers with an error.
382fn is_location_value(bytes: &[u8]) -> bool {
383 let mut buf = bytes;
384 VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
385 && VarInt::decode_moqt::<Wire>(&mut buf).is_ok()
386 && !buf.has_remaining()
387}
388
389/// Whether `bytes` is exactly the wire form of a Track Namespace, with
390/// nothing after it. The same reasoning as [`is_location_value`]: the value
391/// goes out verbatim, so it has to be something this draft can read back.
392fn is_track_namespace_value(bytes: &[u8]) -> bool {
393 let mut buf = bytes;
394 TrackNamespace::decode_allow_empty_moqt::<Wire>(&mut buf).is_ok() && !buf.has_remaining()
395}
396
397/// Encode a count-prefixed list of parameters with delta-encoded types.
398///
399/// Errors with [`CodecError::InvalidField`] on a uint8-valued parameter whose
400/// value [`decode_parameters`] would refuse, so the two directions accept the
401/// same set of frames.
402///
403/// The check is not a mirror added for tidiness. A uint8 parameter's value is
404/// written as one octet, and a value that does not fit one is otherwise
405/// truncated to its low byte: GROUP_ORDER 258 becomes the byte 0x02, which is
406/// Descending — a well-formed frame carrying a value the caller never asked
407/// for, and one no receiver could tell from a genuine Descending. Refusing is
408/// the only outcome that does not silently rewrite the message.
409///
410/// The two structure rules are here for a plainer reason. A value under a type
411/// that defines a structure and is not that structure — a Token, a filter — is
412/// one the receiver must close the session over, so writing it is not a way to
413/// send it; the sender's first sign of trouble would be the session going.
414fn encode_parameters(params: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
415 check_authorization_tokens(params)?;
416 check_subscription_filters(params)?;
417 VarInt::from_usize(params.len()).encode_moqt::<Wire>(buf);
418 let mut prev_key: u64 = 0;
419
420 for (i, p) in params.iter().enumerate() {
421 let abs_key = p.key.into_inner();
422 // The delta is a difference, so a descending pair wraps the subtraction
423 // into a nine-byte delta the peer resolves to an unrelated key, and a
424 // repeated type is a frame `decode_parameters` refuses. Both are
425 // refused here so the two directions accept the same set of frames.
426 let delta = abs_key
427 .checked_sub(prev_key)
428 .ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
429 if i > 0 && delta == 0 && !parameter_may_repeat(abs_key) {
430 return Err(CodecError::DuplicateParameter(abs_key));
431 }
432 prev_key = abs_key;
433 VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
434
435 // The same maximum the decoder below applies, and the same one this
436 // draft's Setup Option encoder has always applied: "The maximum length
437 // of a value is 2^16-1 bytes. If an endpoint receives a length larger
438 // than the maximum, it MUST close the session with a PROTOCOL_VIOLATION."
439 // A value past it is one the peer must end the session over, so writing
440 // it is not a way to send it.
441 //
442 // Hoisted above the shape table rather than repeated inside it: a
443 // Location is bytes as well, and one past the maximum is not a Location.
444 if let KvpValue::Bytes(b) = &p.value {
445 if b.len() > MAX_KVP_VALUE_LEN {
446 return Err(KvpError::ValueTooLong(b.len()).into());
447 }
448 }
449
450 let encoding = param_encoding(abs_key);
451 match (&p.value, encoding) {
452 (KvpValue::Varint(v), Some(ParamEncoding::Varint)) => {
453 v.encode_moqt::<Wire>(buf);
454 }
455 (KvpValue::Varint(v), Some(ParamEncoding::Uint8)) => {
456 let raw = v.into_inner();
457 let byte = u8::try_from(raw).map_err(|_| CodecError::InvalidField)?;
458 if !uint8_value_in_range(abs_key, byte) {
459 return Err(CodecError::ParameterValueOutOfRange {
460 key: abs_key,
461 value: byte as u64,
462 });
463 }
464 buf.put_u8(byte);
465 }
466 // Both values are already stored in their own wire form — two
467 // varints for a Location, a field count and its fields for a Track
468 // Namespace — so they go out as they are. Adding a length here is
469 // the bug these arms exist to avoid.
470 (KvpValue::Bytes(b), Some(ParamEncoding::Location)) => {
471 if !is_location_value(b) {
472 return Err(CodecError::InvalidField);
473 }
474 buf.put_slice(b);
475 }
476 (KvpValue::Bytes(b), Some(ParamEncoding::TrackNamespaceValue)) => {
477 if !is_track_namespace_value(b) {
478 return Err(CodecError::InvalidField);
479 }
480 buf.put_slice(b);
481 }
482 (KvpValue::Bytes(b), Some(ParamEncoding::LengthPrefixed)) => {
483 VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
484 buf.put_slice(b);
485 }
486 _ => {
487 // Fallback: encode as KVP even/odd
488 match &p.value {
489 KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
490 KvpValue::Bytes(b) => {
491 VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
492 buf.put_slice(b);
493 }
494 }
495 }
496 }
497 }
498 Ok(())
499}
500
501/// Decode delta-encoded KVPs with even/odd convention (for setup options
502/// and track properties). Read until buffer is exhausted.
503fn decode_kvp_delta(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
504 let mut pairs = Vec::new();
505 let mut prev_key: u64 = 0;
506
507 while buf.has_remaining() {
508 let delta = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
509 let abs_key = add_delta(prev_key, delta)?;
510 prev_key = abs_key;
511
512 let value = if abs_key.is_multiple_of(2) {
513 let v = VarInt::decode_moqt::<Wire>(buf)?;
514 KvpValue::Varint(v)
515 } else {
516 let len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
517 // Draft-19 Section 1.4.3: "The maximum length of a value is 2^16-1
518 // bytes. If an endpoint receives a length larger than the maximum,
519 // it MUST close the session with a PROTOCOL_VIOLATION." The
520 // standalone `KeyValuePair::decode` already enforces this; stating
521 // it here too means the two readers of the same wire shape answer
522 // the same way, rather than this one leaning on the caller having
523 // clipped the buffer to a control message first.
524 if len > MAX_KVP_VALUE_LEN {
525 return Err(KvpError::ValueTooLong(len).into());
526 }
527 let data = read_bytes(buf, len)?;
528 KvpValue::Bytes(data)
529 };
530
531 pairs.push(KeyValuePair { key: VarInt::from_u64_moqt(abs_key), value });
532 }
533 Ok(pairs)
534}
535
536/// Encode delta-encoded KVPs with even/odd convention.
537///
538/// Refuses a list that is not in ascending order by type, for the same reason
539/// [`encode_parameters`] does: the delta is a difference, and a descending pair
540/// wraps it into a nine-byte delta the peer resolves to an unrelated key.
541fn encode_kvp_delta(pairs: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
542 let mut prev_key: u64 = 0;
543 for p in pairs {
544 let abs_key = p.key.into_inner();
545 let delta = abs_key
546 .checked_sub(prev_key)
547 .ok_or(CodecError::ParametersOutOfOrder(prev_key, abs_key))?;
548 prev_key = abs_key;
549 VarInt::from_u64_moqt(delta).encode_moqt::<Wire>(buf);
550 match &p.value {
551 KvpValue::Varint(v) => v.encode_moqt::<Wire>(buf),
552 KvpValue::Bytes(b) => {
553 if b.len() > MAX_KVP_VALUE_LEN {
554 return Err(KvpError::ValueTooLong(b.len()).into());
555 }
556 VarInt::from_usize(b.len()).encode_moqt::<Wire>(buf);
557 buf.put_slice(b);
558 }
559 }
560 }
561 Ok(())
562}
563
564/// Immutable Properties, Property Type 0xB.
565///
566/// Section 12.7: Immutable Properties are "a Track or Object Property that
567/// contains a sequence of Key-Value-Pairs (see Figure 2) that are themselves
568/// Track or Object Properties, respectively". The Type is odd, so its value is
569/// length-prefixed bytes, and those bytes are another delta-typed run starting
570/// from 0.
571const IMMUTABLE_PROPERTIES: u64 = 0x0B;
572
573/// Whether `value` is inside the range draft-19 allows for a Track Property
574/// type that restricts one.
575///
576/// Two types do, and each answers anything outside its range with a session
577/// close. DEFAULT_PUBLISHER_GROUP_ORDER (0x22), Section 12.5: "The allowed
578/// values are Ascending (0x1) or Descending (0x2). If an endpoint receives a
579/// value outside this range, it MUST close the session with
580/// PROTOCOL_VIOLATION." DYNAMIC_GROUPS (0x30), Section 12.6: "The allowed
581/// values are 0 or 1... If an endpoint receives a value larger than 1, it MUST
582/// close the session with PROTOCOL_VIOLATION."
583///
584/// Both are Track Properties, so the list they arrive in is the one carried by
585/// a control message rather than the properties on an object.
586///
587/// DEFAULT_PUBLISHER_PRIORITY (0x0E) is not here. Section 12.4 says
588/// "Priorities above 255 are invalid" and stops, where the two above name a
589/// consequence in the next clause. A range stated without one is not a close.
590///
591/// The numbers belong to the Property registry and not the Message Parameter
592/// one. Type 0x22 is GROUP_ORDER as a parameter and
593/// DEFAULT_PUBLISHER_GROUP_ORDER as a property, and the two happen to permit the
594/// same pair of values while meaning different things — one subscriber's
595/// preference against a property of the track. Reading either table for the
596/// other's types would be right by accident here and wrong at the next entry.
597fn track_property_value_in_range(key: u64, value: u64) -> bool {
598 match key {
599 // DEFAULT_PUBLISHER_GROUP_ORDER (0x22)
600 0x22 => value == 1 || value == 2,
601 // DYNAMIC_GROUPS (0x30)
602 0x30 => value <= 1,
603 _ => true,
604 }
605}
606
607/// Refuse a Track Property whose value falls outside the range its type allows,
608/// wherever in the list it is carried.
609///
610/// # Inside Immutable Properties as well as beside them
611///
612/// The list is walked one level down through Immutable Properties, whose
613/// contents Section 12.7 defines as properties themselves. The draft asks for
614/// this in as many words: "When looking for the value of a property, processors
615/// MUST search both the mutable properties and the contents of Immutable
616/// Properties." A check applied only to the outer list is one a peer opts out of
617/// by moving a pair inside the block, and the block is where an Original
618/// Publisher puts what a relay must not rewrite — which is where a track's group
619/// order and dynamic-group support belong.
620///
621/// Bytes under 0xB that do not parse as a Key-Value-Pair run are left alone
622/// rather than refused. Section 12.7 says relays "MAY decode and view the
623/// Properties in the Key-Value-Pairs", which is a permission and not a
624/// requirement, so a block this codec cannot read is carried to the caller
625/// intact instead of ending the session.
626fn check_track_property_values(properties: &[KeyValuePair]) -> Result<(), CodecError> {
627 for property in properties {
628 let key = property.key.into_inner();
629 match &property.value {
630 KvpValue::Varint(value) => {
631 let value = value.into_inner();
632 if !track_property_value_in_range(key, value) {
633 return Err(CodecError::TrackPropertyValueOutOfRange { key, value });
634 }
635 }
636 KvpValue::Bytes(bytes) if key == IMMUTABLE_PROPERTIES => {
637 let mut inner = &bytes[..];
638 // A block that is not a Key-Value-Pair run is skipped rather
639 // than refused. See the note above: reading inside it is a
640 // permission, so one that cannot be read is carried.
641 //
642 // Skipped means this block and only this block. The rule the
643 // draft states here is about the block whose pairs will not
644 // parse, and says nothing about its neighbours; ending the
645 // whole walk would let a peer keep an out-of-range property
646 // from being looked at by putting an unparseable block in
647 // front of it.
648 if let Ok(nested) = decode_kvp_delta(&mut inner) {
649 check_track_property_values(&nested)?;
650 }
651 }
652 KvpValue::Bytes(_) => {}
653 }
654 }
655 Ok(())
656}
657
658/// Decode the Track Properties that fill the tail of a control message.
659///
660/// [`decode_kvp_delta`] with the Property registry's value rules applied. The
661/// two are separate because that function also reads Setup Options, which are a
662/// third namespace numbering its entries independently of this one.
663fn decode_track_properties(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
664 let properties = decode_kvp_delta(buf)?;
665 check_track_property_values(&properties)?;
666 Ok(properties)
667}
668
669/// Encode a control message's Track Properties.
670///
671/// Held to the same value ranges as the decoder. A value this codec refuses to
672/// read is one it must not write: the peer that receives it is required to close
673/// the session, so the sender's first sign of trouble would be the session
674/// going.
675fn encode_track_properties(
676 properties: &[KeyValuePair],
677 buf: &mut impl BufMut,
678) -> Result<(), CodecError> {
679 check_track_property_values(properties)?;
680 encode_kvp_delta(properties, buf)
681}
682
683/// The Setup Option types this draft defines.
684///
685/// Section 10.3.1 assigns PATH, AUTHORIZATION TOKEN, MAX_AUTH_TOKEN_CACHE_SIZE, AUTHORITY,
686/// MAX_FILTER_RANGES, MOQT_IMPLEMENTATION and MAX_REQUEST_UPDATES.
687///
688/// The list exists for one rule and one direction. Section 10.3: "Receivers
689/// MUST allow duplicates of unknown Setup Options." A receiver may therefore
690/// refuse a repeat only of a type it can name, and an option outside this list
691/// is one an extension defined and this codec has no business closing a session
692/// over. Nothing else reads it - unknown options are still decoded and carried,
693/// as "Receivers MUST ignore unrecognized Setup Options" requires.
694const KNOWN_SETUP_OPTIONS: &[u64] = &[0x01, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08];
695
696/// The one Setup Option whose definition allows more than one instance.
697///
698/// Section 10.3.1.4: "The AUTHORIZATION TOKEN Setup Option (Option Type 0x03)
699/// is functionally equivalent to the AUTHORIZATION TOKEN message parameter...
700/// The endpoint can specify one or more tokens in SETUP that the peer can use to
701/// authorize MOQT session establishment." That is the "unless the option
702/// definition explicitly allows multiple instances" carve-out, and it is the
703/// only one on this draft.
704const REPEATABLE_SETUP_OPTION: u64 = 0x03;
705
706/// Decode the Setup Options of a SETUP message.
707///
708/// Section 10.3: "Senders MUST NOT repeat the same Option Type in a message
709/// unless the option definition explicitly allows multiple instances. Receivers
710/// MUST allow duplicates of unknown Setup Options."
711///
712/// The second sentence is why this is not the mirror of
713/// [`encode_setup_options`]: a repeat of a type this draft names is refused, and
714/// a repeat of any other type is carried. Types ascend and are delta-encoded, so
715/// a repeat is always a zero delta against the option before it.
716fn decode_setup_options(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
717 let options = decode_kvp_delta(buf)?;
718 for (i, option) in options.iter().enumerate() {
719 let key = option.key.into_inner();
720 if key == REPEATABLE_SETUP_OPTION || !KNOWN_SETUP_OPTIONS.contains(&key) {
721 continue;
722 }
723 if options[..i].iter().any(|earlier| earlier.key == option.key) {
724 return Err(CodecError::DuplicateParameter(key));
725 }
726 }
727 check_authorization_tokens(&options)?;
728 Ok(options)
729}
730
731/// Encode the Setup Options of a SETUP message.
732///
733/// The sender's half of the same sentence, and it is the wider half: "Senders
734/// MUST NOT repeat the same Option Type in a message" names no exception for
735/// types the sender does not recognise, so every repeat is refused here except
736/// the one the draft allows. A caller holding an option this codec has never
737/// heard of still may not send it twice.
738///
739/// The token is in this namespace as well, and is held to its structure here for
740/// the reason [`encode_parameters`] gives.
741fn encode_setup_options(options: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
742 check_authorization_tokens(options)?;
743 for (i, option) in options.iter().enumerate() {
744 if option.key.into_inner() == REPEATABLE_SETUP_OPTION {
745 continue;
746 }
747 if options[..i].iter().any(|earlier| earlier.key == option.key) {
748 return Err(CodecError::DuplicateParameter(option.key.into_inner()));
749 }
750 }
751 encode_kvp_delta(options, buf)
752}
753
754// ============================================================
755// Message Types
756// ============================================================
757
758#[derive(Debug, Clone, Copy, PartialEq, Eq)]
759#[repr(u64)]
760pub enum MessageType {
761 RequestUpdate = 0x02,
762 Subscribe = 0x03,
763 SubscribeOk = 0x04,
764 RequestError = 0x05,
765 PublishNamespace = 0x06,
766 /// REQUEST_OK (0x07). PUBLISH_OK is now an alias of this type.
767 RequestOk = 0x07,
768 Namespace = 0x08,
769 PublishDone = 0x0B,
770 TrackStatus = 0x0D,
771 NamespaceDone = 0x0E,
772 PublishSkipped = 0x0F,
773 GoAway = 0x10,
774 Fetch = 0x16,
775 FetchOk = 0x18,
776 Publish = 0x1D,
777 /// SUBSCRIBE_NAMESPACE (renumbered to 0x50 in draft-18).
778 SubscribeNamespace = 0x50,
779 /// SUBSCRIBE_TRACKS (new message in draft-18).
780 SubscribeTracks = 0x51,
781 Setup = 0x2F00,
782}
783
784impl MessageType {
785 pub fn from_id(id: u64) -> Option<Self> {
786 match id {
787 0x02 => Some(MessageType::RequestUpdate),
788 0x03 => Some(MessageType::Subscribe),
789 0x04 => Some(MessageType::SubscribeOk),
790 0x05 => Some(MessageType::RequestError),
791 0x06 => Some(MessageType::PublishNamespace),
792 0x07 => Some(MessageType::RequestOk),
793 0x08 => Some(MessageType::Namespace),
794 0x0B => Some(MessageType::PublishDone),
795 0x0D => Some(MessageType::TrackStatus),
796 0x0E => Some(MessageType::NamespaceDone),
797 0x0F => Some(MessageType::PublishSkipped),
798 0x10 => Some(MessageType::GoAway),
799 0x16 => Some(MessageType::Fetch),
800 0x18 => Some(MessageType::FetchOk),
801 0x1D => Some(MessageType::Publish),
802 0x50 => Some(MessageType::SubscribeNamespace),
803 0x51 => Some(MessageType::SubscribeTracks),
804 0x2F00 => Some(MessageType::Setup),
805 _ => None,
806 }
807 }
808
809 pub fn id(&self) -> u64 {
810 *self as u64
811 }
812
813 /// This type's name in the shared vector corpus: the `message_type` its
814 /// draft's `codec/messages/*.json` files carry, in `snake_case`.
815 pub fn name(&self) -> &'static str {
816 match self {
817 MessageType::RequestUpdate => "request_update",
818 MessageType::Subscribe => "subscribe",
819 MessageType::SubscribeOk => "subscribe_ok",
820 MessageType::RequestError => "request_error",
821 MessageType::PublishNamespace => "publish_namespace",
822 MessageType::RequestOk => "request_ok",
823 MessageType::Namespace => "namespace",
824 MessageType::PublishDone => "publish_done",
825 MessageType::TrackStatus => "track_status",
826 MessageType::NamespaceDone => "namespace_done",
827 MessageType::PublishSkipped => "publish_skipped",
828 MessageType::GoAway => "goaway",
829 MessageType::Fetch => "fetch",
830 MessageType::FetchOk => "fetch_ok",
831 MessageType::Publish => "publish",
832 MessageType::SubscribeNamespace => "subscribe_namespace",
833 MessageType::SubscribeTracks => "subscribe_tracks",
834 MessageType::Setup => "setup",
835 }
836 }
837}
838
839// ============================================================
840// Session Lifecycle Messages
841// ============================================================
842
843/// Unified SETUP (0x2F00).
844#[derive(Debug, Clone, PartialEq, Eq)]
845pub struct Setup {
846 pub options: Vec<KeyValuePair>,
847}
848
849/// GOAWAY (0x10). Draft-19's GOAWAY has no Request ID field, so the
850/// control-stream and request-stream forms are identical on the wire.
851/// Draft-18 is the one draft that carries the field, and only when the
852/// message is sent on the control stream.
853#[derive(Debug, Clone, PartialEq, Eq)]
854pub struct GoAway {
855 pub new_session_uri: Vec<u8>,
856 pub timeout: VarInt,
857}
858
859// ============================================================
860// Consolidated Response Messages
861// ============================================================
862
863/// REQUEST_OK (0x07). Used as a generic OK response and as the alias for
864/// PUBLISH_OK / REQUEST_UPDATE_OK / TRACK_STATUS_OK / SUBSCRIBE_NAMESPACE_OK
865/// / PUBLISH_NAMESPACE_OK.
866///
867/// `track_properties` is only populated for TRACK_STATUS_OK; for every
868/// other shape it MUST be empty (length implicit from the message length).
869#[derive(Debug, Clone, PartialEq, Eq)]
870pub struct RequestOk {
871 pub parameters: Vec<KeyValuePair>,
872 pub track_properties: Vec<KeyValuePair>,
873}
874
875/// Optional Redirect structure carried in REQUEST_ERROR with code 0x34.
876#[derive(Debug, Clone, PartialEq, Eq)]
877pub struct Redirect {
878 pub connect_uri: Vec<u8>,
879 pub track_namespace: TrackNamespace,
880 pub track_name: Vec<u8>,
881}
882
883/// REQUEST_ERROR (0x05). Adds an optional Redirect structure when
884/// `error_code` is REDIRECT (0x34).
885#[derive(Debug, Clone, PartialEq, Eq)]
886pub struct RequestError {
887 pub error_code: VarInt,
888 pub retry_interval: VarInt,
889 pub reason_phrase: Vec<u8>,
890 pub redirect: Option<Redirect>,
891}
892
893/// REQUEST_ERROR error codes with dedicated meaning.
894///
895/// Note: DUPLICATE_SUBSCRIPTION (0x19) is removed in draft-19, as multiple
896/// concurrent subscriptions per Track are now allowed.
897pub mod request_error_codes {
898 /// A Mandatory Track Property the receiver does not understand.
899 pub const UNSUPPORTED_EXTENSION: u64 = 0x33;
900 /// Response carries a [`super::Redirect`] structure.
901 pub const REDIRECT: u64 = 0x34;
902 /// New in draft-19: SUBSCRIBE_TRACKS filter parameters conflict among too
903 /// many subscribers to aggregate the subscription upstream.
904 pub const CONFLICTING_FILTERS: u64 = 0x35;
905 /// New in draft-19: a Range Filter parameter is invalid or exceeds
906 /// MAX_FILTER_RANGES.
907 pub const INVALID_FILTER: u64 = 0x36;
908}
909
910// ============================================================
911// Subscribe Messages
912// ============================================================
913
914#[derive(Debug, Clone, PartialEq, Eq)]
915pub struct Subscribe {
916 pub request_id: VarInt,
917 pub track_namespace: TrackNamespace,
918 pub track_name: Vec<u8>,
919 pub parameters: Vec<KeyValuePair>,
920}
921
922/// SUBSCRIBE_OK (0x04).
923#[derive(Debug, Clone, PartialEq, Eq)]
924pub struct SubscribeOk {
925 pub track_alias: VarInt,
926 pub parameters: Vec<KeyValuePair>,
927 pub track_properties: Vec<KeyValuePair>,
928}
929
930#[derive(Debug, Clone, PartialEq, Eq)]
931pub struct RequestUpdate {
932 pub request_id: VarInt,
933 pub parameters: Vec<KeyValuePair>,
934}
935
936// ============================================================
937// Publish Messages
938// ============================================================
939
940#[derive(Debug, Clone, PartialEq, Eq)]
941pub struct Publish {
942 pub request_id: VarInt,
943 pub track_namespace: TrackNamespace,
944 pub track_name: Vec<u8>,
945 pub track_alias: VarInt,
946 pub parameters: Vec<KeyValuePair>,
947 pub track_properties: Vec<KeyValuePair>,
948}
949
950/// PUBLISH_DONE (0x0B). Status codes 0x5/0x6 are swapped vs draft-17.
951#[derive(Debug, Clone, PartialEq, Eq)]
952pub struct PublishDone {
953 pub status_code: VarInt,
954 pub stream_count: VarInt,
955 pub reason_phrase: Vec<u8>,
956}
957
958/// Numeric values for the [`PublishDone::status_code`] field.
959pub mod publish_done_codes {
960 /// Draft-18: TOO_FAR_BEHIND is 0x05 (was 0x06 in draft-17).
961 pub const TOO_FAR_BEHIND: u64 = 0x05;
962 /// Draft-18: EXPIRED is 0x06 (was 0x05 in draft-17).
963 pub const EXPIRED: u64 = 0x06;
964}
965
966// ============================================================
967// Publish Namespace Messages
968// ============================================================
969
970#[derive(Debug, Clone, PartialEq, Eq)]
971pub struct PublishNamespace {
972 pub request_id: VarInt,
973 pub track_namespace: TrackNamespace,
974 pub parameters: Vec<KeyValuePair>,
975}
976
977// ============================================================
978// Namespace Messages
979// ============================================================
980
981#[derive(Debug, Clone, PartialEq, Eq)]
982pub struct Namespace {
983 pub namespace_suffix: TrackNamespace,
984}
985
986#[derive(Debug, Clone, PartialEq, Eq)]
987pub struct NamespaceDone {
988 pub namespace_suffix: TrackNamespace,
989}
990
991// ============================================================
992// Subscribe Namespace / Tracks Messages
993// ============================================================
994
995/// SUBSCRIBE_NAMESPACE (0x50). Subscribes to NAMESPACE / NAMESPACE_DONE
996/// advertisements for namespaces matching `namespace_prefix`. The
997/// `subscribe_options` byte from draft-17 is removed; namespace subscriptions
998/// only produce NAMESPACE / NAMESPACE_DONE.
999#[derive(Debug, Clone, PartialEq, Eq)]
1000pub struct SubscribeNamespace {
1001 pub request_id: VarInt,
1002 pub namespace_prefix: TrackNamespace,
1003 pub parameters: Vec<KeyValuePair>,
1004}
1005
1006/// SUBSCRIBE_TRACKS (0x51, new in draft-18). Subscribes to PUBLISH messages
1007/// for tracks whose namespace matches `namespace_prefix`. Carries the FORWARD
1008/// parameter (Section 10.2.17), which on drafts 15 through 17 may appear on
1009/// SUBSCRIBE_NAMESPACE instead.
1010#[derive(Debug, Clone, PartialEq, Eq)]
1011pub struct SubscribeTracks {
1012 pub request_id: VarInt,
1013 pub namespace_prefix: TrackNamespace,
1014 pub parameters: Vec<KeyValuePair>,
1015}
1016
1017// ============================================================
1018// Track Status Messages
1019// ============================================================
1020
1021#[derive(Debug, Clone, PartialEq, Eq)]
1022pub struct TrackStatus {
1023 pub request_id: VarInt,
1024 pub track_namespace: TrackNamespace,
1025 pub track_name: Vec<u8>,
1026 pub parameters: Vec<KeyValuePair>,
1027}
1028
1029// ============================================================
1030// Fetch Messages
1031// ============================================================
1032
1033#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1034#[repr(u64)]
1035pub enum FetchType {
1036 Standalone = 1,
1037 RelativeJoining = 2,
1038 AbsoluteJoining = 3,
1039}
1040
1041impl FetchType {
1042 pub fn from_u64(v: u64) -> Option<Self> {
1043 match v {
1044 1 => Some(FetchType::Standalone),
1045 2 => Some(FetchType::RelativeJoining),
1046 3 => Some(FetchType::AbsoluteJoining),
1047 _ => None,
1048 }
1049 }
1050}
1051
1052#[derive(Debug, Clone, PartialEq, Eq)]
1053pub struct Fetch {
1054 pub request_id: VarInt,
1055 pub fetch_type: FetchType,
1056 pub fetch_payload: FetchPayload,
1057 pub parameters: Vec<KeyValuePair>,
1058}
1059
1060#[derive(Debug, Clone, PartialEq, Eq)]
1061pub enum FetchPayload {
1062 Standalone {
1063 track_namespace: TrackNamespace,
1064 track_name: Vec<u8>,
1065 start_group: VarInt,
1066 start_object: VarInt,
1067 end_group: VarInt,
1068 end_object: VarInt,
1069 },
1070 Joining {
1071 joining_request_id: VarInt,
1072 joining_start: VarInt,
1073 },
1074}
1075
1076/// FETCH_OK (0x18). `end_of_track` is uint8.
1077#[derive(Debug, Clone, PartialEq, Eq)]
1078pub struct FetchOk {
1079 pub end_of_track: u8,
1080 pub end_group: VarInt,
1081 pub end_object: VarInt,
1082 pub parameters: Vec<KeyValuePair>,
1083 pub track_properties: Vec<KeyValuePair>,
1084}
1085
1086// ============================================================
1087// Publish Skipped
1088// ============================================================
1089
1090/// PUBLISH_SKIPPED (0x0F). Drafts 17 and 18 name the same codepoint
1091/// PUBLISH_BLOCKED, with the same two fields.
1092#[derive(Debug, Clone, PartialEq, Eq)]
1093pub struct PublishSkipped {
1094 pub namespace_suffix: TrackNamespace,
1095 pub track_name: Vec<u8>,
1096}
1097
1098// ============================================================
1099// Unified Message Enum
1100// ============================================================
1101
1102#[derive(Debug, Clone, PartialEq, Eq)]
1103pub enum ControlMessage {
1104 Setup(Setup),
1105 GoAway(GoAway),
1106 RequestOk(RequestOk),
1107 RequestError(RequestError),
1108 Subscribe(Subscribe),
1109 SubscribeOk(SubscribeOk),
1110 RequestUpdate(RequestUpdate),
1111 Publish(Publish),
1112 PublishDone(PublishDone),
1113 PublishNamespace(PublishNamespace),
1114 Namespace(Namespace),
1115 NamespaceDone(NamespaceDone),
1116 SubscribeNamespace(SubscribeNamespace),
1117 SubscribeTracks(SubscribeTracks),
1118 TrackStatus(TrackStatus),
1119 Fetch(Fetch),
1120 FetchOk(FetchOk),
1121 PublishSkipped(PublishSkipped),
1122}
1123
1124/// Refuse a FETCH whose range ends before it starts.
1125///
1126/// Section 10.12.3: "Fetch specifies an inclusive range of Objects starting at
1127/// Start Location and ending at End Location. End Location MUST specify the
1128/// same or a larger Location than Start Location for Standalone and Absolute Joining Fetches." A Joining Fetch names
1129/// no explicit range - it is computed from the subscription it joins - so only
1130/// a standalone range is checked here.
1131///
1132/// SUBSCRIBE is not checked here, and needs no check: this draft's
1133/// AbsoluteRange filter carries an End Group Delta measured from the start
1134/// location rather than an absolute End Group, so an end before the start
1135/// has no encoding.
1136///
1137/// Applied on both sides. A range that ends before it starts selects nothing,
1138/// and the peer's only recourse is an error response or a session close, so
1139/// writing one is not a way to ask for anything.
1140fn check_ranges(message: &ControlMessage) -> Result<(), CodecError> {
1141 match message {
1142 ControlMessage::Fetch(m) => match &m.fetch_payload {
1143 FetchPayload::Standalone {
1144 start_group, start_object, end_group, end_object, ..
1145 } => check_location_range(
1146 start_group.into_inner(),
1147 start_object.into_inner(),
1148 end_group.into_inner(),
1149 end_object.into_inner(),
1150 ),
1151 FetchPayload::Joining { .. } => Ok(()),
1152 },
1153 _ => Ok(()),
1154 }
1155}
1156
1157/// Refuse a message whose discriminator disagrees with the fields beside it.
1158///
1159/// Two draft-19 messages carry a field that says which of the following fields
1160/// are on the wire: FETCH's Fetch Type, and REQUEST_ERROR's Error Code, whose
1161/// REDIRECT value (0x34) is what puts the Redirect structure on the wire. This
1162/// codec holds the alternatives in an enum and an `Option`, so a value can say
1163/// one thing in its discriminator and another in its body, and the two sides of
1164/// the codec resolve that differently — the encoder writes whatever the body
1165/// holds, and the decoder reads whatever the discriminator announces.
1166///
1167/// The result is a message that does not survive its own round trip:
1168///
1169/// - A FETCH whose type says Standalone and whose body is a joining pair
1170/// encodes to a joining request id and a joining start where a Track
1171/// Namespace and a Track Name belong, and comes back as a Standalone fetch of
1172/// a track named after two integers — or, more often, as an error, which at
1173/// least is honest. The two joining types share one body shape, so the check
1174/// is between Standalone and everything else rather than one arm per type.
1175/// - A REQUEST_ERROR with code REDIRECT and no Redirect body encodes to a
1176/// message that ends where the decoder expects a Connect URI length, so the
1177/// peer reads the redirect out of whatever follows or runs off the end. The
1178/// mirror case is quieter and no better: a Redirect body under any other
1179/// error code is written out and then skipped by a decoder that was never
1180/// told to look for it, so the sender believes it redirected a peer that
1181/// never saw a redirect.
1182///
1183/// Refusing at the encoder keeps the two readings from ever diverging on the
1184/// wire.
1185fn check_discriminators(message: &ControlMessage) -> Result<(), CodecError> {
1186 match message {
1187 ControlMessage::Fetch(m) => {
1188 let body_is_standalone = matches!(m.fetch_payload, FetchPayload::Standalone { .. });
1189 if body_is_standalone != (m.fetch_type == FetchType::Standalone) {
1190 return Err(CodecError::InvalidField);
1191 }
1192 }
1193 ControlMessage::RequestError(m) => {
1194 let code_is_redirect = m.error_code.into_inner() == request_error_codes::REDIRECT;
1195 if code_is_redirect != m.redirect.is_some() {
1196 return Err(CodecError::InvalidField);
1197 }
1198 }
1199 _ => {}
1200 }
1201 Ok(())
1202}
1203
1204/// Whether draft-19 lets Message Parameter `key` appear in `message`.
1205///
1206/// Section 10.2.1: "Each Message Parameter definition indicates the message
1207/// types in which it can appear. If it appears in some other type of message,
1208/// the receiving endpoint MUST close the connection with a PROTOCOL_VIOLATION."
1209/// One arm per entry in the Message Parameters registry (Section 15.7),
1210/// carrying the message types that entry's own definition names.
1211///
1212/// Three things about this draft the arms below fold in:
1213///
1214/// * Six of the names are one wire type. Section 10.5: "This document uses the
1215/// shorthand PUBLISH_OK, REQUEST_UPDATE_OK, TRACK_STATUS_OK,
1216/// SUBSCRIBE_NAMESPACE_OK, and PUBLISH_NAMESPACE_OK to refer to a REQUEST_OK
1217/// sent in response to the corresponding request type", and the same section
1218/// sends a REQUEST_OK in answer to SUBSCRIBE_TRACKS as well. Which one a
1219/// given REQUEST_OK is depends on the request its Request ID answers, which
1220/// is session state and not in the frame, so each of those names widens the
1221/// same arm and a REQUEST_OK is held to their union.
1222/// * The five Range Filters state their scope in Section 5.1.3 rather than in
1223/// their own subsections: the Track Property filter "MAY appear multiple
1224/// times in a SUBSCRIBE_TRACKS message or REQUEST_UPDATE for it", and "all
1225/// other filter parameters MAY appear multiple times in a FETCH, SUBSCRIBE,
1226/// SUBSCRIBE_TRACKS, PUBLISH_OK, or REQUEST_UPDATE" message. A parameter
1227/// definition is free to state its scope elsewhere, and these do.
1228/// * SUBSCRIBE_TRACKS inherits SUBSCRIBE's whole set. Section 10.19.1: "Any
1229/// Parameter that can be specified on a Subscription (ie: in SUBSCRIBE) is
1230/// valid in SUBSCRIBE_TRACKS, unless otherwise specified." Draft-18 has no
1231/// such sentence, which is why its SUBSCRIBE_TRACKS admits two types and
1232/// this one admits fourteen.
1233///
1234/// FETCH_OK has no arm in the table below, and that is the draft's doing rather
1235/// than an omission here: Section 10.13 gives it a Parameters field and no
1236/// parameter definition names it, so every type this draft defines is "some
1237/// other type of message" there.
1238///
1239/// The table decides scope only. A type this draft does not define has no scope
1240/// to be outside of and is answered by [`CodecError::UnknownMessageParameter`],
1241/// which is why the final arm carries rather than refuses.
1242fn parameter_in_scope(key: u64, message: MessageType) -> bool {
1243 use MessageType as M;
1244 // Section 10.19.1 makes SUBSCRIBE_TRACKS a superset of SUBSCRIBE, so every
1245 // arm admitting one admits the other. The arms spell both out rather than
1246 // wrapping the call, so each still reads against its own sentence.
1247 match key {
1248 // Section 10.2.4 OBJECT_DELIVERY_TIMEOUT: "It MAY appear in a
1249 // PUBLISH_OK, SUBSCRIBE, or REQUEST_UPDATE message."
1250 0x02 => {
1251 matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
1252 }
1253 // Section 10.2.2 AUTHORIZATION TOKEN: "It MAY appear in a PUBLISH,
1254 // SUBSCRIBE, REQUEST_UPDATE, SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS,
1255 // PUBLISH_NAMESPACE, TRACK_STATUS or FETCH message."
1256 0x03 => matches!(
1257 message,
1258 M::Publish
1259 | M::Subscribe
1260 | M::RequestUpdate
1261 | M::SubscribeNamespace
1262 | M::SubscribeTracks
1263 | M::PublishNamespace
1264 | M::TrackStatus
1265 | M::Fetch
1266 ),
1267 // Section 10.2.6 RENDEZVOUS TIMEOUT: it "MAY appear in a SUBSCRIBE
1268 // message".
1269 0x04 => matches!(message, M::Subscribe | M::SubscribeTracks),
1270 // Section 10.2.3 SUBGROUP_DELIVERY_TIMEOUT: "It MAY appear in a
1271 // PUBLISH_OK, SUBSCRIBE, or REQUEST_UPDATE message."
1272 0x06 => {
1273 matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
1274 }
1275 // Section 10.2.15 EXPIRES: "It MAY appear in SUBSCRIBE_OK, PUBLISH,
1276 // PUBLISH_OK, SUBSCRIBE_NAMESPACE_OK, SUBSCRIBE_TRACKS_OK,
1277 // PUBLISH_NAMESPACE_OK, or REQUEST_UPDATE_OK." Five of those seven are
1278 // a REQUEST_OK.
1279 0x08 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
1280 // Section 10.2.16 LARGEST OBJECT: "It MAY appear in SUBSCRIBE_OK,
1281 // PUBLISH, REQUEST_UPDATE_OK, or TRACK_STATUS_OK."
1282 0x09 => matches!(message, M::SubscribeOk | M::Publish | M::RequestOk),
1283 // Section 10.2.5 FILL TIMEOUT: it "MAY appear in a FETCH message".
1284 0x0A => matches!(message, M::Fetch),
1285 // Section 10.2.17 FORWARD: "It MAY appear in SUBSCRIBE, REQUEST_UPDATE
1286 // (for a subscription), PUBLISH, PUBLISH_OK and SUBSCRIBE_TRACKS."
1287 0x10 => matches!(
1288 message,
1289 M::Subscribe | M::RequestUpdate | M::Publish | M::RequestOk | M::SubscribeTracks
1290 ),
1291 // Section 10.2.7 SUBSCRIBER PRIORITY: "It MAY appear in a SUBSCRIBE,
1292 // FETCH, REQUEST_UPDATE (for a subscription or FETCH), or PUBLISH_OK
1293 // message."
1294 0x20 => matches!(
1295 message,
1296 M::Subscribe | M::Fetch | M::RequestUpdate | M::RequestOk | M::SubscribeTracks
1297 ),
1298 // Section 10.2.9 LOCATION FILTER: "It MAY appear in a SUBSCRIBE,
1299 // PUBLISH_OK or REQUEST_UPDATE (for a subscription) message."
1300 0x21 => {
1301 matches!(message, M::Subscribe | M::RequestOk | M::RequestUpdate | M::SubscribeTracks)
1302 }
1303 // Section 10.2.8 GROUP ORDER: "It MAY appear in a SUBSCRIBE,
1304 // SUBSCRIBE_TRACKS, or FETCH."
1305 0x22 => matches!(message, M::Subscribe | M::SubscribeTracks | M::Fetch),
1306 // Section 5.1.3: "All other filter parameters MAY appear multiple times
1307 // in a FETCH, SUBSCRIBE, SUBSCRIBE_TRACKS, PUBLISH_OK, or
1308 // REQUEST_UPDATE (on a subscription, from the subscriber only)
1309 // message." SUBGROUP_FILTER (Section 10.2.10), OBJECTID_FILTER
1310 // (10.2.11), PRIORITY_FILTER (10.2.12) and OBJECT_PROPERTY_FILTER
1311 // (10.2.13) are those four.
1312 0x25..=0x28 => matches!(
1313 message,
1314 M::Fetch | M::Subscribe | M::SubscribeTracks | M::RequestOk | M::RequestUpdate
1315 ),
1316 // Section 5.1.3, of TRACK_PROPERTY_FILTER (Section 10.2.14) alone: it
1317 // "MAY appear multiple times in a SUBSCRIBE_TRACKS message or
1318 // REQUEST_UPDATE for it". It selects tracks rather than objects, which
1319 // is why it is the one filter a SUBSCRIBE may not carry.
1320 0x29 => matches!(message, M::SubscribeTracks | M::RequestUpdate),
1321 // Section 10.2.18 NEW GROUP REQUEST: "It MAY appear in PUBLISH_OK,
1322 // SUBSCRIBE or REQUEST_UPDATE for a subscription."
1323 0x32 => {
1324 matches!(message, M::RequestOk | M::Subscribe | M::RequestUpdate | M::SubscribeTracks)
1325 }
1326 // Section 10.2.19 TRACK_NAMESPACE_PREFIX: "It MAY appear in
1327 // REQUEST_UPDATE for a SUBSCRIBE_NAMESPACE or SUBSCRIBE_TRACKS
1328 // request." The two named there are the request being updated, not two
1329 // more places the parameter may be written.
1330 0x34 => matches!(message, M::RequestUpdate),
1331 _ => true,
1332 }
1333}
1334
1335/// Refuse a message carrying a Message Parameter its own definition does not
1336/// place there.
1337///
1338/// Section 10.2.1 answers this with a close, which the drafts below do not.
1339/// Draft-16 Section 9.2.2, and drafts 07 through 15 under the older name
1340/// Version Specific Parameters, end the same sentence "it MUST be ignored" —
1341/// so this check belongs to drafts 17, 18 and 19 and to no draft before them.
1342///
1343/// Applied on both sides. A parameter outside its scope is one the peer must
1344/// close the session over, so writing one is a way to end a session rather than
1345/// a way to ask for anything.
1346fn check_parameter_scope(message: &ControlMessage) -> Result<(), CodecError> {
1347 let parameters = match message {
1348 ControlMessage::RequestOk(m) => &m.parameters,
1349 ControlMessage::Subscribe(m) => &m.parameters,
1350 ControlMessage::SubscribeOk(m) => &m.parameters,
1351 ControlMessage::RequestUpdate(m) => &m.parameters,
1352 ControlMessage::Publish(m) => &m.parameters,
1353 ControlMessage::PublishNamespace(m) => &m.parameters,
1354 ControlMessage::SubscribeNamespace(m) => &m.parameters,
1355 ControlMessage::SubscribeTracks(m) => &m.parameters,
1356 ControlMessage::TrackStatus(m) => &m.parameters,
1357 ControlMessage::Fetch(m) => &m.parameters,
1358 ControlMessage::FetchOk(m) => &m.parameters,
1359 // No Message Parameters field. SETUP is named here rather than left to
1360 // a wildcard because the draft says why it can never have one: Section
1361 // 10.2.1 notes that "since Setup Options use a separate namespace, it
1362 // is impossible for Message Parameters to appear in Setup messages",
1363 // and this codec keeps the two namespaces in separate fields.
1364 ControlMessage::Setup(_)
1365 | ControlMessage::GoAway(_)
1366 | ControlMessage::RequestError(_)
1367 | ControlMessage::PublishDone(_)
1368 | ControlMessage::Namespace(_)
1369 | ControlMessage::NamespaceDone(_)
1370 | ControlMessage::PublishSkipped(_) => return Ok(()),
1371 };
1372
1373 let message_type = message.message_type();
1374 for parameter in parameters {
1375 let key = parameter.key.into_inner();
1376 if !parameter_in_scope(key, message_type) {
1377 return Err(CodecError::ParameterOutOfScope { key, message_type: message_type.id() });
1378 }
1379 }
1380 Ok(())
1381}
1382
1383impl ControlMessage {
1384 pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1385 check_discriminators(self)?;
1386 check_ranges(self)?;
1387 check_parameter_scope(self)?;
1388 let mut body = Vec::with_capacity(256);
1389 self.encode_body(&mut body)?;
1390
1391 if body.len() > MAX_MESSAGE_LENGTH {
1392 return Err(CodecError::MessageTooLong(body.len()));
1393 }
1394
1395 let msg_type = self.message_type();
1396 VarInt::from_usize(msg_type.id() as usize).encode_moqt::<Wire>(buf);
1397 // Draft-19: 16-bit length (big-endian)
1398 buf.put_u16(body.len() as u16);
1399 buf.put_slice(&body);
1400 Ok(())
1401 }
1402
1403 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1404 let type_id = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1405 let msg_type =
1406 MessageType::from_id(type_id).ok_or(CodecError::UnknownMessageType(type_id))?;
1407 // Draft-19: 16-bit length (big-endian)
1408 if buf.remaining() < 2 {
1409 return Err(CodecError::UnexpectedEnd);
1410 }
1411 let body_len = buf.get_u16() as usize;
1412 if buf.remaining() < body_len {
1413 return Err(CodecError::UnexpectedEnd);
1414 }
1415 let body_bytes = buf.copy_to_bytes(body_len);
1416 let mut body = &body_bytes[..];
1417 let msg = match Self::decode_body(msg_type, &mut body) {
1418 Ok(msg) => msg,
1419 // The fields wanted more bytes than the Length allowed. This buffer
1420 // is already bounded by that Length, so running out inside it cannot
1421 // mean the message is still arriving - which is what the same error
1422 // means everywhere else, and why a reader loops on it rather than
1423 // closing. Here there is nothing left to arrive.
1424 Err(
1425 CodecError::UnexpectedEnd
1426 | CodecError::Kvp(crate::kvp::KvpError::UnexpectedEnd)
1427 | CodecError::Kvp(crate::kvp::KvpError::VarInt(
1428 crate::varint::VarIntError::UnexpectedEnd,
1429 ))
1430 | CodecError::VarInt(crate::varint::VarIntError::UnexpectedEnd),
1431 ) => {
1432 return Err(CodecError::ControlMessageLengthMismatch {
1433 declared: body_len,
1434 detail: "its fields ran past the end",
1435 });
1436 }
1437 Err(e) => return Err(e),
1438 };
1439 check_ranges(&msg)?;
1440 check_parameter_scope(&msg)?;
1441 // Draft-19 Section 10: "If the length does not match the length of the
1442 // Message Body, the receiver MUST close the session with a
1443 // PROTOCOL_VIOLATION." A body parser that stops short leaves bytes
1444 // here; without this the surplus is discarded and a truncated or
1445 // mis-framed field looks like a well-formed message.
1446 if body.has_remaining() {
1447 return Err(CodecError::ControlMessageLengthMismatch {
1448 declared: body_len,
1449 detail: "its fields left bytes unread",
1450 });
1451 }
1452 Ok(msg)
1453 }
1454
1455 fn encode_body(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1456 match self {
1457 ControlMessage::Setup(m) => {
1458 encode_setup_options(&m.options, buf)?;
1459 }
1460 ControlMessage::GoAway(m) => {
1461 if m.new_session_uri.len() > MAX_GOAWAY_URI_LENGTH {
1462 return Err(CodecError::GoAwayUriTooLong);
1463 }
1464 VarInt::from_usize(m.new_session_uri.len()).encode_moqt::<Wire>(buf);
1465 buf.put_slice(&m.new_session_uri);
1466 m.timeout.encode_moqt::<Wire>(buf);
1467 }
1468 ControlMessage::RequestOk(m) => {
1469 encode_parameters(&m.parameters, buf)?;
1470 encode_track_properties(&m.track_properties, buf)?;
1471 }
1472 ControlMessage::RequestError(m) => {
1473 if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1474 return Err(CodecError::ReasonPhraseTooLong);
1475 }
1476 m.error_code.encode_moqt::<Wire>(buf);
1477 m.retry_interval.encode_moqt::<Wire>(buf);
1478 VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
1479 buf.put_slice(&m.reason_phrase);
1480 if let Some(r) = &m.redirect {
1481 r.track_namespace.validate_moqt()?;
1482 check_full_track_name(&r.track_namespace, &r.track_name)?;
1483 VarInt::from_usize(r.connect_uri.len()).encode_moqt::<Wire>(buf);
1484 buf.put_slice(&r.connect_uri);
1485 r.track_namespace.encode_moqt::<Wire>(buf);
1486 VarInt::from_usize(r.track_name.len()).encode_moqt::<Wire>(buf);
1487 buf.put_slice(&r.track_name);
1488 }
1489 }
1490 ControlMessage::Subscribe(m) => {
1491 m.track_namespace.validate_moqt()?;
1492 check_full_track_name(&m.track_namespace, &m.track_name)?;
1493 m.request_id.encode_moqt::<Wire>(buf);
1494 m.track_namespace.encode_moqt::<Wire>(buf);
1495 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1496 buf.put_slice(&m.track_name);
1497 encode_parameters(&m.parameters, buf)?;
1498 }
1499 ControlMessage::SubscribeOk(m) => {
1500 m.track_alias.encode_moqt::<Wire>(buf);
1501 encode_parameters(&m.parameters, buf)?;
1502 encode_track_properties(&m.track_properties, buf)?;
1503 }
1504 ControlMessage::RequestUpdate(m) => {
1505 m.request_id.encode_moqt::<Wire>(buf);
1506 encode_parameters(&m.parameters, buf)?;
1507 }
1508 ControlMessage::Publish(m) => {
1509 m.track_namespace.validate_moqt()?;
1510 check_full_track_name(&m.track_namespace, &m.track_name)?;
1511 m.request_id.encode_moqt::<Wire>(buf);
1512 m.track_namespace.encode_moqt::<Wire>(buf);
1513 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1514 buf.put_slice(&m.track_name);
1515 m.track_alias.encode_moqt::<Wire>(buf);
1516 encode_parameters(&m.parameters, buf)?;
1517 encode_track_properties(&m.track_properties, buf)?;
1518 }
1519 ControlMessage::PublishDone(m) => {
1520 if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1521 return Err(CodecError::ReasonPhraseTooLong);
1522 }
1523 m.status_code.encode_moqt::<Wire>(buf);
1524 m.stream_count.encode_moqt::<Wire>(buf);
1525 VarInt::from_usize(m.reason_phrase.len()).encode_moqt::<Wire>(buf);
1526 buf.put_slice(&m.reason_phrase);
1527 }
1528 ControlMessage::PublishNamespace(m) => {
1529 m.track_namespace.validate_moqt()?;
1530 m.request_id.encode_moqt::<Wire>(buf);
1531 m.track_namespace.encode_moqt::<Wire>(buf);
1532 encode_parameters(&m.parameters, buf)?;
1533 }
1534 ControlMessage::Namespace(m) => {
1535 m.namespace_suffix.validate_moqt()?;
1536 m.namespace_suffix.encode_moqt::<Wire>(buf);
1537 }
1538 ControlMessage::NamespaceDone(m) => {
1539 m.namespace_suffix.validate_moqt()?;
1540 m.namespace_suffix.encode_moqt::<Wire>(buf);
1541 }
1542 ControlMessage::SubscribeNamespace(m) => {
1543 m.namespace_prefix.validate_moqt()?;
1544 m.request_id.encode_moqt::<Wire>(buf);
1545 m.namespace_prefix.encode_moqt::<Wire>(buf);
1546 encode_parameters(&m.parameters, buf)?;
1547 }
1548 ControlMessage::SubscribeTracks(m) => {
1549 m.namespace_prefix.validate_moqt()?;
1550 m.request_id.encode_moqt::<Wire>(buf);
1551 m.namespace_prefix.encode_moqt::<Wire>(buf);
1552 encode_parameters(&m.parameters, buf)?;
1553 }
1554 ControlMessage::TrackStatus(m) => {
1555 m.track_namespace.validate_moqt()?;
1556 check_full_track_name(&m.track_namespace, &m.track_name)?;
1557 m.request_id.encode_moqt::<Wire>(buf);
1558 m.track_namespace.encode_moqt::<Wire>(buf);
1559 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1560 buf.put_slice(&m.track_name);
1561 encode_parameters(&m.parameters, buf)?;
1562 }
1563 ControlMessage::Fetch(m) => {
1564 m.request_id.encode_moqt::<Wire>(buf);
1565 VarInt::from_usize(m.fetch_type as usize).encode_moqt::<Wire>(buf);
1566 match &m.fetch_payload {
1567 FetchPayload::Standalone {
1568 track_namespace,
1569 track_name,
1570 start_group,
1571 start_object,
1572 end_group,
1573 end_object,
1574 } => {
1575 track_namespace.validate_moqt()?;
1576 check_full_track_name(track_namespace, track_name)?;
1577 track_namespace.encode_moqt::<Wire>(buf);
1578 VarInt::from_usize(track_name.len()).encode_moqt::<Wire>(buf);
1579 buf.put_slice(track_name);
1580 start_group.encode_moqt::<Wire>(buf);
1581 start_object.encode_moqt::<Wire>(buf);
1582 end_group.encode_moqt::<Wire>(buf);
1583 end_object.encode_moqt::<Wire>(buf);
1584 }
1585 FetchPayload::Joining { joining_request_id, joining_start } => {
1586 joining_request_id.encode_moqt::<Wire>(buf);
1587 joining_start.encode_moqt::<Wire>(buf);
1588 }
1589 }
1590 encode_parameters(&m.parameters, buf)?;
1591 }
1592 ControlMessage::FetchOk(m) => {
1593 buf.put_u8(m.end_of_track);
1594 m.end_group.encode_moqt::<Wire>(buf);
1595 m.end_object.encode_moqt::<Wire>(buf);
1596 encode_parameters(&m.parameters, buf)?;
1597 encode_track_properties(&m.track_properties, buf)?;
1598 }
1599 ControlMessage::PublishSkipped(m) => {
1600 m.namespace_suffix.validate_moqt()?;
1601 check_full_track_name(&m.namespace_suffix, &m.track_name)?;
1602 m.namespace_suffix.encode_moqt::<Wire>(buf);
1603 VarInt::from_usize(m.track_name.len()).encode_moqt::<Wire>(buf);
1604 buf.put_slice(&m.track_name);
1605 }
1606 }
1607 Ok(())
1608 }
1609
1610 fn decode_body(msg_type: MessageType, buf: &mut impl Buf) -> Result<Self, CodecError> {
1611 match msg_type {
1612 MessageType::Setup => {
1613 let options = decode_setup_options(buf)?;
1614 Ok(ControlMessage::Setup(Setup { options }))
1615 }
1616 MessageType::GoAway => {
1617 let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1618 // Draft-19 Section 10.4: an endpoint that receives a New
1619 // Session URI Length above the maximum MUST close the session
1620 // with a PROTOCOL_VIOLATION. Checked here as well as on encode
1621 // so an oversized URI never reaches the application.
1622 if uri_len > MAX_GOAWAY_URI_LENGTH {
1623 return Err(CodecError::GoAwayUriTooLong);
1624 }
1625 let uri = read_bytes(buf, uri_len)?;
1626 let timeout = VarInt::decode_moqt::<Wire>(buf)?;
1627 Ok(ControlMessage::GoAway(GoAway { new_session_uri: uri, timeout }))
1628 }
1629 MessageType::RequestOk => {
1630 let parameters = decode_parameters(buf)?;
1631 let track_properties = decode_track_properties(buf)?;
1632 Ok(ControlMessage::RequestOk(RequestOk { parameters, track_properties }))
1633 }
1634 MessageType::RequestError => {
1635 let error_code = VarInt::decode_moqt::<Wire>(buf)?;
1636 let retry_interval = VarInt::decode_moqt::<Wire>(buf)?;
1637 let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1638 // Draft-19 Section 1.4.4: a received reason phrase length above
1639 // the maximum MUST close the session with a PROTOCOL_VIOLATION.
1640 if reason_len > MAX_REASON_PHRASE_LENGTH {
1641 return Err(CodecError::ReasonPhraseTooLong);
1642 }
1643 let reason_phrase = read_bytes(buf, reason_len)?;
1644 let redirect = if error_code.into_inner() == request_error_codes::REDIRECT {
1645 let uri_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1646 let connect_uri = read_bytes(buf, uri_len)?;
1647 let track_namespace = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1648 let name_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1649 let track_name = read_bytes(buf, name_len)?;
1650 check_full_track_name(&track_namespace, &track_name)?;
1651 Some(Redirect { connect_uri, track_namespace, track_name })
1652 } else {
1653 None
1654 };
1655 Ok(ControlMessage::RequestError(RequestError {
1656 error_code,
1657 retry_interval,
1658 reason_phrase,
1659 redirect,
1660 }))
1661 }
1662 MessageType::Subscribe => {
1663 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1664 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1665 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1666 let track_name = read_bytes(buf, tn_len)?;
1667 check_full_track_name(&track_namespace, &track_name)?;
1668 let parameters = decode_parameters(buf)?;
1669 Ok(ControlMessage::Subscribe(Subscribe {
1670 request_id,
1671 track_namespace,
1672 track_name,
1673 parameters,
1674 }))
1675 }
1676 MessageType::SubscribeOk => {
1677 let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
1678 let parameters = decode_parameters(buf)?;
1679 let track_properties = decode_track_properties(buf)?;
1680 Ok(ControlMessage::SubscribeOk(SubscribeOk {
1681 track_alias,
1682 parameters,
1683 track_properties,
1684 }))
1685 }
1686 MessageType::RequestUpdate => {
1687 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1688 let parameters = decode_parameters(buf)?;
1689 Ok(ControlMessage::RequestUpdate(RequestUpdate { request_id, parameters }))
1690 }
1691 MessageType::Publish => {
1692 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1693 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1694 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1695 let track_name = read_bytes(buf, tn_len)?;
1696 let track_alias = VarInt::decode_moqt::<Wire>(buf)?;
1697 check_full_track_name(&track_namespace, &track_name)?;
1698 let parameters = decode_parameters(buf)?;
1699 let track_properties = decode_track_properties(buf)?;
1700 Ok(ControlMessage::Publish(Publish {
1701 request_id,
1702 track_namespace,
1703 track_name,
1704 track_alias,
1705 parameters,
1706 track_properties,
1707 }))
1708 }
1709 MessageType::PublishDone => {
1710 let status_code = VarInt::decode_moqt::<Wire>(buf)?;
1711 let stream_count = VarInt::decode_moqt::<Wire>(buf)?;
1712 let reason_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1713 // Draft-19 Section 1.4.4, same bound as REQUEST_ERROR above.
1714 if reason_len > MAX_REASON_PHRASE_LENGTH {
1715 return Err(CodecError::ReasonPhraseTooLong);
1716 }
1717 let reason_phrase = read_bytes(buf, reason_len)?;
1718 Ok(ControlMessage::PublishDone(PublishDone {
1719 status_code,
1720 stream_count,
1721 reason_phrase,
1722 }))
1723 }
1724 MessageType::PublishNamespace => {
1725 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1726 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1727 let parameters = decode_parameters(buf)?;
1728 Ok(ControlMessage::PublishNamespace(PublishNamespace {
1729 request_id,
1730 track_namespace,
1731 parameters,
1732 }))
1733 }
1734 MessageType::Namespace => {
1735 let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1736 Ok(ControlMessage::Namespace(Namespace { namespace_suffix }))
1737 }
1738 MessageType::NamespaceDone => {
1739 let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1740 Ok(ControlMessage::NamespaceDone(NamespaceDone { namespace_suffix }))
1741 }
1742 MessageType::SubscribeNamespace => {
1743 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1744 let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1745 let parameters = decode_parameters(buf)?;
1746 Ok(ControlMessage::SubscribeNamespace(SubscribeNamespace {
1747 request_id,
1748 namespace_prefix,
1749 parameters,
1750 }))
1751 }
1752 MessageType::SubscribeTracks => {
1753 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1754 let namespace_prefix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1755 let parameters = decode_parameters(buf)?;
1756 Ok(ControlMessage::SubscribeTracks(SubscribeTracks {
1757 request_id,
1758 namespace_prefix,
1759 parameters,
1760 }))
1761 }
1762 MessageType::TrackStatus => {
1763 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1764 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1765 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1766 let track_name = read_bytes(buf, tn_len)?;
1767 check_full_track_name(&track_namespace, &track_name)?;
1768 let parameters = decode_parameters(buf)?;
1769 Ok(ControlMessage::TrackStatus(TrackStatus {
1770 request_id,
1771 track_namespace,
1772 track_name,
1773 parameters,
1774 }))
1775 }
1776 MessageType::Fetch => {
1777 let request_id = VarInt::decode_moqt::<Wire>(buf)?;
1778 let fetch_type_val = VarInt::decode_moqt::<Wire>(buf)?.into_inner();
1779 let fetch_type = FetchType::from_u64(fetch_type_val)
1780 .ok_or(CodecError::InvalidFetchType(fetch_type_val))?;
1781 let fetch_payload = match fetch_type {
1782 FetchType::Standalone => {
1783 let track_namespace = TrackNamespace::decode_moqt::<Wire>(buf)?;
1784 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1785 let track_name = read_bytes(buf, tn_len)?;
1786 let start_group = VarInt::decode_moqt::<Wire>(buf)?;
1787 let start_object = VarInt::decode_moqt::<Wire>(buf)?;
1788 let end_group = VarInt::decode_moqt::<Wire>(buf)?;
1789 let end_object = VarInt::decode_moqt::<Wire>(buf)?;
1790 check_full_track_name(&track_namespace, &track_name)?;
1791 FetchPayload::Standalone {
1792 track_namespace,
1793 track_name,
1794 start_group,
1795 start_object,
1796 end_group,
1797 end_object,
1798 }
1799 }
1800 FetchType::RelativeJoining | FetchType::AbsoluteJoining => {
1801 let joining_request_id = VarInt::decode_moqt::<Wire>(buf)?;
1802 let joining_start = VarInt::decode_moqt::<Wire>(buf)?;
1803 FetchPayload::Joining { joining_request_id, joining_start }
1804 }
1805 };
1806 let parameters = decode_parameters(buf)?;
1807 Ok(ControlMessage::Fetch(Fetch {
1808 request_id,
1809 fetch_type,
1810 fetch_payload,
1811 parameters,
1812 }))
1813 }
1814 MessageType::FetchOk => {
1815 if buf.remaining() < 1 {
1816 return Err(CodecError::UnexpectedEnd);
1817 }
1818 let end_of_track = buf.get_u8();
1819 let end_group = VarInt::decode_moqt::<Wire>(buf)?;
1820 let end_object = VarInt::decode_moqt::<Wire>(buf)?;
1821 let parameters = decode_parameters(buf)?;
1822 let track_properties = decode_track_properties(buf)?;
1823 Ok(ControlMessage::FetchOk(FetchOk {
1824 end_of_track,
1825 end_group,
1826 end_object,
1827 parameters,
1828 track_properties,
1829 }))
1830 }
1831 MessageType::PublishSkipped => {
1832 let namespace_suffix = TrackNamespace::decode_allow_empty_moqt::<Wire>(buf)?;
1833 let tn_len = VarInt::decode_moqt::<Wire>(buf)?.into_inner() as usize;
1834 let track_name = read_bytes(buf, tn_len)?;
1835 check_full_track_name(&namespace_suffix, &track_name)?;
1836 Ok(ControlMessage::PublishSkipped(PublishSkipped { namespace_suffix, track_name }))
1837 }
1838 }
1839 }
1840
1841 pub fn message_type(&self) -> MessageType {
1842 match self {
1843 ControlMessage::Setup(_) => MessageType::Setup,
1844 ControlMessage::GoAway(_) => MessageType::GoAway,
1845 ControlMessage::RequestOk(_) => MessageType::RequestOk,
1846 ControlMessage::RequestError(_) => MessageType::RequestError,
1847 ControlMessage::Subscribe(_) => MessageType::Subscribe,
1848 ControlMessage::SubscribeOk(_) => MessageType::SubscribeOk,
1849 ControlMessage::RequestUpdate(_) => MessageType::RequestUpdate,
1850 ControlMessage::Publish(_) => MessageType::Publish,
1851 ControlMessage::PublishDone(_) => MessageType::PublishDone,
1852 ControlMessage::PublishNamespace(_) => MessageType::PublishNamespace,
1853 ControlMessage::Namespace(_) => MessageType::Namespace,
1854 ControlMessage::NamespaceDone(_) => MessageType::NamespaceDone,
1855 ControlMessage::SubscribeNamespace(_) => MessageType::SubscribeNamespace,
1856 ControlMessage::SubscribeTracks(_) => MessageType::SubscribeTracks,
1857 ControlMessage::TrackStatus(_) => MessageType::TrackStatus,
1858 ControlMessage::Fetch(_) => MessageType::Fetch,
1859 ControlMessage::FetchOk(_) => MessageType::FetchOk,
1860 ControlMessage::PublishSkipped(_) => MessageType::PublishSkipped,
1861 }
1862 }
1863}
1864
1865#[cfg(test)]
1866mod tests {
1867 use super::*;
1868
1869 /// Frame `body` as a draft-19 control message of `type_id`, declaring
1870 /// `declared_len` rather than the body's real length. Used to build the
1871 /// mismatched frame the length rule is about.
1872 fn frame_with_declared_len(type_id: u64, declared_len: u16, body: &[u8]) -> Vec<u8> {
1873 let mut out = Vec::new();
1874 VarInt::from_u64_moqt(type_id).encode_moqt::<Wire>(&mut out);
1875 out.put_u16(declared_len);
1876 out.put_slice(body);
1877 out
1878 }
1879
1880 fn frame(type_id: u64, body: &[u8]) -> Vec<u8> {
1881 frame_with_declared_len(type_id, body.len() as u16, body)
1882 }
1883
1884 /// A SUBSCRIBE body: request id 1, namespace ("a"), track name "b", and
1885 /// `params` already encoded.
1886 fn subscribe_body(params: &[u8]) -> Vec<u8> {
1887 let mut body = vec![0x01, 0x01, 0x01, b'a', 0x01, b'b'];
1888 body.extend_from_slice(params);
1889 body
1890 }
1891
1892 /// Draft-19 Section 10: "If the length does not match the length of the
1893 /// Message Body, the receiver MUST close the session with a
1894 /// PROTOCOL_VIOLATION."
1895 ///
1896 /// Without the trailing-byte check in `decode` this SUBSCRIBE parses and
1897 /// the two surplus bytes vanish:
1898 ///
1899 /// ```text
1900 /// assertion `left == right` failed
1901 /// left: Ok(Subscribe(Subscribe { request_id: VarInt(1), track_namespace:
1902 /// TrackNamespace([[97]]), track_name: [98], parameters: [] }))
1903 /// right: Err(InvalidField)
1904 /// ```
1905 #[test]
1906 fn a_message_body_shorter_than_the_declared_length_is_refused() {
1907 let body = subscribe_body(&[0x00]);
1908 let mut junked = body.clone();
1909 junked.extend_from_slice(&[0xff, 0xff]);
1910 let bytes = frame_with_declared_len(0x03, (body.len() + 2) as u16, &junked);
1911
1912 let mut buf = &bytes[..];
1913 assert_eq!(
1914 ControlMessage::decode(&mut buf),
1915 Err(CodecError::ControlMessageLengthMismatch {
1916 declared: (body.len() + 2),
1917 detail: "its fields left bytes unread",
1918 })
1919 );
1920
1921 // The same body with an honest length still decodes, so the guard
1922 // rejects the mismatch and not the message.
1923 let honest = frame(0x03, &body);
1924 let mut buf = &honest[..];
1925 assert!(ControlMessage::decode(&mut buf).is_ok());
1926 }
1927
1928 /// Draft-19 Section 1.4.4: "The reason phrase length has a maximum value of
1929 /// 1024 bytes. If an endpoint receives a length exceeding the maximum, it
1930 /// MUST close the session with a PROTOCOL_VIOLATION".
1931 ///
1932 /// Without the decode-side bound the 2000-byte phrase is handed to the
1933 /// application:
1934 ///
1935 /// ```text
1936 /// assertion `left == right` failed
1937 /// left: Ok(RequestError(RequestError { error_code: VarInt(1),
1938 /// retry_interval: VarInt(0), reason_phrase: [120, 120, ...],
1939 /// redirect: None }))
1940 /// right: Err(ReasonPhraseTooLong)
1941 /// ```
1942 ///
1943 /// (The 2000 repeated bytes of the phrase are elided from that transcript.)
1944 #[test]
1945 fn an_over_long_reason_phrase_is_refused_on_decode() {
1946 for (type_id, prefix) in [(0x05u64, vec![0x01, 0x00]), (0x0B, vec![0x01, 0x00])] {
1947 let mut body = prefix;
1948 let over = MAX_REASON_PHRASE_LENGTH + 976;
1949 VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
1950 body.extend(std::iter::repeat_n(b'x', over));
1951 let bytes = frame(type_id, &body);
1952
1953 let mut buf = &bytes[..];
1954 assert_eq!(
1955 ControlMessage::decode(&mut buf),
1956 Err(CodecError::ReasonPhraseTooLong),
1957 "message type 0x{type_id:x}"
1958 );
1959 }
1960 }
1961
1962 /// Draft-19 Section 10.4: "The maximum length of the New Session URI is
1963 /// 8,192 bytes. If an endpoint receives a length exceeding the maximum, it
1964 /// MUST close the session with a PROTOCOL_VIOLATION."
1965 ///
1966 /// Without the decode-side bound the oversized URI reaches the application
1967 /// and a migrating endpoint follows it:
1968 ///
1969 /// ```text
1970 /// assertion `left == right` failed
1971 /// left: Ok(GoAway(GoAway { new_session_uri: [117, 117, ...],
1972 /// timeout: VarInt(0) }))
1973 /// right: Err(GoAwayUriTooLong)
1974 /// ```
1975 ///
1976 /// (The 9000 repeated bytes of the URI are elided from that transcript.)
1977 #[test]
1978 fn an_over_long_goaway_uri_is_refused_on_decode() {
1979 let over = MAX_GOAWAY_URI_LENGTH + 808;
1980 let mut body = Vec::new();
1981 VarInt::from_usize(over).encode_moqt::<Wire>(&mut body);
1982 body.extend(std::iter::repeat_n(b'u', over));
1983 body.push(0x00); // timeout
1984 let bytes = frame(0x10, &body);
1985
1986 let mut buf = &bytes[..];
1987 assert_eq!(ControlMessage::decode(&mut buf), Err(CodecError::GoAwayUriTooLong));
1988 }
1989
1990 /// Draft-19 Section 10.2.8 (GROUP_ORDER): "The allowed values are Ascending
1991 /// (0x1) or Descending (0x2). If an endpoint receives a value outside this
1992 /// range, it MUST close the session with PROTOCOL_VIOLATION." Section
1993 /// 10.2.17 says the same of FORWARD with the values 0 and 1.
1994 ///
1995 /// Without `uint8_value_in_range` the out-of-range byte is handed up as an
1996 /// ordinary parameter:
1997 ///
1998 /// ```text
1999 /// assertion `left == right` failed
2000 /// left: Ok(Subscribe(Subscribe { request_id: VarInt(1), track_namespace:
2001 /// TrackNamespace([[97]]), track_name: [98], parameters:
2002 /// [KeyValuePair { key: VarInt(34), value: Varint(VarInt(7)) }] }))
2003 /// right: Err(InvalidField)
2004 /// ```
2005 #[test]
2006 fn a_uint8_parameter_outside_its_range_is_refused() {
2007 // key, rejected value, accepted value
2008 let cases = [(0x22u8, 7u8, 2u8), (0x10, 9, 1)];
2009 for (key, bad, good) in cases {
2010 let bytes = frame(0x03, &subscribe_body(&[0x01, key, bad]));
2011 let mut buf = &bytes[..];
2012 assert_eq!(
2013 ControlMessage::decode(&mut buf),
2014 Err(CodecError::ParameterValueOutOfRange { key: key as u64, value: bad as u64 }),
2015 "parameter 0x{key:x} value {bad}"
2016 );
2017
2018 let bytes = frame(0x03, &subscribe_body(&[0x01, key, good]));
2019 let mut buf = &bytes[..];
2020 assert!(
2021 ControlMessage::decode(&mut buf).is_ok(),
2022 "parameter 0x{key:x} value {good} should still decode"
2023 );
2024 }
2025 }
2026
2027 /// SUBSCRIBER_PRIORITY (0x20) is a uint8 with no restricted range, so it
2028 /// must keep accepting the whole 0-255 span. This is the negative half of
2029 /// the range check: a table that over-reached would fail here.
2030 #[test]
2031 fn subscriber_priority_still_accepts_the_whole_byte_range() {
2032 for value in [0u8, 1, 2, 128, 255] {
2033 let bytes = frame(0x03, &subscribe_body(&[0x01, 0x20, value]));
2034 let mut buf = &bytes[..];
2035 assert!(ControlMessage::decode(&mut buf).is_ok(), "priority {value}");
2036 }
2037 }
2038
2039 fn param(key: u64, value: &[u8]) -> KeyValuePair {
2040 KeyValuePair { key: VarInt::from_u64_moqt(key), value: KvpValue::Bytes(value.to_vec()) }
2041 }
2042
2043 /// Draft-19 Section 10.2.16: "The LARGEST_OBJECT parameter (Parameter Type
2044 /// 0x9) is a Location." Section 10.2 defines Location as "Two consecutive
2045 /// varints (Group, Object)" — the value carries no length of its own.
2046 ///
2047 /// The frame below is built from the draft rather than from this encoder:
2048 /// REQUEST_OK, four body bytes, one parameter, type delta `0x09`, then the
2049 /// two varints `0x0a` and `0x03` for Location (10, 3). A length-prefixed
2050 /// spelling would need a fifth byte.
2051 ///
2052 /// With `0x09` back in the Length-prefixed arm, the decoder reads the
2053 /// group varint `0x0a` as a value length of 10 and runs off the end of a
2054 /// four-byte body. Both this test and
2055 /// [`a_location_does_not_eat_the_block_that_follows_it`] fail with:
2056 ///
2057 /// ```text
2058 /// spec-correct frame must decode: UnexpectedEnd
2059 /// ```
2060 #[test]
2061 fn largest_object_is_two_bare_varints() {
2062 let body = [0x01, 0x09, 0x0a, 0x03];
2063 let bytes = frame(0x07, &body);
2064
2065 let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2066 let ControlMessage::RequestOk(ok) = &msg else {
2067 panic!("expected REQUEST_OK, got {msg:?}")
2068 };
2069 assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
2070 assert!(ok.track_properties.is_empty(), "the four body bytes are all parameter");
2071
2072 let mut out = Vec::new();
2073 msg.encode(&mut out).expect("re-encode");
2074 assert_eq!(out, bytes, "the value must go back out as the two bare varints it came in as");
2075 }
2076
2077 /// A Location value the encoder was handed but the decoder could not read
2078 /// back is refused on the way out, not written.
2079 ///
2080 /// LARGEST_OBJECT carries no length of its own — that is the whole point
2081 /// of the encoding — so `encode_parameters` writes its bytes verbatim. A
2082 /// value built in memory rather than decoded is under no obligation to be
2083 /// two varints, and without this check the codec answers `Ok(())` and puts
2084 /// a frame on the wire that `ControlMessage::decode` then refuses. One
2085 /// varint short and one varint long are the two ways to get it wrong.
2086 ///
2087 /// # What it catches
2088 ///
2089 /// Dropping the `is_location_value` guard from this draft's encode arm,
2090 /// run:
2091 ///
2092 /// ```text
2093 /// panicked at crates\moqtap-codec\src\draft19\message.rs:
2094 /// LARGEST_OBJECT of one varint must not encode: the decoder cannot read it back
2095 ///
2096 /// test result: FAILED. 1 passed; 1 failed; 0 ignored; 0 measured; 108 filtered out
2097 /// ```
2098 ///
2099 /// The sibling draft kept its guard and kept passing, which is what shows
2100 /// the check is per-draft and not inherited from somewhere shared.
2101 #[test]
2102 fn a_location_value_that_is_not_two_varints_is_refused_on_encode() {
2103 for (label, value) in
2104 [("one varint", vec![0x0a]), ("three varints", vec![0x0a, 0x03, 0x05])]
2105 {
2106 let msg = ControlMessage::RequestOk(RequestOk {
2107 parameters: vec![param(0x09, &value)],
2108 track_properties: Vec::new(),
2109 });
2110 let mut out = Vec::new();
2111 assert!(
2112 msg.encode(&mut out).is_err(),
2113 "LARGEST_OBJECT of {label} must not encode: the decoder cannot read it back"
2114 );
2115 }
2116
2117 // The well-formed value still goes out, so the check refuses the
2118 // malformed case and not the encoding itself.
2119 let msg = ControlMessage::RequestOk(RequestOk {
2120 parameters: vec![param(0x09, &[0x0a, 0x03])],
2121 track_properties: Vec::new(),
2122 });
2123 let mut out = Vec::new();
2124 msg.encode(&mut out).expect("a Location of exactly two varints must still encode");
2125 ControlMessage::decode(&mut &out[..]).expect("and must decode back");
2126 }
2127
2128 /// The same Location read through a message that carries other fields
2129 /// after it, so a stray length byte cannot hide in a trailing block.
2130 ///
2131 /// SUBSCRIBE_OK is track alias `0x05`, then the parameters, then the track
2132 /// properties. With LARGEST_OBJECT (10, 3) and one property
2133 /// (OBJECT_DELIVERY_TIMEOUT, type `0x02`, 5000ms as the two-byte varint
2134 /// `0x93 0x88`), the body is `05 01 09 0a 03 02 93 88`.
2135 #[test]
2136 fn a_location_does_not_eat_the_block_that_follows_it() {
2137 let body = [0x05, 0x01, 0x09, 0x0a, 0x03, 0x02, 0x93, 0x88];
2138 let bytes = frame(0x04, &body);
2139
2140 let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2141 let ControlMessage::SubscribeOk(ok) = &msg else {
2142 panic!("expected SUBSCRIBE_OK, got {msg:?}")
2143 };
2144 assert_eq!(ok.parameters, vec![param(0x09, &[0x0a, 0x03])]);
2145 assert_eq!(
2146 ok.track_properties,
2147 vec![KeyValuePair {
2148 key: VarInt::from_u64_moqt(0x02),
2149 value: KvpValue::Varint(VarInt::from_u64_moqt(5000)),
2150 }]
2151 );
2152
2153 let mut out = Vec::new();
2154 msg.encode(&mut out).expect("re-encode");
2155 assert_eq!(out, bytes);
2156 }
2157
2158 /// Draft-19 Section 10.2.19: the TRACK_NAMESPACE_PREFIX parameter
2159 /// (Parameter Type 0x34) "uses the Track Namespace encoding described in
2160 /// Section 2.4.1" — a varint field count followed by that many
2161 /// length-prefixed fields, and nothing in front of it. That encoding is not
2162 /// one of the four Section 10.2 lists, so it cannot be assumed to be
2163 /// Length-prefixed by default.
2164 ///
2165 /// The frame below is built from Section 2.4.1: REQUEST_UPDATE for request
2166 /// `7`, one parameter, type delta `0x34`, then the namespace ("live",
2167 /// "sports") as `02 04 "live" 06 "sports"`. Sixteen body bytes; a
2168 /// length-prefixed spelling would need a seventeenth for the outer length.
2169 ///
2170 /// With `0x34` back in the Length-prefixed arm the field count `0x02` is
2171 /// read as an outer length of two bytes, leaving eleven bytes of namespace
2172 /// unread. Draft-19's Section 10 body-length check turns that into a
2173 /// refusal rather than a truncated value:
2174 ///
2175 /// ```text
2176 /// spec-correct frame must decode: InvalidField
2177 /// ```
2178 ///
2179 /// That check is not a safety net here. Where the surplus lands inside the
2180 /// declared body — as in
2181 /// [`an_empty_track_namespace_prefix_is_one_zero_byte`], whose namespace is
2182 /// one byte long — the misread is silent, and that test fails instead with:
2183 ///
2184 /// ```text
2185 /// assertion `left == right` failed
2186 /// left: [KeyValuePair { key: VarInt(52), value: Bytes([]) }]
2187 /// right: [KeyValuePair { key: VarInt(52), value: Bytes([0]) }]
2188 /// ```
2189 #[test]
2190 fn track_namespace_prefix_is_a_bare_track_namespace() {
2191 let namespace: Vec<u8> = [&[0x02, 0x04][..], b"live", &[0x06][..], b"sports"].concat();
2192 assert_eq!(namespace.len(), 13);
2193
2194 let body: Vec<u8> = [&[0x07, 0x01, 0x34][..], &namespace].concat();
2195 assert_eq!(body.len(), 16);
2196 let bytes = frame(0x02, &body);
2197
2198 let msg = ControlMessage::decode(&mut &bytes[..]).expect("spec-correct frame must decode");
2199 let ControlMessage::RequestUpdate(update) = &msg else {
2200 panic!("expected REQUEST_UPDATE, got {msg:?}")
2201 };
2202 assert_eq!(update.request_id.into_inner(), 7);
2203 assert_eq!(update.parameters, vec![param(0x34, &namespace)]);
2204
2205 let mut out = Vec::new();
2206 msg.encode(&mut out).expect("re-encode");
2207 assert_eq!(out, bytes, "no outer length may appear in front of the Track Namespace");
2208 }
2209
2210 /// An empty prefix is a legal Track Namespace: Section 2.4.1 puts one at
2211 /// "between 0 and 32 Track Namespace Fields". On the wire that is the
2212 /// single byte `0x00`, and it must not be confused with a length-prefixed
2213 /// value of zero bytes.
2214 #[test]
2215 fn an_empty_track_namespace_prefix_is_one_zero_byte() {
2216 let body = [0x07, 0x01, 0x34, 0x00];
2217 let bytes = frame(0x02, &body);
2218
2219 let msg = ControlMessage::decode(&mut &bytes[..]).expect("empty prefix must decode");
2220 let ControlMessage::RequestUpdate(update) = &msg else {
2221 panic!("expected REQUEST_UPDATE, got {msg:?}")
2222 };
2223 assert_eq!(update.parameters, vec![param(0x34, &[0x00])]);
2224
2225 let mut out = Vec::new();
2226 msg.encode(&mut out).expect("re-encode");
2227 assert_eq!(out, bytes);
2228 }
2229}