moqtap_codec/draft17/
fields.rs

1use crate::draft17::message::ControlMessage;
2use crate::fields::{FieldMap as Map, FieldValue as Value};
3use crate::kvp::{KeyValuePair, KvpValue};
4use crate::types::*;
5use crate::varint::{Moqt17 as Wire, 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
17// Draft-17 known parameter types and their encodings
18fn d17_param_name(key: u64) -> Option<&'static str> {
19    match key {
20        0x02 => Some("delivery_timeout"),
21        0x03 => Some("authorization_token"),
22        0x04 => Some("rendezvous_timeout"),
23        0x08 => Some("expires"),
24        0x09 => Some("largest_object"),
25        0x10 => Some("forward"),
26        0x20 => Some("subscriber_priority"),
27        0x21 => Some("subscription_filter"),
28        0x22 => Some("group_order"),
29        0x32 => Some("new_group_request"),
30        _ => None,
31    }
32}
33
34// Draft-17 setup option names
35fn d17_option_name(key: u64) -> Option<&'static str> {
36    match key {
37        0x01 => Some("path"),
38        0x03 => Some("authorization_token"),
39        0x04 => Some("max_auth_token_cache_size"),
40        0x05 => Some("authority"),
41        0x07 => Some("moqt_implementation"),
42        _ => None,
43    }
44}
45
46/// Render a SUBSCRIPTION FILTER (0x21) parameter value: a Filter Type and the
47/// Start Location and End Group that type promises.
48///
49/// # Nothing has checked that the value holds the fields its Filter Type names
50///
51/// 0x21 is an odd Type, so `KeyValuePair::decode` keeps whatever
52/// length-prefixed bytes arrived and the value reaches here unexamined. None of
53/// `decode_parameters`' own checks looks at the *contents* of a filter value,
54/// and `crate::dispatch::AnyControlMessage::fields` renders every message that
55/// decoded — so a peer's bytes reach this function directly. That is the chain
56/// `tests/hostile_parameter_values.rs` sets out in full.
57///
58/// The truncation that follows an AbsoluteStart or AbsoluteRange Filter Type is
59/// the nastier shape, because the value looks well formed right up to the point
60/// where it is not: the Filter Type decodes cleanly and the Start Location it
61/// promises is simply not there.
62///
63/// # What a value it cannot read renders as
64///
65/// The raw bytes, as `fields::params` and `decode_largest_object` do. Field
66/// extraction runs on a message that has already decoded, so it has no refusal
67/// to give: what a peer sent is what there is to show. Every read below falls
68/// back to those bytes rather than unwrapping.
69fn decode_subscription_filter(bytes: &[u8]) -> Value {
70    let mut buf = bytes;
71    let Ok(filter_type) = VarInt::decode_moqt::<Wire>(&mut buf) else {
72        return Value::Bytes(bytes.to_vec());
73    };
74    let filter_type = filter_type.into_inner();
75    let mut obj = Map::new();
76    obj.insert("filter_type".into(), vi(filter_type));
77    match filter_type {
78        3 => {
79            let Ok(start_group) = VarInt::decode_moqt::<Wire>(&mut buf) else {
80                return Value::Bytes(bytes.to_vec());
81            };
82            let start_group = start_group.into_inner();
83            let Ok(start_object) = VarInt::decode_moqt::<Wire>(&mut buf) else {
84                return Value::Bytes(bytes.to_vec());
85            };
86            let start_object = start_object.into_inner();
87            obj.insert("start_group".into(), vi(start_group));
88            obj.insert("start_object".into(), vi(start_object));
89        }
90        4 => {
91            let Ok(start_group) = VarInt::decode_moqt::<Wire>(&mut buf) else {
92                return Value::Bytes(bytes.to_vec());
93            };
94            let start_group = start_group.into_inner();
95            let Ok(start_object) = VarInt::decode_moqt::<Wire>(&mut buf) else {
96                return Value::Bytes(bytes.to_vec());
97            };
98            let start_object = start_object.into_inner();
99            let Ok(end_group) = VarInt::decode_moqt::<Wire>(&mut buf) else {
100                return Value::Bytes(bytes.to_vec());
101            };
102            let end_group = end_group.into_inner();
103            obj.insert("start_group".into(), vi(start_group));
104            obj.insert("start_object".into(), vi(start_object));
105            obj.insert("end_group".into(), vi(end_group));
106        }
107        _ => {}
108    }
109    Value::Map(obj)
110}
111
112fn auth_token_to_json_d17(bytes: &[u8]) -> Value {
113    let mut buf = bytes;
114    let alias_type = match VarInt::decode_moqt::<Wire>(&mut buf) {
115        Ok(v) => v,
116        Err(_) => return Value::Bytes(bytes.to_vec()),
117    };
118    let at = alias_type.into_inner();
119    let mut o = Map::new();
120    o.insert("alias_type".into(), vi(at));
121    match at {
122        0 | 2 => {
123            if let Ok(ta) = VarInt::decode_moqt::<Wire>(&mut buf) {
124                o.insert("token_alias".into(), vi(ta.into_inner()));
125            }
126        }
127        1 => {
128            if let Ok(ta) = VarInt::decode_moqt::<Wire>(&mut buf) {
129                o.insert("token_alias".into(), vi(ta.into_inner()));
130            }
131            if let Ok(tt) = VarInt::decode_moqt::<Wire>(&mut buf) {
132                o.insert("token_type".into(), vi(tt.into_inner()));
133            }
134            // Draft-17: token_value runs to end of bytes (no inner length).
135            o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
136        }
137        _ => {
138            if let Ok(tt) = VarInt::decode_moqt::<Wire>(&mut buf) {
139                o.insert("token_type".into(), vi(tt.into_inner()));
140            }
141            o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
142        }
143    }
144    Value::Map(o)
145}
146
147/// Render a LARGEST OBJECT (0x09) parameter value: a Group and an Object, as
148/// two varints.
149///
150/// # What a value it cannot read renders as
151///
152/// The raw bytes, as `fields::params` does. Field extraction runs on a message
153/// that has already decoded, so it has no refusal to give, and it has no
154/// guarantee the two varints are there: nothing between `KeyValuePair::decode`
155/// and here looks at a 0x09 value's contents. An empty value fails the first
156/// read and a single `0x00` fails the second, which is the nastier of the two —
157/// the value looks well formed right up to the point where it is not. Neither
158/// read may panic on it.
159fn decode_largest_object(bytes: &[u8]) -> Value {
160    let mut buf = bytes;
161    let Ok(group) = VarInt::decode_moqt::<Wire>(&mut buf) else {
162        return Value::Bytes(bytes.to_vec());
163    };
164    let Ok(object) = VarInt::decode_moqt::<Wire>(&mut buf) else {
165        return Value::Bytes(bytes.to_vec());
166    };
167    let (group, object) = (group.into_inner(), object.into_inner());
168    let mut obj = Map::new();
169    obj.insert("group".into(), vi(group));
170    obj.insert("object".into(), vi(object));
171    Value::Map(obj)
172}
173
174fn params_to_json(params: &[KeyValuePair]) -> Value {
175    crate::fields::kvp_entries(params, |key, value| {
176        let Some(name) = d17_param_name(key) else {
177            return (None, None);
178        };
179        let rendered = match (value, key) {
180            (KvpValue::Bytes(b), 0x21) => decode_subscription_filter(b),
181            (KvpValue::Bytes(b), 0x09) => decode_largest_object(b),
182            (KvpValue::Bytes(b), _) if name == "authorization_token" => auth_token_to_json_d17(b),
183            (KvpValue::Varint(v), _) => vi(v.into_inner()),
184            (KvpValue::Bytes(b), _) => Value::Text(String::from_utf8_lossy(b).into_owned()),
185        };
186        (Some(name), Some(rendered))
187    })
188}
189
190pub(crate) fn options_to_json(options: &[KeyValuePair]) -> Value {
191    crate::fields::kvp_entries(options, |key, value| {
192        let Some(name) = d17_option_name(key) else {
193            return (None, None);
194        };
195        let rendered = match value {
196            KvpValue::Varint(v) => vi(v.into_inner()),
197            KvpValue::Bytes(b) if name == "authorization_token" => auth_token_to_json_d17(b),
198            KvpValue::Bytes(b) => Value::Text(String::from_utf8_lossy(b).into_owned()),
199        };
200        (Some(name), Some(rendered))
201    })
202}
203
204fn d17_track_prop_name(key: u64) -> Option<&'static str> {
205    match key {
206        0x02 => Some("delivery_timeout"),
207        0x04 => Some("max_cache_duration"),
208        0x0b => Some("immutable_properties"),
209        0x0e => Some("default_publisher_priority"),
210        0x22 => Some("default_publisher_group_order"),
211        0x30 => Some("dynamic_groups"),
212        _ => None,
213    }
214}
215
216fn track_props_to_json(props: &[KeyValuePair]) -> Value {
217    crate::fields::kvp_entries(props, |key, value| {
218        let name = d17_track_prop_name(key);
219        let rendered = match value {
220            KvpValue::Varint(v) => Some(vi(v.into_inner())),
221            // A property this draft does not name keeps its bytes rather than
222            // a name invented from its type, which is what an entry's absent
223            // `name` already says.
224            KvpValue::Bytes(_) => None,
225        };
226        (name, rendered)
227    })
228}
229
230/// This draft's field names for a decoded control message.
231///
232/// Keys are the names this draft gives its fields, in the order it defines
233/// them. An optional field the message did not carry is absent rather than
234/// zero.
235pub fn message_fields(msg: &ControlMessage) -> Map {
236    let obj = match msg {
237        ControlMessage::Setup(m) => {
238            let mut o = Map::new();
239            o.insert("options".into(), options_to_json(&m.options));
240            o
241        }
242        ControlMessage::GoAway(m) => {
243            let mut o = Map::new();
244            o.insert(
245                "new_session_uri".into(),
246                Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
247            );
248            o.insert("timeout".into(), vi(m.timeout.into_inner()));
249            o
250        }
251        ControlMessage::RequestOk(m) => {
252            let mut o = Map::new();
253            o.insert("parameters".into(), params_to_json(&m.parameters));
254            o
255        }
256        ControlMessage::RequestError(m) => {
257            let mut o = Map::new();
258            o.insert("error_code".into(), vi(m.error_code.into_inner()));
259            o.insert("retry_interval".into(), vi(m.retry_interval.into_inner()));
260            o.insert(
261                "reason_phrase".into(),
262                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
263            );
264            o
265        }
266        ControlMessage::Subscribe(m) => {
267            let mut o = Map::new();
268            o.insert("request_id".into(), vi(m.request_id.into_inner()));
269            o.insert(
270                "required_request_id_delta".into(),
271                vi(m.required_request_id_delta.into_inner()),
272            );
273            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
274            o.insert(
275                "track_name".into(),
276                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
277            );
278            o.insert("parameters".into(), params_to_json(&m.parameters));
279            o
280        }
281        ControlMessage::SubscribeOk(m) => {
282            let mut o = Map::new();
283            o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
284            o.insert("parameters".into(), params_to_json(&m.parameters));
285            o.insert("track_properties".into(), track_props_to_json(&m.track_properties));
286            o
287        }
288        ControlMessage::RequestUpdate(m) => {
289            let mut o = Map::new();
290            o.insert("request_id".into(), vi(m.request_id.into_inner()));
291            o.insert(
292                "required_request_id_delta".into(),
293                vi(m.required_request_id_delta.into_inner()),
294            );
295            o.insert("parameters".into(), params_to_json(&m.parameters));
296            o
297        }
298        ControlMessage::Publish(m) => {
299            let mut o = Map::new();
300            o.insert("request_id".into(), vi(m.request_id.into_inner()));
301            o.insert(
302                "required_request_id_delta".into(),
303                vi(m.required_request_id_delta.into_inner()),
304            );
305            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
306            o.insert(
307                "track_name".into(),
308                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
309            );
310            o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
311            o.insert("parameters".into(), params_to_json(&m.parameters));
312            o.insert("track_properties".into(), track_props_to_json(&m.track_properties));
313            o
314        }
315        ControlMessage::PublishOk(m) => {
316            let mut o = Map::new();
317            o.insert("parameters".into(), params_to_json(&m.parameters));
318            o
319        }
320        ControlMessage::PublishDone(m) => {
321            let mut o = Map::new();
322            o.insert("status_code".into(), vi(m.status_code.into_inner()));
323            o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
324            o.insert(
325                "reason_phrase".into(),
326                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
327            );
328            o
329        }
330        ControlMessage::PublishNamespace(m) => {
331            let mut o = Map::new();
332            o.insert("request_id".into(), vi(m.request_id.into_inner()));
333            o.insert(
334                "required_request_id_delta".into(),
335                vi(m.required_request_id_delta.into_inner()),
336            );
337            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
338            o.insert("parameters".into(), params_to_json(&m.parameters));
339            o
340        }
341        ControlMessage::Namespace(m) => {
342            let mut o = Map::new();
343            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
344            o
345        }
346        ControlMessage::NamespaceDone(m) => {
347            let mut o = Map::new();
348            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
349            o
350        }
351        ControlMessage::SubscribeNamespace(m) => {
352            let mut o = Map::new();
353            o.insert("request_id".into(), vi(m.request_id.into_inner()));
354            o.insert(
355                "required_request_id_delta".into(),
356                vi(m.required_request_id_delta.into_inner()),
357            );
358            o.insert("namespace_prefix".into(), ns_to_json(&m.namespace_prefix));
359            o.insert("subscribe_options".into(), vi(m.subscribe_options.into_inner()));
360            o.insert("parameters".into(), params_to_json(&m.parameters));
361            o
362        }
363        ControlMessage::TrackStatus(m) => {
364            let mut o = Map::new();
365            o.insert("request_id".into(), vi(m.request_id.into_inner()));
366            o.insert(
367                "required_request_id_delta".into(),
368                vi(m.required_request_id_delta.into_inner()),
369            );
370            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
371            o.insert(
372                "track_name".into(),
373                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
374            );
375            o.insert("parameters".into(), params_to_json(&m.parameters));
376            o
377        }
378        ControlMessage::Fetch(m) => {
379            let mut o = Map::new();
380            o.insert("request_id".into(), vi(m.request_id.into_inner()));
381            o.insert(
382                "required_request_id_delta".into(),
383                vi(m.required_request_id_delta.into_inner()),
384            );
385            o.insert("fetch_type".into(), vi(m.fetch_type as u64));
386            match &m.fetch_payload {
387                crate::draft17::message::FetchPayload::Standalone {
388                    track_namespace,
389                    track_name,
390                    start_group,
391                    start_object,
392                    end_group,
393                    end_object,
394                } => {
395                    o.insert("track_namespace".into(), ns_to_json(track_namespace));
396                    o.insert(
397                        "track_name".into(),
398                        Value::Text(String::from_utf8_lossy(track_name).into_owned()),
399                    );
400                    o.insert("start_group".into(), vi(start_group.into_inner()));
401                    o.insert("start_object".into(), vi(start_object.into_inner()));
402                    o.insert("end_group".into(), vi(end_group.into_inner()));
403                    o.insert("end_object".into(), vi(end_object.into_inner()));
404                }
405                crate::draft17::message::FetchPayload::Joining {
406                    joining_request_id,
407                    joining_start,
408                } => {
409                    o.insert("joining_request_id".into(), vi(joining_request_id.into_inner()));
410                    o.insert("joining_start".into(), vi(joining_start.into_inner()));
411                }
412            }
413            o.insert("parameters".into(), params_to_json(&m.parameters));
414            o
415        }
416        ControlMessage::FetchOk(m) => {
417            let mut o = Map::new();
418            o.insert("end_of_track".into(), vi(m.end_of_track as u64));
419            o.insert("end_group".into(), vi(m.end_group.into_inner()));
420            o.insert("end_object".into(), vi(m.end_object.into_inner()));
421            o.insert("parameters".into(), params_to_json(&m.parameters));
422            o.insert("track_properties".into(), track_props_to_json(&m.track_properties));
423            o
424        }
425        ControlMessage::PublishBlocked(m) => {
426            let mut o = Map::new();
427            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
428            o.insert(
429                "track_name".into(),
430                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
431            );
432            o
433        }
434    };
435    obj
436}
moqtap_codec/draft17/fields.rs

moqtap_codec/draft17/
fields.rs
