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