moqtap_codec/draft16/
fields.rs

1use crate::draft16::message::ControlMessage;
2use crate::fields::{FieldMap as Map, FieldValue as Value};
3use crate::kvp::{KeyValuePair, KvpValue};
4use crate::types::*;
5use crate::varint::VarInt;
6
7fn vi(v: u64) -> Value {
8    Value::Uint(v)
9}
10
11fn ns_to_json(ns: &TrackNamespace) -> Value {
12    Value::Array(
13        ns.0.iter().map(|e| Value::Text(String::from_utf8_lossy(e).into_owned())).collect(),
14    )
15}
16
17fn d16_setup_param_name(key: u64) -> Option<&'static str> {
18    match key {
19        0x01 => Some("path"),
20        0x02 => Some("max_request_id"),
21        0x03 => Some("authorization_token"),
22        0x04 => Some("max_auth_token_cache_size"),
23        0x05 => Some("authority"),
24        0x07 => Some("moqt_implementation"),
25        _ => None,
26    }
27}
28
29/// The nine Message Parameter Types draft-16 Section 13.2 Table 8 assigns.
30///
31/// Exactly the nine, and the same nine `message.rs`'s `KNOWN_MESSAGE_PARAMETERS`
32/// holds. Three more are easy to carry here by mistake — `0x04` as
33/// `max_cache_duration`, `0x0e` as `publisher_priority` and `0x30` as
34/// `dynamic_groups` — because draft-15's Table 10 really does have all three in
35/// this namespace.
36///
37/// Draft-16 did not delete those three; it moved them. 0x04 and 0x0e are
38/// MAX_CACHE_DURATION and DEFAULT_PUBLISHER_PRIORITY in Table 9's Extension
39/// Header registry and 0x30 is DYNAMIC_GROUPS there, which is
40/// [`d16_track_ext_name`]'s table and not this one. PUBLISHER_PRIORITY as a
41/// Message Parameter is gone outright.
42///
43/// So the three names would be not merely unused but unreachable *and*
44/// wrong: draft-16 answers an unknown Message Parameter with a session close —
45/// Section 9.2, "An endpoint that receives an unknown Message Parameter MUST
46/// close the session with PROTOCOL_VIOLATION" — and `decode_parameters` applies
47/// it, so no decoded draft-16 message can carry one of these three keys in a
48/// Message Parameter list. A name this table gave them could only ever be read
49/// about a parameter the same file had already refused.
50fn d16_msg_param_name(key: u64) -> Option<&'static str> {
51    match key {
52        0x02 => Some("delivery_timeout"),
53        0x03 => Some("authorization_token"),
54        0x08 => Some("expires"),
55        0x09 => Some("largest_object"),
56        0x10 => Some("forward"),
57        0x20 => Some("subscriber_priority"),
58        0x21 => Some("subscription_filter"),
59        0x22 => Some("group_order"),
60        0x32 => Some("new_group_request"),
61        _ => None,
62    }
63}
64
65fn auth_token_to_json_d16(bytes: &[u8]) -> Value {
66    let mut buf = bytes;
67    let alias_type = match VarInt::decode(&mut buf) {
68        Ok(v) => v,
69        Err(_) => return Value::Bytes(bytes.to_vec()),
70    };
71    let at = alias_type.into_inner();
72    let mut o = Map::new();
73    o.insert("alias_type".into(), vi(at));
74    match at {
75        0 | 2 => {
76            if let Ok(ta) = VarInt::decode(&mut buf) {
77                o.insert("token_alias".into(), vi(ta.into_inner()));
78            }
79        }
80        1 => {
81            if let Ok(ta) = VarInt::decode(&mut buf) {
82                o.insert("token_alias".into(), vi(ta.into_inner()));
83            }
84            if let Ok(tt) = VarInt::decode(&mut buf) {
85                o.insert("token_type".into(), vi(tt.into_inner()));
86            }
87            o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
88        }
89        _ => {
90            if let Ok(tt) = VarInt::decode(&mut buf) {
91                o.insert("token_type".into(), vi(tt.into_inner()));
92            }
93            o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
94        }
95    }
96    Value::Map(o)
97}
98
99/// Render a SUBSCRIPTION FILTER (0x21) parameter value: a Filter Type and the
100/// Start Location and End Group that type promises.
101///
102/// # Nothing has checked that the value holds the fields its Filter Type names
103///
104/// 0x21 is an odd Type, so `KeyValuePair::decode` keeps whatever
105/// length-prefixed bytes arrived and the value reaches here unexamined. None of
106/// `decode_parameters`' own checks looks at the *contents* of a filter value,
107/// and `crate::dispatch::AnyControlMessage::fields` renders every message that
108/// decoded — so a peer's bytes reach this function directly. That is the chain
109/// `tests/hostile_parameter_values.rs` sets out in full.
110///
111/// The truncation that follows an AbsoluteStart or AbsoluteRange Filter Type is
112/// the nastier shape, because the value looks well formed right up to the point
113/// where it is not: the Filter Type decodes cleanly and the Start Location it
114/// promises is simply not there.
115///
116/// # What a value it cannot read renders as
117///
118/// The raw bytes, as `fields::params` and `decode_largest_object` do. Field
119/// extraction runs on a message that has already decoded, so it has no refusal
120/// to give: what a peer sent is what there is to show.
121fn decode_subscription_filter(bytes: &[u8]) -> Value {
122    let mut buf = bytes;
123    let Ok(filter_type) = VarInt::decode(&mut buf) else {
124        return Value::Bytes(bytes.to_vec());
125    };
126    let filter_type = filter_type.into_inner();
127    let mut obj = Map::new();
128    obj.insert("filter_type".into(), vi(filter_type));
129    match filter_type {
130        3 => {
131            let Ok(start_group) = VarInt::decode(&mut buf) else {
132                return Value::Bytes(bytes.to_vec());
133            };
134            let start_group = start_group.into_inner();
135            let Ok(start_object) = VarInt::decode(&mut buf) else {
136                return Value::Bytes(bytes.to_vec());
137            };
138            let start_object = start_object.into_inner();
139            obj.insert("start_group".into(), vi(start_group));
140            obj.insert("start_object".into(), vi(start_object));
141        }
142        4 => {
143            let Ok(start_group) = VarInt::decode(&mut buf) else {
144                return Value::Bytes(bytes.to_vec());
145            };
146            let start_group = start_group.into_inner();
147            let Ok(start_object) = VarInt::decode(&mut buf) else {
148                return Value::Bytes(bytes.to_vec());
149            };
150            let start_object = start_object.into_inner();
151            let Ok(end_group) = VarInt::decode(&mut buf) else {
152                return Value::Bytes(bytes.to_vec());
153            };
154            let end_group = end_group.into_inner();
155            obj.insert("start_group".into(), vi(start_group));
156            obj.insert("start_object".into(), vi(start_object));
157            obj.insert("end_group".into(), vi(end_group));
158        }
159        _ => {}
160    }
161    Value::Map(obj)
162}
163
164/// Render a draft-16 LARGEST_OBJECT (0x09) parameter value: a Group and an
165/// Object, as two varints.
166///
167/// # Nothing has checked that the value is two varints
168///
169/// Drafts 17 and later give 0x09 a `Location` encoding: their decoders read the
170/// two varints and re-serialise them into the stored value, so what reaches
171/// their extractor is two varints by construction. Draft-16 has no such table.
172/// 0x09 is an odd Type, so `KeyValuePair::decode` keeps whatever
173/// length-prefixed bytes arrived, and `decode_parameters_in` checks duplicates,
174/// authorization tokens, varint value ranges and subscription filters — none of
175/// which looks at 0x09. `KNOWN_MESSAGE_PARAMETERS` admits it and
176/// `check_parameter_scope` permits it on SUBSCRIBE_OK, so a short SUBSCRIBE_OK
177/// carrying `0x09` with an empty value reaches here.
178///
179/// An empty value fails the first read and a single `0x00` fails the *second*,
180/// which is the nastier of the two: the first varint decodes cleanly and the
181/// value looks well formed right up to the point where it is not.
182///
183/// # What a value it cannot read renders as
184///
185/// The raw bytes, as `fields::params` and this file's own
186/// `auth_token_to_json_d16` do. Field extraction runs on a message that has
187/// already decoded, so it has no refusal to give: what a peer sent is what
188/// there is to show.
189fn decode_largest_object(bytes: &[u8]) -> Value {
190    let mut buf = bytes;
191    let Ok(group) = VarInt::decode(&mut buf) else {
192        return Value::Bytes(bytes.to_vec());
193    };
194    let Ok(object) = VarInt::decode(&mut buf) else {
195        return Value::Bytes(bytes.to_vec());
196    };
197    let mut obj = Map::new();
198    obj.insert("group".into(), vi(group.into_inner()));
199    obj.insert("object".into(), vi(object.into_inner()));
200    Value::Map(obj)
201}
202
203fn kvp_to_json_d16_inner(
204    params: &[KeyValuePair],
205    name_fn: fn(u64) -> Option<&'static str>,
206) -> Value {
207    crate::fields::kvp_entries(params, |key, value| {
208        let Some(name) = name_fn(key) else {
209            return (None, None);
210        };
211        let rendered = match (value, key) {
212            (KvpValue::Bytes(b), 0x21) => decode_subscription_filter(b),
213            (KvpValue::Bytes(b), 0x09) => decode_largest_object(b),
214            (KvpValue::Bytes(b), _) if name == "authorization_token" => auth_token_to_json_d16(b),
215            (KvpValue::Varint(v), _) => vi(v.into_inner()),
216            (KvpValue::Bytes(b), _) => Value::Text(String::from_utf8_lossy(b).into_owned()),
217        };
218        (Some(name), Some(rendered))
219    })
220}
221
222fn kvp_to_json_d16(params: &[KeyValuePair]) -> Value {
223    kvp_to_json_d16_inner(params, d16_msg_param_name)
224}
225
226pub(crate) fn kvp_to_json_d16_setup(params: &[KeyValuePair]) -> Value {
227    kvp_to_json_d16_inner(params, d16_setup_param_name)
228}
229
230/// Every Track-scoped Extension Header Type draft-16 Section 13.3 Table 9
231/// assigns.
232///
233/// A separate table from [`d16_msg_param_name`] because the two registries
234/// reuse numbers for different things: 0x22 is GROUP_ORDER as a Message
235/// Parameter and DEFAULT_PUBLISHER_GROUP_ORDER as an Extension Header, and 0x02
236/// is DELIVERY_TIMEOUT in both while meaning one endpoint's request in one and a
237/// property of the track in the other.
238///
239/// 0x0B, 0x22 and 0x30 are the whole of what an Original Publisher puts in an
240/// Immutable Extensions block, and both of the Track Extensions draft-16 gives
241/// a value range to. They are the ones worth naming: the two with ranges are
242/// exactly the two `message.rs` closes the session over, so an unnamed type
243/// number here is the one a trace most needs named.
244///
245/// 0x3C and 0x3E are deliberately absent. Table 9 scopes both to Object, and the
246/// only caller of this table renders a `track_extensions` field.
247fn d16_track_ext_name(key: u64) -> Option<&'static str> {
248    match key {
249        0x02 => Some("delivery_timeout"),
250        0x04 => Some("max_cache_duration"),
251        0x0b => Some("immutable_extensions"),
252        0x0e => Some("default_publisher_priority"),
253        0x22 => Some("default_publisher_group_order"),
254        0x30 => Some("dynamic_groups"),
255        _ => None,
256    }
257}
258
259fn kvp_to_json_d16_track_ext(params: &[KeyValuePair]) -> Value {
260    kvp_to_json_d16_inner(params, d16_track_ext_name)
261}
262
263/// This draft's field names for a decoded control message.
264///
265/// Keys are the names this draft gives its fields, in the order it defines
266/// them. An optional field the message did not carry is absent rather than
267/// zero.
268pub fn message_fields(msg: &ControlMessage) -> Map {
269    let obj = match msg {
270        ControlMessage::ClientSetup(m) => {
271            let mut o = Map::new();
272            o.insert("parameters".into(), kvp_to_json_d16_setup(&m.parameters));
273            o
274        }
275        ControlMessage::ServerSetup(m) => {
276            let mut o = Map::new();
277            o.insert("parameters".into(), kvp_to_json_d16_setup(&m.parameters));
278            o
279        }
280        ControlMessage::GoAway(m) => {
281            let mut o = Map::new();
282            o.insert(
283                "new_session_uri".into(),
284                Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
285            );
286            o
287        }
288        ControlMessage::MaxRequestId(m) => {
289            let mut o = Map::new();
290            o.insert("max_request_id".into(), vi(m.request_id.into_inner()));
291            o
292        }
293        ControlMessage::RequestsBlocked(m) => {
294            let mut o = Map::new();
295            o.insert("maximum_request_id".into(), vi(m.maximum_request_id.into_inner()));
296            o
297        }
298        ControlMessage::RequestOk(m) => {
299            let mut o = Map::new();
300            o.insert("request_id".into(), vi(m.request_id.into_inner()));
301            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
302            o
303        }
304        ControlMessage::RequestError(m) => {
305            let mut o = Map::new();
306            o.insert("request_id".into(), vi(m.request_id.into_inner()));
307            o.insert("error_code".into(), vi(m.error_code.into_inner()));
308            o.insert("retry_interval".into(), vi(m.retry_interval.into_inner()));
309            o.insert(
310                "reason_phrase".into(),
311                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
312            );
313            o
314        }
315        ControlMessage::Subscribe(m) => {
316            let mut o = Map::new();
317            o.insert("request_id".into(), vi(m.request_id.into_inner()));
318            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
319            o.insert(
320                "track_name".into(),
321                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
322            );
323            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
324            o
325        }
326        ControlMessage::SubscribeOk(m) => {
327            let mut o = Map::new();
328            o.insert("request_id".into(), vi(m.request_id.into_inner()));
329            o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
330            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
331            if !m.track_extensions.is_empty() {
332                o.insert("track_extensions".into(), kvp_to_json_d16_track_ext(&m.track_extensions));
333            }
334            o
335        }
336        ControlMessage::RequestUpdate(m) => {
337            let mut o = Map::new();
338            o.insert("request_id".into(), vi(m.request_id.into_inner()));
339            o.insert("existing_request_id".into(), vi(m.existing_request_id.into_inner()));
340            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
341            o
342        }
343        ControlMessage::Unsubscribe(m) => {
344            let mut o = Map::new();
345            o.insert("request_id".into(), vi(m.request_id.into_inner()));
346            o
347        }
348        ControlMessage::Publish(m) => {
349            let mut o = Map::new();
350            o.insert("request_id".into(), vi(m.request_id.into_inner()));
351            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
352            o.insert(
353                "track_name".into(),
354                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
355            );
356            o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
357            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
358            if !m.track_extensions.is_empty() {
359                o.insert("track_extensions".into(), kvp_to_json_d16_track_ext(&m.track_extensions));
360            }
361            o
362        }
363        ControlMessage::PublishOk(m) => {
364            let mut o = Map::new();
365            o.insert("request_id".into(), vi(m.request_id.into_inner()));
366            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
367            o
368        }
369        ControlMessage::PublishDone(m) => {
370            let mut o = Map::new();
371            o.insert("request_id".into(), vi(m.request_id.into_inner()));
372            o.insert("status_code".into(), vi(m.status_code.into_inner()));
373            o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
374            o.insert(
375                "reason_phrase".into(),
376                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
377            );
378            o
379        }
380        ControlMessage::PublishNamespace(m) => {
381            let mut o = Map::new();
382            o.insert("request_id".into(), vi(m.request_id.into_inner()));
383            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
384            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
385            o
386        }
387        ControlMessage::PublishNamespaceDone(m) => {
388            let mut o = Map::new();
389            o.insert("request_id".into(), vi(m.request_id.into_inner()));
390            o
391        }
392        ControlMessage::PublishNamespaceCancel(m) => {
393            let mut o = Map::new();
394            o.insert("request_id".into(), vi(m.request_id.into_inner()));
395            o.insert("error_code".into(), vi(m.error_code.into_inner()));
396            o.insert(
397                "reason_phrase".into(),
398                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
399            );
400            o
401        }
402        ControlMessage::Namespace(m) => {
403            let mut o = Map::new();
404            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
405            o
406        }
407        ControlMessage::NamespaceDone(m) => {
408            let mut o = Map::new();
409            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
410            o
411        }
412        ControlMessage::SubscribeNamespace(m) => {
413            let mut o = Map::new();
414            o.insert("request_id".into(), vi(m.request_id.into_inner()));
415            o.insert("namespace_prefix".into(), ns_to_json(&m.namespace_prefix));
416            o.insert("subscribe_options".into(), vi(m.subscribe_options.into_inner()));
417            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
418            o
419        }
420        ControlMessage::TrackStatus(m) => {
421            let mut o = Map::new();
422            o.insert("request_id".into(), vi(m.request_id.into_inner()));
423            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
424            o.insert(
425                "track_name".into(),
426                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
427            );
428            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
429            o
430        }
431        ControlMessage::Fetch(m) => {
432            let mut o = Map::new();
433            o.insert("request_id".into(), vi(m.request_id.into_inner()));
434            o.insert("fetch_type".into(), vi(m.fetch_type as u64));
435            match &m.fetch_payload {
436                crate::draft16::message::FetchPayload::Standalone {
437                    track_namespace,
438                    track_name,
439                    start_group,
440                    start_object,
441                    end_group,
442                    end_object,
443                } => {
444                    o.insert("track_namespace".into(), ns_to_json(track_namespace));
445                    o.insert(
446                        "track_name".into(),
447                        Value::Text(String::from_utf8_lossy(track_name).into_owned()),
448                    );
449                    o.insert("start_group".into(), vi(start_group.into_inner()));
450                    o.insert("start_object".into(), vi(start_object.into_inner()));
451                    o.insert("end_group".into(), vi(end_group.into_inner()));
452                    o.insert("end_object".into(), vi(end_object.into_inner()));
453                }
454                crate::draft16::message::FetchPayload::Joining {
455                    joining_request_id,
456                    joining_start,
457                } => {
458                    o.insert("joining_request_id".into(), vi(joining_request_id.into_inner()));
459                    o.insert("joining_start".into(), vi(joining_start.into_inner()));
460                }
461            }
462            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
463            o
464        }
465        ControlMessage::FetchOk(m) => {
466            let mut o = Map::new();
467            o.insert("request_id".into(), vi(m.request_id.into_inner()));
468            o.insert("end_of_track".into(), vi(m.end_of_track as u64));
469            o.insert("end_group".into(), vi(m.end_group.into_inner()));
470            o.insert("end_object".into(), vi(m.end_object.into_inner()));
471            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
472            if !m.track_extensions.is_empty() {
473                o.insert("track_extensions".into(), kvp_to_json_d16_track_ext(&m.track_extensions));
474            }
475            o
476        }
477        ControlMessage::FetchCancel(m) => {
478            let mut o = Map::new();
479            o.insert("request_id".into(), vi(m.request_id.into_inner()));
480            o
481        }
482    };
483    obj
484}
moqtap_codec/draft16/fields.rs

moqtap_codec/draft16/
fields.rs
