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