1use serde::Serialize;
2use serde_json::{Map, Value};
3use std::fmt;
4
5pub const CHANNEL_ENVELOPE_MAGIC: u8 = 0x7f;
6pub const CHANNEL_ENVELOPE_VERSION: u8 = 0x01;
7pub const CHANNEL_ENVELOPE_HEADER_BYTES: usize = 1 + 1 + 2 + 2;
8pub const DEFAULT_PEER_CHANNEL_ID: &str = "system/peer-default";
9pub const PUBLIC_PEER_STREAM_CHANNEL_ID: &str = "openrtc-peer-stream/1";
13pub const EXPLICIT_PROTOCOL_BYTE: u8 = 0x02;
14pub const NATIVE_MAIN_PROTOCOL_BYTE: u8 = 0x00;
15pub const NATIVE_MAIN_LABEL: &str = "main";
16pub const NATIVE_SIGNAL_LABEL: &str = "signal";
17pub const NATIVE_MAIN_MAX_LABEL_BYTES: usize = 64;
18
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
20#[serde(rename_all = "kebab-case")]
21pub enum ProtocolHint {
22 Control,
23 Explicit,
24 NativeMain,
25 Unknown,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
29#[serde(rename_all = "camelCase")]
30pub struct ChannelMetadata {
31 pub channel_id: String,
32 #[serde(skip_serializing_if = "Option::is_none")]
33 pub metadata: Option<Map<String, Value>>,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct StreamInspection {
38 pub protocol_hint: ProtocolHint,
39 pub channel: Option<ChannelMetadata>,
40 pub consumed_prefix_len: usize,
41 pub native_main_label: Option<String>,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub enum InspectionDecision {
46 NeedMore(usize),
47 Complete(StreamInspection),
48}
49
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct EnvelopePrefix {
52 pub channel: ChannelMetadata,
53 pub consumed_len: usize,
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub enum DecodeDecision {
58 NeedMore(usize),
59 NotMatched,
60 Decoded(EnvelopePrefix),
61}
62
63#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
64#[serde(tag = "status", rename_all = "kebab-case")]
65pub enum ExportedInspectionDecision {
66 NeedMore {
67 #[serde(rename = "requiredBytes")]
68 required_bytes: usize,
69 },
70 Complete {
71 #[serde(rename = "protocolHint")]
72 protocol_hint: ProtocolHint,
73 channel: Option<ChannelMetadata>,
74 #[serde(rename = "consumedPrefixLen")]
75 consumed_prefix_len: usize,
76 #[serde(rename = "nativeMainLabel")]
77 native_main_label: Option<String>,
78 },
79}
80
81#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
82#[serde(tag = "status", rename_all = "kebab-case")]
83pub enum ExportedDecodeDecision {
84 NeedMore {
85 #[serde(rename = "requiredBytes")]
86 required_bytes: usize,
87 },
88 NotMatched,
89 Decoded {
90 channel: ChannelMetadata,
91 #[serde(rename = "consumedPrefixLen")]
92 consumed_prefix_len: usize,
93 },
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub enum EncodeError {
98 EmptyChannelId,
99 ChannelIdTooLarge(usize),
100 MetadataTooLarge(usize),
101 MetadataSerialize(String),
102}
103
104impl fmt::Display for EncodeError {
105 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
106 match self {
107 Self::EmptyChannelId => write!(f, "channel envelope requires a non-empty channel id"),
108 Self::ChannelIdTooLarge(len) => {
109 write!(f, "channel envelope channel id is too large: {len} bytes")
110 }
111 Self::MetadataTooLarge(len) => {
112 write!(f, "channel envelope metadata is too large: {len} bytes")
113 }
114 Self::MetadataSerialize(error) => {
115 write!(f, "channel envelope metadata failed to serialize: {error}")
116 }
117 }
118 }
119}
120
121impl std::error::Error for EncodeError {}
122
123#[derive(Debug, Clone, PartialEq, Eq)]
124enum PrefixParseResult<T> {
125 NeedMore(usize),
126 NotMatched,
127 Parsed(T),
128}
129
130pub fn encode_envelope(
131 channel_id: &str,
132 metadata: Option<&Map<String, Value>>,
133) -> Result<Vec<u8>, EncodeError> {
134 let channel_id = channel_id.trim();
135 if channel_id.is_empty() {
136 return Err(EncodeError::EmptyChannelId);
137 }
138
139 let channel_bytes = channel_id.as_bytes();
140 if channel_bytes.len() > u16::MAX as usize {
141 return Err(EncodeError::ChannelIdTooLarge(channel_bytes.len()));
142 }
143
144 let metadata_bytes = metadata
145 .map(|value| {
146 serde_json::to_vec(value)
147 .map_err(|error| EncodeError::MetadataSerialize(error.to_string()))
148 })
149 .transpose()?
150 .unwrap_or_default();
151 if metadata_bytes.len() > u16::MAX as usize {
152 return Err(EncodeError::MetadataTooLarge(metadata_bytes.len()));
153 }
154
155 let mut bytes = Vec::with_capacity(
156 CHANNEL_ENVELOPE_HEADER_BYTES + channel_bytes.len() + metadata_bytes.len(),
157 );
158 bytes.push(CHANNEL_ENVELOPE_MAGIC);
159 bytes.push(CHANNEL_ENVELOPE_VERSION);
160 bytes.extend_from_slice(&(channel_bytes.len() as u16).to_be_bytes());
161 bytes.extend_from_slice(&(metadata_bytes.len() as u16).to_be_bytes());
162 bytes.extend_from_slice(channel_bytes);
163 bytes.extend_from_slice(&metadata_bytes);
164 Ok(bytes)
165}
166
167pub fn decode_prefix(bytes: &[u8]) -> DecodeDecision {
168 match parse_channel_envelope(bytes) {
169 PrefixParseResult::NeedMore(required) => DecodeDecision::NeedMore(required),
170 PrefixParseResult::NotMatched => DecodeDecision::NotMatched,
171 PrefixParseResult::Parsed(prefix) => DecodeDecision::Decoded(prefix),
172 }
173}
174
175pub fn decode_prefix_for_binding(bytes: &[u8]) -> ExportedDecodeDecision {
176 match decode_prefix(bytes) {
177 DecodeDecision::NeedMore(required_bytes) => {
178 ExportedDecodeDecision::NeedMore { required_bytes }
179 }
180 DecodeDecision::NotMatched => ExportedDecodeDecision::NotMatched,
181 DecodeDecision::Decoded(prefix) => ExportedDecodeDecision::Decoded {
182 channel: prefix.channel,
183 consumed_prefix_len: prefix.consumed_len,
184 },
185 }
186}
187
188pub fn inspect_prefix(bytes: &[u8], reached_eof: bool) -> InspectionDecision {
189 let channel_prefix = match parse_channel_envelope(bytes) {
190 PrefixParseResult::Parsed(channel) => Some(channel),
191 PrefixParseResult::NeedMore(required) if !reached_eof => {
192 return InspectionDecision::NeedMore(required);
193 }
194 PrefixParseResult::NeedMore(_) | PrefixParseResult::NotMatched => None,
195 };
196
197 let channel = channel_prefix.as_ref().map(|value| value.channel.clone());
198 let consumed_prefix_len = channel_prefix
199 .as_ref()
200 .map(|value| value.consumed_len)
201 .unwrap_or(0);
202 let remaining = &bytes[consumed_prefix_len..];
203
204 if remaining.is_empty() {
205 let protocol_hint = if reached_eof {
206 ProtocolHint::Unknown
207 } else {
208 return InspectionDecision::NeedMore(consumed_prefix_len + 1);
209 };
210
211 return InspectionDecision::Complete(StreamInspection {
212 protocol_hint,
213 channel,
214 consumed_prefix_len,
215 native_main_label: None,
216 });
217 }
218
219 let first_byte = remaining[0];
220 if first_byte == EXPLICIT_PROTOCOL_BYTE {
221 return InspectionDecision::Complete(StreamInspection {
222 protocol_hint: ProtocolHint::Explicit,
223 channel,
224 consumed_prefix_len,
225 native_main_label: None,
226 });
227 }
228
229 let protocol_hint = match parse_native_main_prefix(remaining) {
230 PrefixParseResult::Parsed((native_main_prefix_len, label)) => {
231 return InspectionDecision::Complete(StreamInspection {
232 protocol_hint: ProtocolHint::NativeMain,
233 channel,
234 consumed_prefix_len: consumed_prefix_len + native_main_prefix_len,
235 native_main_label: Some(label),
236 });
237 }
238 PrefixParseResult::NeedMore(required) if !reached_eof => {
239 return InspectionDecision::NeedMore(consumed_prefix_len + required);
240 }
241 PrefixParseResult::NeedMore(_) | PrefixParseResult::NotMatched => ProtocolHint::Control,
242 };
243
244 InspectionDecision::Complete(StreamInspection {
245 protocol_hint,
246 channel,
247 consumed_prefix_len,
248 native_main_label: None,
249 })
250}
251
252pub fn inspect_prefix_for_binding(bytes: &[u8], reached_eof: bool) -> ExportedInspectionDecision {
253 match inspect_prefix(bytes, reached_eof) {
254 InspectionDecision::NeedMore(required_bytes) => {
255 ExportedInspectionDecision::NeedMore { required_bytes }
256 }
257 InspectionDecision::Complete(inspection) => ExportedInspectionDecision::Complete {
258 protocol_hint: inspection.protocol_hint,
259 channel: inspection.channel,
260 consumed_prefix_len: inspection.consumed_prefix_len,
261 native_main_label: inspection.native_main_label,
262 },
263 }
264}
265
266fn parse_channel_envelope(bytes: &[u8]) -> PrefixParseResult<EnvelopePrefix> {
267 if bytes.is_empty() {
268 return PrefixParseResult::NeedMore(1);
269 }
270
271 if bytes[0] != CHANNEL_ENVELOPE_MAGIC {
272 return PrefixParseResult::NotMatched;
273 }
274
275 if bytes.len() < 2 {
276 return PrefixParseResult::NeedMore(2);
277 }
278
279 if bytes[1] != CHANNEL_ENVELOPE_VERSION {
280 return PrefixParseResult::NotMatched;
281 }
282
283 if bytes.len() < CHANNEL_ENVELOPE_HEADER_BYTES {
284 return PrefixParseResult::NeedMore(CHANNEL_ENVELOPE_HEADER_BYTES);
285 }
286
287 let channel_len = u16::from_be_bytes([bytes[2], bytes[3]]) as usize;
288 let metadata_len = u16::from_be_bytes([bytes[4], bytes[5]]) as usize;
289 let total_envelope_bytes = CHANNEL_ENVELOPE_HEADER_BYTES + channel_len + metadata_len;
290
291 if channel_len == 0 {
292 return PrefixParseResult::NotMatched;
293 }
294
295 if bytes.len() < total_envelope_bytes {
296 return PrefixParseResult::NeedMore(total_envelope_bytes);
297 }
298
299 let channel_start = CHANNEL_ENVELOPE_HEADER_BYTES;
300 let channel_end = channel_start + channel_len;
301 let metadata_end = channel_end + metadata_len;
302 let channel_id = String::from_utf8_lossy(&bytes[channel_start..channel_end])
303 .trim()
304 .to_string();
305
306 if channel_id.is_empty() {
307 return PrefixParseResult::NotMatched;
308 }
309
310 let metadata = if metadata_len == 0 {
311 None
312 } else {
313 match serde_json::from_slice::<Value>(&bytes[channel_end..metadata_end]) {
314 Ok(Value::Object(map)) => Some(map),
315 _ => None,
316 }
317 };
318
319 PrefixParseResult::Parsed(EnvelopePrefix {
320 channel: ChannelMetadata {
321 channel_id,
322 metadata,
323 },
324 consumed_len: total_envelope_bytes,
325 })
326}
327
328fn parse_native_main_prefix(bytes: &[u8]) -> PrefixParseResult<(usize, String)> {
329 if bytes.is_empty() {
330 return PrefixParseResult::NeedMore(1);
331 }
332
333 if bytes[0] != NATIVE_MAIN_PROTOCOL_BYTE {
334 return PrefixParseResult::NotMatched;
335 }
336
337 if bytes.len() < 5 {
338 return PrefixParseResult::NeedMore(5);
339 }
340
341 let label_len = u32::from_be_bytes([bytes[1], bytes[2], bytes[3], bytes[4]]) as usize;
342 if label_len == 0 || label_len > NATIVE_MAIN_MAX_LABEL_BYTES {
343 return PrefixParseResult::NotMatched;
344 }
345
346 let total_prefix_bytes = 5 + label_len;
347 if bytes.len() < total_prefix_bytes {
348 return PrefixParseResult::NeedMore(total_prefix_bytes);
349 }
350
351 let label = String::from_utf8_lossy(&bytes[5..total_prefix_bytes]).to_string();
352 if label != NATIVE_MAIN_LABEL && label != NATIVE_SIGNAL_LABEL {
353 return PrefixParseResult::NotMatched;
354 }
355
356 PrefixParseResult::Parsed((total_prefix_bytes, label))
357}
358
359#[cfg(test)]
360mod tests {
361 use super::*;
362
363 fn encode_native_main_prefix() -> Vec<u8> {
364 let label = NATIVE_MAIN_LABEL.as_bytes();
365 let mut bytes = Vec::with_capacity(1 + 4 + label.len());
366 bytes.push(NATIVE_MAIN_PROTOCOL_BYTE);
367 bytes.extend_from_slice(&(label.len() as u32).to_be_bytes());
368 bytes.extend_from_slice(label);
369 bytes
370 }
371
372 #[test]
373 fn inspects_explicit_streams_without_consuming_protocol_byte() {
374 let bytes = vec![EXPLICIT_PROTOCOL_BYTE, 0xaa, 0xbb];
375 let inspected = inspect_prefix(&bytes, false);
376
377 assert_eq!(
378 inspected,
379 InspectionDecision::Complete(StreamInspection {
380 protocol_hint: ProtocolHint::Explicit,
381 channel: None,
382 consumed_prefix_len: 0,
383 native_main_label: None,
384 })
385 );
386 }
387
388 #[test]
389 fn inspects_channel_envelope_and_preserves_explicit_payload() {
390 let metadata = serde_json::json!({ "scope": "ticket-only" })
391 .as_object()
392 .cloned()
393 .expect("object metadata");
394 let mut bytes =
395 encode_envelope("share-control", Some(&metadata)).expect("channel envelope");
396 bytes.extend_from_slice(&[EXPLICIT_PROTOCOL_BYTE, 0x01, 0x02]);
397
398 let inspected = inspect_prefix(&bytes, false);
399
400 assert_eq!(
401 inspected,
402 InspectionDecision::Complete(StreamInspection {
403 protocol_hint: ProtocolHint::Explicit,
404 channel: Some(ChannelMetadata {
405 channel_id: "share-control".to_string(),
406 metadata: Some(
407 serde_json::json!({ "scope": "ticket-only" })
408 .as_object()
409 .cloned()
410 .expect("object metadata"),
411 ),
412 }),
413 consumed_prefix_len: CHANNEL_ENVELOPE_HEADER_BYTES
414 + "share-control".len()
415 + serde_json::to_string(&serde_json::json!({ "scope": "ticket-only" }))
416 .expect("json string")
417 .len(),
418 native_main_label: None,
419 })
420 );
421 }
422
423 #[test]
424 fn inspects_native_main_and_consumes_label_prefix_only() {
425 let mut bytes = encode_native_main_prefix();
426 bytes.extend_from_slice(&[0x00, 0x00, 0x00, 0x10]);
427
428 let inspected = inspect_prefix(&bytes, false);
429
430 assert_eq!(
431 inspected,
432 InspectionDecision::Complete(StreamInspection {
433 protocol_hint: ProtocolHint::NativeMain,
434 channel: None,
435 consumed_prefix_len: 1 + 4 + NATIVE_MAIN_LABEL.len(),
436 native_main_label: Some(NATIVE_MAIN_LABEL.to_string()),
437 })
438 );
439 }
440
441 #[test]
442 fn incomplete_native_main_prefix_falls_back_to_control_at_eof() {
443 let bytes = vec![NATIVE_MAIN_PROTOCOL_BYTE, 0x00];
444 let inspected = inspect_prefix(&bytes, true);
445
446 assert_eq!(
447 inspected,
448 InspectionDecision::Complete(StreamInspection {
449 protocol_hint: ProtocolHint::Control,
450 channel: None,
451 consumed_prefix_len: 0,
452 native_main_label: None,
453 })
454 );
455 }
456
457 #[test]
458 fn parsed_channel_with_no_remaining_payload_is_unknown_at_eof() {
459 let bytes = encode_envelope("share", None).expect("channel envelope");
460 let inspected = inspect_prefix(&bytes, true);
461
462 assert_eq!(
463 inspected,
464 InspectionDecision::Complete(StreamInspection {
465 protocol_hint: ProtocolHint::Unknown,
466 channel: Some(ChannelMetadata {
467 channel_id: "share".to_string(),
468 metadata: None,
469 }),
470 consumed_prefix_len: CHANNEL_ENVELOPE_HEADER_BYTES + "share".len(),
471 native_main_label: None,
472 })
473 );
474 }
475
476 #[test]
477 fn inspects_native_signal_with_the_shared_classifier() {
478 let label = NATIVE_SIGNAL_LABEL.as_bytes();
479 let mut bytes = vec![NATIVE_MAIN_PROTOCOL_BYTE];
480 bytes.extend_from_slice(&(label.len() as u32).to_be_bytes());
481 bytes.extend_from_slice(label);
482 bytes.extend_from_slice(&[0xaa]);
483
484 assert_eq!(
485 inspect_prefix(&bytes, false),
486 InspectionDecision::Complete(StreamInspection {
487 protocol_hint: ProtocolHint::NativeMain,
488 channel: None,
489 consumed_prefix_len: 1 + 4 + NATIVE_SIGNAL_LABEL.len(),
490 native_main_label: Some(NATIVE_SIGNAL_LABEL.to_string()),
491 })
492 );
493 }
494
495 #[test]
496 fn encodes_latency_probe_channel_envelope_golden_vector() {
497 let bytes = encode_envelope("system/device-latency-probe", None).expect("channel envelope");
498
499 assert_eq!(
500 bytes,
501 hex_to_bytes("7f01001b000073797374656d2f6465766963652d6c6174656e63792d70726f6265"),
502 );
503 }
504
505 #[test]
506 fn decodes_channel_envelope_prefix_directly() {
507 let metadata = serde_json::json!({ "scope": "ticket-only" })
508 .as_object()
509 .cloned()
510 .expect("object metadata");
511 let bytes = encode_envelope("share-control", Some(&metadata)).expect("channel envelope");
512
513 assert_eq!(
514 decode_prefix(&bytes),
515 DecodeDecision::Decoded(EnvelopePrefix {
516 channel: ChannelMetadata {
517 channel_id: "share-control".to_string(),
518 metadata: Some(metadata),
519 },
520 consumed_len: bytes.len(),
521 })
522 );
523 }
524
525 #[test]
526 fn channel_envelope_prefix_reports_each_incremental_read_boundary() {
527 let bytes = encode_envelope("drive-view", None).expect("channel envelope");
528 let mut visible = 1;
529 let mut requested_boundaries = Vec::new();
530
531 loop {
532 match decode_prefix(&bytes[..visible]) {
533 DecodeDecision::NeedMore(required) => {
534 assert!(required > visible, "decoder must make forward progress");
535 requested_boundaries.push(required);
536 visible = required;
537 }
538 DecodeDecision::Decoded(_) => break,
539 other => panic!("valid incremental envelope should decode, got {other:?}"),
540 }
541 }
542
543 assert_eq!(
544 requested_boundaries,
545 vec![2, CHANNEL_ENVELOPE_HEADER_BYTES, bytes.len()]
546 );
547 }
548
549 fn hex_to_bytes(value: &str) -> Vec<u8> {
550 value
551 .as_bytes()
552 .chunks(2)
553 .map(|chunk| {
554 let text = std::str::from_utf8(chunk).expect("hex utf8");
555 u8::from_str_radix(text, 16).expect("hex byte")
556 })
557 .collect()
558 }
559}