Skip to main content

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 &parameter.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 &parameter.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(&params)?;
325    check_subscription_filters(&params)?;
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}