1use crate::MessageElement;
13use crate::{
14 DataEndpoint, TelemetryError, TelemetryResult, config::DataType, is_reliable_type,
15 message_meta, packet::Packet,
16};
17
18use crate::config::StandardSmallPayload;
19use crate::packet::{hash_bytes_u64, sender_address_u32};
20#[cfg(feature = "std")]
21use alloc::collections::BTreeMap;
22#[cfg(feature = "cryptography")]
23use alloc::vec;
24use alloc::{sync::Arc, vec::Vec};
25use crc32fast::Hasher as Crc32Hasher;
26#[cfg(feature = "std")]
27use std::sync::{Mutex, OnceLock};
28
29#[derive(Clone, Debug, PartialEq, Eq)]
33pub struct TelemetryEnvelope {
34 pub ty: DataType,
36 pub endpoints: Arc<[DataEndpoint]>,
38 pub sender: Arc<str>,
40 pub source_address: u32,
42 pub timestamp_ms: u64,
44 pub wire_shape: Option<MessageElement>,
46 pub target_senders: Arc<[u64]>,
48}
49
50#[derive(Clone, Copy, Debug, PartialEq, Eq)]
52pub struct ReliableHeader {
53 pub flags: u8,
54 pub seq: u32,
55 pub ack: u32,
56}
57
58pub const RELIABLE_FLAG_ACK_ONLY: u8 = 0x01;
60
61pub const RELIABLE_FLAG_UNORDERED: u8 = 0x02;
63
64pub const RELIABLE_FLAG_UNSEQUENCED: u8 = 0x80;
66
67pub const RELIABLE_HEADER_BYTES: usize = 1 + 4 + 4;
69pub const CRC32_BYTES: usize = 4;
71
72const FLAG_COMPRESSED_PAYLOAD: u8 = 0x01;
103const FLAG_RELIABLE_HEADER_PRESENT: u8 = 0x02;
104const FLAG_WIRE_CONTRACT: u8 = 0x04;
105const FLAG_PACKET_NONCE: u8 = 0x08;
106#[cfg(feature = "cryptography")]
107const FLAG_E2E_ENCRYPTED_PAYLOAD: u8 = 0x10;
108const FLAG_ENDPOINT_BITMAP_PRESENT: u8 = 0x20;
109const FLAG_ENDPOINT_LIST_PRESENT: u8 = 0x80;
110const FLAG_COMPACT_RELIABLE_HEADER: u8 = 0x40;
111const CONTRACT_FLAG_TARGETS: u8 = 0x01;
112const CONTRACT_FLAG_SHAPE: u8 = 0x02;
113const CONTRACT_FLAG_RELIABLE_HEADER: u8 = 0x04;
114const RELIABLE_WIRE_FLAG_SEQ_PRESENT: u8 = 0x04;
115const RELIABLE_WIRE_FLAG_ACK_PRESENT: u8 = 0x08;
116const RELIABLE_PUBLIC_FLAGS_MASK: u8 =
117 RELIABLE_FLAG_ACK_ONLY | RELIABLE_FLAG_UNORDERED | RELIABLE_FLAG_UNSEQUENCED;
118#[cfg(feature = "cryptography")]
119const E2E_NONCE_LEN: usize = 12;
120#[cfg(feature = "cryptography")]
121const E2E_TAG_CAP: usize = 32;
122
123#[derive(Clone, Debug, PartialEq, Eq)]
124struct WireContract {
125 shape: Option<MessageElement>,
126 target_senders: Option<Arc<[u64]>>,
127 has_reliable_header: bool,
128}
129
130#[inline]
132fn write_uleb128<T>(mut v: u64, out: &mut Vec<T>)
133where
134 T: From<u8>,
135{
136 loop {
137 let mut byte = (v & 0x7F) as u8;
138 v >>= 7;
139 if v != 0 {
140 byte |= 0x80;
141 }
142 out.push(T::from(byte));
143 if v == 0 {
144 break;
145 }
146 }
147}
148
149#[inline]
151fn read_uleb128(r: &mut ByteReader) -> Result<u64, TelemetryError> {
152 crate::varint::read_u64(r.buf, &mut r.off)
153}
154
155#[inline]
157fn uleb128_size(mut v: u64) -> usize {
158 let mut n = 1;
159 while v >= 0x80 {
160 v >>= 7;
161 n += 1;
162 }
163 n
164}
165
166#[inline]
168fn bitmap_popcount(bm: &[u8]) -> usize {
169 bm.iter().map(|b| b.count_ones() as usize).sum()
170}
171
172#[derive(Clone, Copy)]
177struct ByteReader<'a> {
178 buf: &'a [u8],
179 off: usize,
180}
181
182impl<'a> ByteReader<'a> {
183 fn new(buf: &'a [u8]) -> Self {
185 Self { buf, off: 0 }
186 }
187
188 fn remaining(&self) -> usize {
190 self.buf.len().saturating_sub(self.off)
191 }
192
193 #[inline(always)]
195 fn read_bytes(&mut self, n: usize) -> Result<&'a [u8], TelemetryError> {
196 if self.remaining() < n {
197 return Err(TelemetryError::Unpack("short read"));
198 }
199 let s = &self.buf[self.off..self.off + n];
200 self.off += n;
201 Ok(s)
202 }
203}
204
205#[inline]
206fn write_u32_le(v: u32, out: &mut Vec<u8>) {
207 out.extend_from_slice(&v.to_le_bytes());
208}
209
210#[inline]
211fn read_u32_le(r: &mut ByteReader) -> Result<u32, TelemetryError> {
212 let b = r.read_bytes(4)?;
213 Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
214}
215
216#[inline]
217fn write_wire_shape(shape: MessageElement, out: &mut Vec<u8>) -> Result<(), TelemetryError> {
218 let dt = crate::config::message_data_type_code(shape.data_type());
219 let class = crate::config::message_class_code(shape.message_type());
220 let mut packed = dt | (class << 4);
221 if matches!(shape, MessageElement::Static(_, _, _)) {
222 packed |= 1 << 6;
223 }
224 out.push(packed);
225 if let MessageElement::Static(count, _, _) = shape {
226 let count = u64::try_from(count).map_err(|_| TelemetryError::Pack("wire shape count"))?;
227 write_uleb128(count, out);
228 }
229 Ok(())
230}
231
232#[inline]
233fn decode_wire_shape(r: &mut ByteReader) -> Result<MessageElement, TelemetryError> {
234 let packed = r.read_bytes(1)?[0];
235 let dt = crate::config::message_data_type_from_code(packed & 0x0F)
236 .ok_or(TelemetryError::Unpack("wire shape type"))?;
237 let class = crate::config::message_class_from_code((packed >> 4) & 0x03)
238 .ok_or(TelemetryError::Unpack("wire shape class"))?;
239 if (packed & (1 << 6)) != 0 {
240 let count = usize::try_from(read_uleb128(r)?)
241 .map_err(|_| TelemetryError::Unpack("wire shape count"))?;
242 Ok(MessageElement::Static(count, dt, class))
243 } else {
244 Ok(MessageElement::Dynamic(dt, class))
245 }
246}
247
248#[inline]
249fn write_wire_contract(
250 shape: Option<MessageElement>,
251 target_senders: &[u64],
252 has_reliable_header: bool,
253 out: &mut Vec<u8>,
254) -> Result<(), TelemetryError> {
255 let mut flags = 0u8;
259 if !target_senders.is_empty() {
260 flags |= CONTRACT_FLAG_TARGETS;
261 }
262 if shape.is_some() {
263 flags |= CONTRACT_FLAG_SHAPE;
264 }
265 if has_reliable_header {
266 flags |= CONTRACT_FLAG_RELIABLE_HEADER;
267 }
268 out.push(flags);
269 if let Some(shape) = shape {
270 write_wire_shape(shape, out)?;
271 }
272 if !target_senders.is_empty() {
273 write_uleb128(target_senders.len() as u64, out);
274 for hash in target_senders {
275 out.extend_from_slice(&hash.to_le_bytes());
276 }
277 }
278 Ok(())
279}
280
281fn wire_contract_size(shape: Option<MessageElement>, target_senders: &[u64]) -> usize {
283 1 + shape.map_or(0, |shape| match shape {
284 MessageElement::Static(count, _, _) => 1 + uleb128_size(count as u64),
285 _ => 1,
286 }) + if target_senders.is_empty() {
287 0
288 } else {
289 uleb128_size(target_senders.len() as u64) + target_senders.len() * 8
290 }
291}
292
293#[inline]
294fn decode_wire_contract(
295 r: &mut ByteReader,
296 has_contract: bool,
297) -> Result<WireContract, TelemetryError> {
298 if !has_contract {
299 return Ok(WireContract {
300 shape: None,
301 target_senders: None,
302 has_reliable_header: false,
303 });
304 }
305 let contract_len = usize::try_from(read_uleb128(r)?)
306 .map_err(|_| TelemetryError::Unpack("wire contract length"))?;
307 let contract_bytes = r.read_bytes(contract_len)?;
308 let mut cr = ByteReader::new(contract_bytes);
309 let flags = cr.read_bytes(1)?[0];
313 let shape = if (flags & CONTRACT_FLAG_SHAPE) != 0 {
314 Some(decode_wire_shape(&mut cr)?)
315 } else {
316 None
317 };
318 let target_senders: Option<Arc<[u64]>> = if (flags & CONTRACT_FLAG_TARGETS) != 0 {
319 let count = usize::try_from(read_uleb128(&mut cr)?)
320 .map_err(|_| TelemetryError::Unpack("wire contract target count"))?;
321 if count > cr.remaining() / 8 {
322 return Err(TelemetryError::Unpack("wire contract target count"));
323 }
324 if count == 0 {
325 None
326 } else {
327 let mut targets = Arc::<[u64]>::new_uninit_slice(count);
328 for slot in Arc::get_mut(&mut targets).expect("new target storage is unique") {
329 let bytes = cr.read_bytes(8)?;
330 slot.write(u64::from_le_bytes([
331 bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
332 ]));
333 }
334 Some(unsafe { targets.assume_init() })
336 }
337 } else {
338 None
339 };
340 if cr.remaining() != 0 {
341 return Err(TelemetryError::Unpack("wire contract trailing bytes"));
342 }
343 Ok(WireContract {
344 shape,
345 target_senders,
346 has_reliable_header: (flags & CONTRACT_FLAG_RELIABLE_HEADER) != 0,
347 })
348}
349
350#[inline]
351fn crc32_bytes(data: &[u8]) -> u32 {
352 let mut hasher = Crc32Hasher::new();
353 hasher.update(data);
354 hasher.finalize()
355}
356
357#[inline]
358fn append_crc32(out: &mut Vec<u8>) {
359 let crc = crc32_bytes(out);
360 out.extend_from_slice(&crc.to_le_bytes());
361}
362
363fn join_frame_with_crc(header: &[u8], payload: &[u8]) -> Arc<[u8]> {
366 let mut hasher = Crc32Hasher::new();
367 hasher.update(header);
368 hasher.update(payload);
369 let crc = hasher.finalize().to_le_bytes();
370 let mut frame = Arc::<[u8]>::new_uninit_slice(header.len() + payload.len() + crc.len());
371 let bytes = Arc::get_mut(&mut frame).unwrap();
372 for (slot, value) in bytes
373 .iter_mut()
374 .zip(header.iter().chain(payload).chain(&crc))
375 {
376 slot.write(*value);
377 }
378 unsafe { frame.assume_init() }
381}
382
383#[inline]
384fn split_crc32(buf: &[u8]) -> Result<(&[u8], u32), TelemetryError> {
385 if buf.len() < CRC32_BYTES {
386 return Err(TelemetryError::Unpack("short buffer"));
387 }
388 let data_len = buf.len() - CRC32_BYTES;
389 let crc = u32::from_le_bytes([
390 buf[data_len],
391 buf[data_len + 1],
392 buf[data_len + 2],
393 buf[data_len + 3],
394 ]);
395 Ok((&buf[..data_len], crc))
396}
397
398#[cfg(all(test, feature = "std", feature = "discovery"))]
399std::thread_local! {
400 static CRC_VERIFICATIONS: core::cell::Cell<usize> = const { core::cell::Cell::new(0) };
401}
402#[cfg(all(test, feature = "std", feature = "discovery"))]
403pub(crate) fn crc_verification_count() -> usize {
404 CRC_VERIFICATIONS.with(core::cell::Cell::get)
405}
406
407#[inline]
408fn verify_crc32(buf: &[u8]) -> Result<&[u8], TelemetryError> {
409 #[cfg(all(test, feature = "std", feature = "discovery"))]
410 CRC_VERIFICATIONS.with(|count| count.set(count.get().wrapping_add(1)));
411 let (data, expected) = split_crc32(buf)?;
412 let actual = crc32_bytes(data);
413 if actual != expected {
414 return Err(TelemetryError::Unpack("crc32 mismatch"));
415 }
416 Ok(data)
417}
418
419#[cfg(feature = "cryptography")]
420#[inline]
421fn e2e_nonce_for_packet(pkt: &Packet) -> [u8; E2E_NONCE_LEN] {
422 let mut nonce = [0u8; E2E_NONCE_LEN];
423 nonce[..4].copy_from_slice(&pkt.data_type().as_u32().to_le_bytes());
424 nonce[4..10].copy_from_slice(&(pkt.timestamp() & 0x0000_FFFF_FFFF_FFFF).to_le_bytes()[..6]);
425 nonce[10..].copy_from_slice(&pkt.nonce().to_le_bytes());
426 nonce
427}
428
429#[cfg(feature = "cryptography")]
430fn write_encrypted_payload(
431 pkt: &Packet,
432 key_id: u32,
433 plaintext_wire_payload: &[u8],
434 out: &mut Vec<u8>,
435) -> TelemetryResult<()> {
436 let aad_end = out.len();
437 let nonce = e2e_nonce_for_packet(pkt);
438 let mut ciphertext = vec![0u8; plaintext_wire_payload.len()];
439 let mut tag = [0u8; E2E_TAG_CAP];
440 let (ciphertext_len, tag_len) = crate::crypto::seal_with_registered_crypto(
441 key_id,
442 &nonce,
443 &out[..aad_end],
444 plaintext_wire_payload,
445 &mut ciphertext,
446 &mut tag,
447 )?;
448 if ciphertext_len > ciphertext.len() || tag_len > tag.len() {
449 return Err(TelemetryError::SizeMismatchError);
450 }
451 write_uleb128(u64::from(key_id), out);
452 write_uleb128(plaintext_wire_payload.len() as u64, out);
453 write_uleb128(nonce.len() as u64, out);
454 out.extend_from_slice(&nonce);
455 write_uleb128(tag_len as u64, out);
456 out.extend_from_slice(&tag[..tag_len]);
457 out.extend_from_slice(&ciphertext[..ciphertext_len]);
458 Ok(())
459}
460
461#[cfg(feature = "cryptography")]
462fn read_encrypted_payload(
463 r: &mut ByteReader,
464 aad: &[u8],
465 plaintext_len: usize,
466) -> TelemetryResult<Vec<u8>> {
467 let key_id = u32::try_from(read_uleb128(r)?)
468 .map_err(|_| TelemetryError::Unpack("e2e key id too large"))?;
469 let wire_payload_len = usize::try_from(read_uleb128(r)?)
470 .map_err(|_| TelemetryError::Unpack("e2e payload length"))?;
471 if wire_payload_len > plaintext_len {
472 return Err(TelemetryError::Unpack("bad e2e payload length"));
473 }
474 let nonce_len = usize::try_from(read_uleb128(r)?)
475 .map_err(|_| TelemetryError::Unpack("e2e nonce length"))?;
476 if nonce_len == 0 || nonce_len > 64 {
477 return Err(TelemetryError::Unpack("bad e2e nonce length"));
478 }
479 let nonce = r.read_bytes(nonce_len)?;
480 let tag_len =
481 usize::try_from(read_uleb128(r)?).map_err(|_| TelemetryError::Unpack("e2e tag length"))?;
482 if tag_len == 0 || tag_len > E2E_TAG_CAP {
483 return Err(TelemetryError::Unpack("bad e2e tag length"));
484 }
485 let tag = r.read_bytes(tag_len)?;
486 let ciphertext_len = r.remaining();
487 let ciphertext = r.read_bytes(ciphertext_len)?;
488 let mut plaintext = vec![0u8; wire_payload_len];
489 let opened_len = crate::crypto::open_with_registered_crypto(
490 key_id,
491 nonce,
492 aad,
493 ciphertext,
494 tag,
495 &mut plaintext,
496 )?;
497 if opened_len != wire_payload_len {
498 return Err(TelemetryError::SizeMismatchError);
499 }
500 Ok(plaintext)
501}
502
503#[inline]
504fn write_reliable_header(h: ReliableHeader, out: &mut Vec<u8>) {
505 out.push(h.flags);
506 write_u32_le(h.seq, out);
507 write_u32_le(h.ack, out);
508}
509
510#[inline]
511fn reliable_compact_size(h: ReliableHeader) -> usize {
512 let seq_present = (h.flags & RELIABLE_FLAG_ACK_ONLY) == 0 || h.seq != 0;
513 let ack_present = h.ack != 0 || (h.flags & RELIABLE_FLAG_ACK_ONLY) != 0;
514 1 + if seq_present {
515 uleb128_size(h.seq as u64)
516 } else {
517 0
518 } + if ack_present {
519 uleb128_size(h.ack as u64)
520 } else {
521 0
522 }
523}
524
525#[inline]
526fn should_compact_reliable_header(h: ReliableHeader) -> bool {
527 reliable_compact_size(h) < RELIABLE_HEADER_BYTES
528}
529
530#[inline]
531fn reliable_wire_size(h: ReliableHeader, compact: bool) -> usize {
532 if compact {
533 reliable_compact_size(h)
534 } else {
535 RELIABLE_HEADER_BYTES
536 }
537}
538
539pub(crate) fn write_reliable_header_encoded(h: ReliableHeader, compact: bool, out: &mut Vec<u8>) {
540 if !compact {
541 write_reliable_header(h, out);
542 return;
543 }
544
545 let seq_present = (h.flags & RELIABLE_FLAG_ACK_ONLY) == 0 || h.seq != 0;
546 let ack_present = h.ack != 0 || (h.flags & RELIABLE_FLAG_ACK_ONLY) != 0;
547 let mut wire_flags = h.flags & RELIABLE_PUBLIC_FLAGS_MASK;
548 if seq_present {
549 wire_flags |= RELIABLE_WIRE_FLAG_SEQ_PRESENT;
550 }
551 if ack_present {
552 wire_flags |= RELIABLE_WIRE_FLAG_ACK_PRESENT;
553 }
554 out.push(wire_flags);
555 if seq_present {
556 write_uleb128(h.seq as u64, out);
557 }
558 if ack_present {
559 write_uleb128(h.ack as u64, out);
560 }
561}
562
563#[inline]
564fn read_reliable_header(r: &mut ByteReader) -> Result<ReliableHeader, TelemetryError> {
565 let flags = r.read_bytes(1)?[0];
566 let seq = read_u32_le(r)?;
567 let ack = read_u32_le(r)?;
568 Ok(ReliableHeader { flags, seq, ack })
569}
570
571fn read_reliable_header_compact(r: &mut ByteReader) -> Result<ReliableHeader, TelemetryError> {
572 let wire_flags = r.read_bytes(1)?[0];
573 let seq = if (wire_flags & RELIABLE_WIRE_FLAG_SEQ_PRESENT) != 0 {
574 u32::try_from(read_uleb128(r)?)
575 .map_err(|_| TelemetryError::Unpack("reliable seq too large"))?
576 } else {
577 0
578 };
579 let ack = if (wire_flags & RELIABLE_WIRE_FLAG_ACK_PRESENT) != 0 {
580 u32::try_from(read_uleb128(r)?)
581 .map_err(|_| TelemetryError::Unpack("reliable ack too large"))?
582 } else {
583 0
584 };
585 Ok(ReliableHeader {
586 flags: wire_flags & RELIABLE_PUBLIC_FLAGS_MASK,
587 seq,
588 ack,
589 })
590}
591
592#[inline]
593fn read_reliable_header_encoded(
594 r: &mut ByteReader,
595 compact: bool,
596) -> Result<ReliableHeader, TelemetryError> {
597 if compact {
598 read_reliable_header_compact(r)
599 } else {
600 read_reliable_header(r)
601 }
602}
603
604#[cfg(feature = "std")]
605static ADDRESS_BOOK: OnceLock<Mutex<BTreeMap<u32, Arc<str>>>> = OnceLock::new();
606
607#[cfg(feature = "std")]
608fn address_book() -> &'static Mutex<BTreeMap<u32, Arc<str>>> {
609 ADDRESS_BOOK.get_or_init(|| Mutex::new(BTreeMap::new()))
610}
611
612#[inline]
613pub(crate) fn source_address_for_sender(sender: &str) -> u32 {
614 let addr = sender_address_u32(sender);
615 remember_source_address(addr, sender);
616 addr
617}
618
619#[cfg(feature = "std")]
620pub(crate) fn remember_source_address(addr: u32, sender: &str) {
621 address_book()
622 .lock()
623 .expect("wire address book poisoned")
624 .entry(addr)
625 .or_insert_with(|| Arc::<str>::from(sender));
626}
627
628#[cfg(not(feature = "std"))]
629pub(crate) fn remember_source_address(_addr: u32, _sender: &str) {}
630
631#[cfg(feature = "std")]
632fn sender_name_for_address(addr: u32) -> Arc<str> {
633 address_book()
634 .lock()
635 .expect("wire address book poisoned")
636 .get(&addr)
637 .cloned()
638 .unwrap_or_else(|| format_sender_address(addr))
639}
640
641#[cfg(not(feature = "std"))]
642fn sender_name_for_address(addr: u32) -> Arc<str> {
643 format_sender_address(addr)
644}
645
646fn format_sender_address(addr: u32) -> Arc<str> {
647 let mut bytes = [0u8; 16];
649 let mut start = bytes.len();
650 let mut remaining = addr;
651 loop {
652 start -= 1;
653 bytes[start] = b'0' + (remaining % 10) as u8;
654 remaining /= 10;
655 if remaining == 0 {
656 break;
657 }
658 }
659 start -= 6;
660 bytes[start..start + 6].copy_from_slice(b"@addr:");
661 Arc::from(core::str::from_utf8(&bytes[start..]).expect("numeric alias is ASCII"))
662}
663
664const EP_BITMAP_BITS: usize = 256;
670#[cfg(test)]
671use crate::MAX_VALUE_DATA_ENDPOINT;
672
673const EP_BITMAP_BYTES: usize = EP_BITMAP_BITS.div_ceil(8);
675
676#[inline]
681fn build_endpoint_bitmap(eps: &[DataEndpoint]) -> [u8; EP_BITMAP_BYTES] {
682 let mut bm = [0u8; EP_BITMAP_BYTES];
683 for &ep in eps {
684 let idx = ep.as_u32() as usize;
685 debug_assert!(idx < EP_BITMAP_BITS, "endpoint discriminant out of range");
686 if idx < EP_BITMAP_BITS {
687 let byte = idx / 8;
688 let bit = idx % 8;
689 bm[byte] |= 1u8 << bit;
690 }
691 }
692 bm
693}
694
695fn expand_endpoint_bitmap(bm: &[u8]) -> TelemetryResult<Arc<[DataEndpoint]>> {
697 if bm.len() != EP_BITMAP_BYTES {
698 return Err(TelemetryError::Unpack("bad endpoint bitmap size"));
699 }
700 let count = bitmap_popcount(bm);
703 let mut endpoints = Arc::<[DataEndpoint]>::new_uninit_slice(count);
704 let slots = Arc::get_mut(&mut endpoints).expect("new endpoint storage is unique");
705 let mut written = 0;
706 for (byte_index, &byte) in bm.iter().enumerate() {
707 let mut bits = byte;
708 while bits != 0 {
709 let idx = byte_index * 8 + bits.trailing_zeros() as usize;
710 slots[written].write(
711 DataEndpoint::try_from_u32(idx as u32)
712 .ok_or(TelemetryError::Unpack("bad endpoint bit set"))?,
713 );
714 written += 1;
715 bits &= bits - 1;
716 }
717 }
718 debug_assert_eq!(written, count);
720 Ok(unsafe { endpoints.assume_init() })
721}
722
723pub(crate) fn unique_sorted_endpoints(
724 endpoints: &[DataEndpoint],
725) -> impl Iterator<Item = DataEndpoint> + '_ {
726 let mut previous = None;
727 core::iter::from_fn(move || {
728 let next = endpoints
729 .iter()
730 .copied()
731 .filter(|endpoint| previous.is_none_or(|previous| *endpoint > previous))
732 .min();
733 previous = next.or(previous);
736 next
737 })
738}
739
740fn endpoint_arc_from_iter(
741 count: usize,
742 endpoints: impl Iterator<Item = DataEndpoint>,
743) -> TelemetryResult<Arc<[DataEndpoint]>> {
744 let mut output = Arc::<[DataEndpoint]>::new_uninit_slice(count);
745 let slots = Arc::get_mut(&mut output).unwrap();
746 let mut written = 0;
747 for endpoint in endpoints {
748 if written == count {
749 return Err(TelemetryError::Unpack("endpoint count mismatch"));
750 }
751 slots[written].write(endpoint);
752 written += 1;
753 }
754 if written != count {
755 return Err(TelemetryError::Unpack("endpoint count mismatch"));
756 }
757 Ok(unsafe { output.assume_init() })
759}
760fn endpoint_list_size(eps: &[DataEndpoint]) -> usize {
761 unique_sorted_endpoints(eps)
762 .map(|ep| uleb128_size(u64::from(ep.0)))
763 .sum()
764}
765fn explicit_endpoint_list(eps: &[DataEndpoint]) -> bool {
766 eps.iter().any(|ep| ep.0 >= EP_BITMAP_BITS as u32) || endpoint_list_size(eps) < EP_BITMAP_BYTES
767}
768fn endpoints_match_schema(ty: DataType, eps: &[DataEndpoint]) -> bool {
769 let meta = message_meta(ty);
770 unique_sorted_endpoints(eps).eq(unique_sorted_endpoints(meta.endpoints_ref()))
771}
772fn read_endpoint_list<'a>(r: &mut ByteReader<'a>, count: usize) -> TelemetryResult<&'a [u8]> {
773 let start = r.off;
774 let mut previous = None;
775 for _ in 0..count {
776 let id = u32::try_from(read_uleb128(r)?)
777 .map_err(|_| TelemetryError::Unpack("endpoint too large"))?;
778 if previous.is_some_and(|previous| id <= previous) {
779 return Err(TelemetryError::Unpack(
780 "endpoint list is not unique and sorted",
781 ));
782 }
783 previous = Some(id);
784 }
785 Ok(&r.buf[start..r.off])
786}
787fn endpoint_list_iter(bytes: &[u8]) -> impl Iterator<Item = DataEndpoint> + '_ {
788 let mut r = ByteReader::new(bytes);
789 core::iter::from_fn(move || {
790 if r.remaining() == 0 {
791 return None;
792 }
793 read_uleb128(&mut r)
794 .ok()
795 .and_then(|id| u32::try_from(id).ok())
796 .map(DataEndpoint)
797 })
798}
799
800#[inline]
801fn schema_endpoints_from_type(ty: DataType, nep: usize) -> TelemetryResult<Arc<[DataEndpoint]>> {
802 let meta = message_meta(ty);
805 let schema = meta.endpoints_ref();
806 if schema.len() == nep
807 && schema
808 .windows(2)
809 .all(|pair| pair[0].as_u32() < pair[1].as_u32())
810 && schema
811 .iter()
812 .all(|ep| DataEndpoint::try_from_u32(ep.as_u32()).is_some())
813 {
814 #[cfg(feature = "std")]
817 return Ok(meta.endpoints);
818 #[cfg(not(feature = "std"))]
819 return Ok(Arc::from(schema));
820 }
821 endpoint_arc_from_iter(nep, unique_sorted_endpoints(schema))
822}
823
824#[inline]
825fn endpoints_from_wire_or_schema(
826 r: &mut ByteReader<'_>,
827 bitmap_present: bool,
828 list_present: bool,
829 ty: Option<DataType>,
830 nep: usize,
831) -> TelemetryResult<Arc<[DataEndpoint]>> {
832 if list_present {
833 if bitmap_present {
834 return Err(TelemetryError::Unpack("ambiguous endpoint encoding"));
835 }
836 let bytes = read_endpoint_list(r, nep)?;
837 #[cfg(feature = "std")]
838 if let Some(meta) = ty.and_then(crate::config::try_get_message_meta)
839 && endpoint_list_iter(bytes).eq(meta.endpoints.iter().copied())
840 {
841 return Ok(meta.endpoints);
842 }
843 return endpoint_arc_from_iter(nep, endpoint_list_iter(bytes));
844 }
845 if bitmap_present {
846 let bm = r.read_bytes(EP_BITMAP_BYTES)?;
847 let endpoints = expand_endpoint_bitmap(bm)?;
848 if endpoints.len() != nep {
849 return Err(TelemetryError::Unpack("endpoint count mismatch"));
850 }
851 Ok(endpoints)
852 } else {
853 let ty = ty.ok_or(TelemetryError::InvalidType)?;
854 schema_endpoints_from_type(ty, nep)
855 }
856}
857
858#[inline]
859fn data_type_id_from_wire(ty_v: u64) -> TelemetryResult<u32> {
860 let ty_u32 = u32::try_from(ty_v).map_err(|_| TelemetryError::Unpack("type too large"))?;
861 Ok(ty_u32)
862}
863
864pub fn pack_packet(pkt: &Packet) -> Arc<[u8]> {
877 if is_reliable_type(pkt.data_type()) {
878 let hdr = ReliableHeader {
881 flags: RELIABLE_FLAG_UNSEQUENCED,
882 seq: 0,
883 ack: 0,
884 };
885 return pack_packet_with_reliable(pkt, hdr);
886 }
887 pack_packet_inner(pkt, None)
888}
889
890pub fn pack_packet_with_reliable(pkt: &Packet, header: ReliableHeader) -> Arc<[u8]> {
894 pack_packet_inner(pkt, Some(header))
895}
896
897pub fn pack_reliable_ack(sender: &str, ty: DataType, timestamp_ms: u64, ack: u32) -> Arc<[u8]> {
902 let source_address = source_address_for_sender(sender);
903
904 let mut out = Vec::with_capacity(32 + EP_BITMAP_BYTES + CRC32_BYTES);
906
907 let flags: u8 = FLAG_ENDPOINT_LIST_PRESENT | FLAG_RELIABLE_HEADER_PRESENT;
908 out.push(flags);
909 out.push(0u8); write_uleb128(ty.as_u32() as u64, &mut out);
912 write_uleb128(0u64, &mut out); write_uleb128(timestamp_ms, &mut out);
914 write_uleb128(source_address as u64, &mut out);
915
916 let reliable = ReliableHeader {
917 flags: RELIABLE_FLAG_ACK_ONLY,
918 seq: 0,
919 ack,
920 };
921 if should_compact_reliable_header(reliable) {
922 out[0] |= FLAG_COMPACT_RELIABLE_HEADER;
923 write_reliable_header_encoded(reliable, true, &mut out);
924 } else {
925 write_reliable_header_encoded(reliable, false, &mut out);
926 }
927 append_crc32(&mut out);
928
929 Arc::<[u8]>::from(out)
930}
931
932fn payload_for_wire(pkt: &Packet) -> (bool, alloc::borrow::Cow<'_, [u8]>) {
935 payload_for_encoding(pkt, true)
936}
937
938fn payload_for_encoding(
939 pkt: &Packet,
940 #[cfg_attr(not(feature = "compression"), allow(unused_variables))] compression_allowed: bool,
941) -> (bool, alloc::borrow::Cow<'_, [u8]>) {
942 #[cfg(feature = "compression")]
943 if !compression_allowed {
944 return (false, alloc::borrow::Cow::Borrowed(pkt.payload()));
945 }
946 #[cfg(feature = "discovery")]
947 if crate::discovery::is_discovery_type(pkt.data_type()) {
948 return (false, alloc::borrow::Cow::Borrowed(pkt.payload()));
949 }
950 payload_compression::compress_if_beneficial(pkt.payload())
951}
952
953fn pack_packet_inner(pkt: &Packet, reliable: Option<ReliableHeader>) -> Arc<[u8]> {
954 pack_packet_inner_with_contract(
955 pkt,
956 reliable,
957 pkt.wire_shape(),
958 pkt.wire_target_senders(),
959 None,
960 )
961 .expect("plaintext packet packing failed")
962}
963
964#[cfg(any(test, not(feature = "compression")))]
984pub(crate) fn pack_packet_with_wire_contract(
985 pkt: &Packet,
986 reliable: Option<ReliableHeader>,
987 shape: Option<MessageElement>,
988 target_senders: &[u64],
989) -> TelemetryResult<Arc<[u8]>> {
990 pack_packet_inner_with_contract(pkt, reliable, shape, target_senders, None)
991}
992
993#[derive(Clone, Copy, Debug)]
994#[cfg_attr(not(feature = "cryptography"), allow(dead_code))]
995pub(crate) struct E2eSealConfig {
996 pub key_id: u32,
997}
998
999#[cfg(all(feature = "cryptography", not(feature = "compression")))]
1000pub(crate) fn pack_packet_with_wire_contract_e2e(
1001 pkt: &Packet,
1002 reliable: Option<ReliableHeader>,
1003 shape: Option<MessageElement>,
1004 target_senders: &[u64],
1005 e2e: E2eSealConfig,
1006) -> TelemetryResult<Arc<[u8]>> {
1007 pack_packet_inner_with_contract(pkt, reliable, shape, target_senders, Some(e2e))
1008}
1009
1010fn pack_packet_inner_with_contract(
1011 pkt: &Packet,
1012 reliable: Option<ReliableHeader>,
1013 shape: Option<MessageElement>,
1014 target_senders: &[u64],
1015 e2e: Option<E2eSealConfig>,
1016) -> TelemetryResult<Arc<[u8]>> {
1017 with_packet_frame_parts(
1018 pkt,
1019 reliable,
1020 shape,
1021 target_senders,
1022 e2e,
1023 true,
1024 |header, payload| Ok(join_frame_with_crc(header, payload)),
1025 )
1026}
1027
1028#[cfg(feature = "compression")]
1032pub(crate) fn pack_routed_packet(
1033 pkt: &Packet,
1034 reliable: Option<ReliableHeader>,
1035 shape: Option<MessageElement>,
1036 target_senders: &[u64],
1037 e2e: Option<E2eSealConfig>,
1038) -> TelemetryResult<Arc<[u8]>> {
1039 with_packet_frame_parts(
1040 pkt,
1041 reliable,
1042 shape,
1043 target_senders,
1044 e2e,
1045 false,
1046 |header, payload| Ok(join_frame_with_crc(header, payload)),
1047 )
1048}
1049
1050#[cfg(feature = "compression")]
1054pub(crate) fn payload_encoding_for_side(
1055 raw: &[u8],
1056 allowed: bool,
1057 max_decoded_bytes: usize,
1058) -> TelemetryResult<Option<Arc<[u8]>>> {
1059 let flags = *raw.first().ok_or(TelemetryError::Unpack("short prelude"))?;
1060 let compressed = flags & FLAG_COMPRESSED_PAYLOAD != 0;
1061 if flags & 0x10 != 0 {
1062 if compressed && !allowed {
1063 return Err(TelemetryError::Io(
1064 "opaque compressed payload requires codec-capable peers",
1065 ));
1066 }
1067 return Ok(None);
1068 }
1069 if !compressed && !allowed {
1070 return Ok(None);
1071 }
1072 if !compressed {
1076 let mut prelude = ByteReader::new(raw);
1077 prelude.read_bytes(2)?;
1078 read_uleb128(&mut prelude)?;
1079 if read_uleb128(&mut prelude)? < crate::config::runtime_payload_compress_threshold() as u64
1080 {
1081 return Ok(None);
1082 }
1083 }
1084 let data = verify_crc32(raw)?;
1085 let mut reader = ByteReader::new(data);
1086 let envelope = read_routing_envelope(&mut reader)?;
1087 #[cfg(feature = "discovery")]
1088 let allowed = allowed && !crate::discovery::is_discovery_type(envelope.ty);
1089 if envelope.is_reliable || envelope.has_reliable_header {
1090 read_reliable_header_encoded(&mut reader, flags & FLAG_COMPACT_RELIABLE_HEADER != 0)?;
1091 }
1092 let header = &data[..reader.off];
1093 let payload = &data[reader.off..];
1094 if compressed && allowed {
1095 return Ok(None);
1096 }
1097 if !compressed && !allowed {
1098 return Ok(None);
1099 }
1100 let mut prelude = ByteReader::new(data);
1101 prelude.read_bytes(2)?;
1102 read_uleb128(&mut prelude)?;
1103 let logical = usize::try_from(read_uleb128(&mut prelude)?)
1104 .map_err(|_| TelemetryError::Unpack("payload too large"))?;
1105 if logical > max_decoded_bytes || logical > isize::MAX as usize - header.len() - CRC32_BYTES {
1106 if !compressed {
1107 return Ok(None);
1108 }
1109 return Err(TelemetryError::Io("payload exceeds routing memory budget"));
1110 }
1111 let bytes = logical
1112 .saturating_mul(2)
1113 .saturating_add(header.len().saturating_mul(2))
1114 .saturating_add(64);
1115 let largest = logical
1116 .saturating_add(header.len())
1117 .saturating_add(CRC32_BYTES)
1118 .saturating_add(32);
1119 if let Err(error) = crate::memory_admission::check(bytes, largest) {
1120 if !compressed {
1123 return Ok(None);
1124 }
1125 return Err(error);
1126 }
1127 let (new_flag, new_payload) = if compressed {
1128 (
1129 flags & !FLAG_COMPRESSED_PAYLOAD,
1130 alloc::borrow::Cow::Owned(payload_compression::decompress(payload, logical)?),
1131 )
1132 } else {
1133 if payload.len() != logical {
1134 return Err(TelemetryError::Unpack("payload length mismatch"));
1135 }
1136 let (beneficial, value) = payload_compression::compress_if_beneficial(payload);
1137 if !beneficial {
1138 return Ok(None);
1139 }
1140 (flags | FLAG_COMPRESSED_PAYLOAD, value)
1141 };
1142 let mut prefix = header.to_vec();
1143 prefix[0] = new_flag;
1144 Ok(Some(join_frame_with_crc(&prefix, &new_payload)))
1145}
1146
1147#[cfg(feature = "compression")]
1151pub(crate) fn decoded_frame_upper_bound(raw: &[u8]) -> TelemetryResult<usize> {
1152 let mut reader = ByteReader::new(raw);
1153 reader.read_bytes(2)?;
1154 read_uleb128(&mut reader)?;
1155 let logical = usize::try_from(read_uleb128(&mut reader)?)
1156 .map_err(|_| TelemetryError::Unpack("payload too large"))?;
1157 raw.len()
1158 .checked_add(logical)
1159 .ok_or(TelemetryError::Io("memory pressure"))
1160}
1161
1162#[cfg(feature = "compression")]
1166pub(crate) fn payload_encoding_for_retention(
1167 raw: &[u8],
1168 budget: usize,
1169) -> TelemetryResult<Option<Arc<[u8]>>> {
1170 if raw.first().is_some_and(|flags| flags & 0x10 != 0) {
1171 return Ok(None);
1172 }
1173 payload_encoding_for_side(raw, false, budget)
1174}
1175
1176pub(crate) fn with_plain_packet_parts<T>(
1179 pkt: &Packet,
1180 consume: impl FnOnce(&[u8], &[u8]) -> TelemetryResult<T>,
1181) -> TelemetryResult<T> {
1182 let reliable = is_reliable_type(pkt.data_type()).then_some(ReliableHeader {
1183 flags: RELIABLE_FLAG_UNSEQUENCED,
1184 seq: 0,
1185 ack: 0,
1186 });
1187 with_packet_frame_parts(
1188 pkt,
1189 reliable,
1190 pkt.wire_shape(),
1191 pkt.wire_target_senders(),
1192 None,
1193 true,
1194 consume,
1195 )
1196}
1197
1198fn packet_endpoint_bitmap_present(
1199 pkt: &Packet,
1200 shape: Option<MessageElement>,
1201 target_senders: &[u64],
1202) -> bool {
1203 let carries_wire_contract = shape.is_some() || !target_senders.is_empty();
1204 let endpoints_are_schema_default = endpoints_match_schema(pkt.data_type(), pkt.endpoints());
1205 let fixed_control_endpoints = matches!(
1210 pkt.data_type(),
1211 DataType::ReliableAck | DataType::ReliablePartialAck | DataType::ReliablePacketRequest
1212 );
1213 pkt.data_type().0 > DataType::P2pMessage.0
1216 || !endpoints_are_schema_default
1217 || (carries_wire_contract && !fixed_control_endpoints)
1218}
1219
1220fn with_packet_frame_parts<T>(
1221 pkt: &Packet,
1222 reliable: Option<ReliableHeader>,
1223 shape: Option<MessageElement>,
1224 target_senders: &[u64],
1225 #[cfg_attr(not(feature = "cryptography"), allow(unused_variables))] e2e: Option<E2eSealConfig>,
1226 compression_allowed: bool,
1227 consume: impl FnOnce(&[u8], &[u8]) -> TelemetryResult<T>,
1228) -> TelemetryResult<T> {
1229 let carries_wire_contract = shape.is_some() || !target_senders.is_empty();
1230 let endpoint_bitmap_present = packet_endpoint_bitmap_present(pkt, shape, target_senders);
1231 let endpoint_list_present = endpoint_bitmap_present && explicit_endpoint_list(pkt.endpoints());
1232 let nep_unique = unique_sorted_endpoints(pkt.endpoints()).count();
1233 let endpoint_bytes = if endpoint_list_present {
1234 endpoint_list_size(pkt.endpoints())
1235 } else if endpoint_bitmap_present {
1236 EP_BITMAP_BYTES
1237 } else {
1238 0
1239 };
1240
1241 let source_address = source_address_for_sender(pkt.sender());
1242
1243 let (payload_compressed, payload_wire) = payload_for_encoding(pkt, compression_allowed);
1245
1246 let reliable_is_compact = reliable.is_some_and(should_compact_reliable_header);
1250 let reliable_len = if let Some(hdr) = reliable {
1251 reliable_wire_size(hdr, reliable_is_compact)
1252 } else {
1253 0
1254 };
1255 let contract_bytes = wire_contract_size(shape, target_senders);
1256 let contract_len = if carries_wire_contract {
1257 uleb128_size(contract_bytes as u64) + contract_bytes
1258 } else {
1259 0
1260 };
1261 let capacity = header_size_bytes(pkt)
1262 + endpoint_bytes
1263 + contract_len
1264 + reliable_len
1265 + payload_wire.len()
1266 + CRC32_BYTES;
1267 #[cfg(feature = "cryptography")]
1268 let capacity = capacity
1269 + e2e.map_or(0, |seal| {
1270 uleb128_size(u64::from(seal.key_id))
1271 + uleb128_size(payload_wire.len() as u64)
1272 + uleb128_size(E2E_NONCE_LEN as u64)
1273 + E2E_NONCE_LEN
1274 + uleb128_size(E2E_TAG_CAP as u64)
1275 + E2E_TAG_CAP
1276 });
1277 #[cfg(feature = "cryptography")]
1281 let encrypted = e2e.is_some();
1282 #[cfg(not(feature = "cryptography"))]
1283 let encrypted = false;
1284 let scratch_capacity = if encrypted {
1285 capacity
1286 } else {
1287 capacity - payload_wire.len() - CRC32_BYTES
1288 };
1289 let mut out = Vec::with_capacity(scratch_capacity);
1290 #[cfg(debug_assertions)]
1291 let initial_capacity = out.capacity();
1292
1293 let mut flags: u8 = 0;
1295 if payload_compressed {
1296 flags |= FLAG_COMPRESSED_PAYLOAD;
1297 }
1298 if carries_wire_contract {
1299 flags |= FLAG_WIRE_CONTRACT;
1300 }
1301 if pkt.nonce() != 0 {
1302 flags |= FLAG_PACKET_NONCE;
1303 }
1304 if endpoint_list_present {
1305 flags |= FLAG_ENDPOINT_LIST_PRESENT;
1306 } else if endpoint_bitmap_present {
1307 flags |= FLAG_ENDPOINT_BITMAP_PRESENT;
1308 }
1309 if reliable.is_some() {
1310 flags |= FLAG_RELIABLE_HEADER_PRESENT;
1311 }
1312 if reliable_is_compact {
1313 flags |= FLAG_COMPACT_RELIABLE_HEADER;
1314 }
1315 #[cfg(feature = "cryptography")]
1316 if e2e.is_some() {
1317 flags |= FLAG_E2E_ENCRYPTED_PAYLOAD;
1318 }
1319 out.push(flags);
1320
1321 assert!(
1322 nep_unique <= u8::MAX as usize,
1323 "too many endpoints selected to fit in NEP u8"
1324 );
1325 out.push(nep_unique as u8);
1326
1327 write_uleb128(pkt.data_type().as_u32() as u64, &mut out);
1329 write_uleb128(pkt.data_size() as u64, &mut out);
1330 write_uleb128(pkt.timestamp(), &mut out);
1331 if pkt.nonce() != 0 {
1332 write_uleb128(pkt.nonce() as u64, &mut out);
1333 }
1334
1335 write_uleb128(source_address as u64, &mut out);
1336
1337 if endpoint_list_present {
1338 for endpoint in unique_sorted_endpoints(pkt.endpoints()) {
1339 write_uleb128(u64::from(endpoint.0), &mut out);
1340 }
1341 } else if endpoint_bitmap_present {
1342 out.extend_from_slice(&build_endpoint_bitmap(pkt.endpoints()));
1343 }
1344 if (flags & FLAG_WIRE_CONTRACT) != 0 {
1345 write_uleb128(contract_bytes as u64, &mut out);
1349 write_wire_contract(shape, target_senders, reliable.is_some(), &mut out)?;
1350 }
1351 if let Some(hdr) = reliable {
1352 write_reliable_header_encoded(hdr, reliable_is_compact, &mut out);
1353 }
1354 #[cfg(feature = "cryptography")]
1355 if let Some(e2e) = e2e {
1356 write_encrypted_payload(pkt, e2e.key_id, &payload_wire, &mut out)?;
1357 return consume(&out, &[]);
1358 }
1359 #[cfg(debug_assertions)]
1360 debug_assert_eq!(
1361 out.capacity(),
1362 initial_capacity,
1363 "packet header grew its buffer"
1364 );
1365 consume(&out, &payload_wire)
1366}
1367
1368pub fn unpack_packet(buf: &[u8]) -> Result<Packet, TelemetryError> {
1394 unpack_packet_with_role(buf, false)
1395}
1396
1397pub(crate) fn unpack_packet_with_role(buf: &[u8], managed: bool) -> Result<Packet, TelemetryError> {
1398 let data = verify_crc32(buf)?;
1399 if data.is_empty() {
1400 return Err(TelemetryError::Unpack("short prelude"));
1401 }
1402 let mut r = ByteReader::new(data);
1403
1404 let flags = r.read_bytes(1)?[0];
1405 let payload_is_compressed = (flags & FLAG_COMPRESSED_PAYLOAD) != 0;
1406 let endpoint_bitmap_present = (flags & FLAG_ENDPOINT_BITMAP_PRESENT) != 0;
1407 let compact_reliable_header = (flags & FLAG_COMPACT_RELIABLE_HEADER) != 0;
1408 #[cfg(feature = "cryptography")]
1409 let payload_is_encrypted = (flags & FLAG_E2E_ENCRYPTED_PAYLOAD) != 0;
1410 #[cfg(not(feature = "cryptography"))]
1411 if (flags & 0x10) != 0 {
1412 return Err(TelemetryError::Unpack("e2e crypto unsupported"));
1413 }
1414
1415 let nep = r.read_bytes(1)?[0] as usize;
1416
1417 let ty_v = read_uleb128(&mut r)?;
1418 let dsz = read_uleb128(&mut r)? as usize; let ts_v = read_uleb128(&mut r)?;
1420 let nonce = if (flags & FLAG_PACKET_NONCE) != 0 {
1421 u16::try_from(read_uleb128(&mut r)?)
1422 .map_err(|_| TelemetryError::Unpack("packet nonce too large"))?
1423 } else {
1424 0
1425 };
1426 let source_address = u32::try_from(read_uleb128(&mut r)?)
1427 .map_err(|_| TelemetryError::Unpack("source address too large"))?;
1428 let sender_str = sender_name_for_address(source_address);
1429 let ty_u32 = data_type_id_from_wire(ty_v)?;
1430 let known_ty = DataType::try_from_u32(ty_u32);
1431 let endpoint_bytes = if endpoint_bitmap_present {
1432 EP_BITMAP_BYTES
1433 } else {
1434 0
1435 };
1436
1437 if !payload_is_compressed {
1440 if r.remaining() < endpoint_bytes + dsz {
1441 return Err(TelemetryError::Unpack("short buffer"));
1442 }
1443 } else if r.remaining() < endpoint_bytes + 1 {
1444 return Err(TelemetryError::Unpack("short buffer"));
1445 }
1446
1447 let eps = endpoints_from_wire_or_schema(
1448 &mut r,
1449 endpoint_bitmap_present,
1450 flags & FLAG_ENDPOINT_LIST_PRESENT != 0,
1451 known_ty,
1452 nep,
1453 )?;
1454
1455 let contract = decode_wire_contract(&mut r, (flags & FLAG_WIRE_CONTRACT) != 0)?;
1456 let ty = known_ty
1457 .or_else(|| contract.shape.map(|_| DataType(ty_u32)))
1458 .ok_or(TelemetryError::InvalidType)?;
1459
1460 let mut reliable_hdr: Option<ReliableHeader> = None;
1462 if flags & FLAG_RELIABLE_HEADER_PRESENT != 0 || contract.has_reliable_header {
1463 let hdr = read_reliable_header_encoded(&mut r, compact_reliable_header)?;
1464 if (hdr.flags & RELIABLE_FLAG_ACK_ONLY) != 0 {
1465 return Err(TelemetryError::Unpack("reliable control frame"));
1466 }
1467 reliable_hdr = Some(hdr);
1468 }
1469
1470 let payload_copies = if payload_is_compressed { 3 } else { 1 };
1474 #[cfg(feature = "cryptography")]
1475 let payload_copies = payload_copies + usize::from(payload_is_encrypted);
1476 let small_ack = matches!(ty, DataType::ReliableAck | DataType::ReliablePartialAck)
1477 && dsz <= 32
1478 && payload_copies == 1;
1479 if crate::memory_admission::supports_work_kinds()
1480 && !small_ack
1481 && buf.len() <= 128
1482 && flags & 0x11 == 0
1483 && (crate::memory_admission::protected_discovery_request_shape(
1484 ty,
1485 dsz == 0 && r.remaining() == 0,
1486 eps.as_ref() == [DataEndpoint::Discovery],
1487 contract
1488 .target_senders
1489 .as_ref()
1490 .map_or(0, |owners| owners.len()),
1491 contract.shape,
1492 ) || crate::memory_admission::protected_control_shape(
1493 ty,
1494 nep,
1495 contract
1496 .target_senders
1497 .as_ref()
1498 .map_or(0, |owners| owners.len()),
1499 contract.shape,
1500 reliable_hdr.is_some(),
1501 managed,
1502 ))
1503 {
1504 crate::memory_admission::check_small_protected_decode(buf.len())?;
1505 } else {
1506 crate::memory_admission::check(
1507 if small_ack {
1508 512
1509 } else {
1510 dsz.saturating_mul(payload_copies).saturating_add(512)
1511 },
1512 dsz.saturating_add(32),
1513 )?;
1514 }
1515 let payload: StandardSmallPayload = {
1517 #[cfg(feature = "cryptography")]
1518 let payload_wire_owned;
1519 #[cfg(feature = "cryptography")]
1520 let payload_wire: &[u8] = if payload_is_encrypted {
1521 let aad_end = r.off;
1522 payload_wire_owned = read_encrypted_payload(&mut r, &data[..aad_end], dsz)?;
1523 &payload_wire_owned
1524 } else if !payload_is_compressed {
1525 r.read_bytes(dsz)?
1526 } else {
1527 let comp_len = r.remaining();
1528 r.read_bytes(comp_len)?
1529 };
1530
1531 #[cfg(not(feature = "cryptography"))]
1532 let payload_wire: &[u8] = if !payload_is_compressed {
1533 r.read_bytes(dsz)?
1534 } else {
1535 let comp_len = r.remaining();
1536 r.read_bytes(comp_len)?
1537 };
1538
1539 if payload_is_compressed {
1540 let decompressed = payload_compression::decompress(payload_wire, dsz)?;
1541 crate::memory_admission::check(dsz.saturating_add(512), dsz.saturating_add(32))?;
1542 StandardSmallPayload::new(&decompressed)
1543 } else {
1544 if payload_wire.len() != dsz {
1545 return Err(TelemetryError::Unpack("payload length mismatch"));
1546 }
1547 StandardSmallPayload::new(payload_wire)
1548 }
1549 };
1550
1551 let _ = reliable_hdr;
1555 Packet::new_with_owned_wire_contract(
1556 ty,
1557 eps,
1558 sender_str,
1559 ts_v,
1560 nonce,
1561 payload,
1562 contract.shape,
1563 contract.target_senders,
1564 )
1565}
1566
1567pub fn peek_envelope(buf: &[u8]) -> TelemetryResult<TelemetryEnvelope> {
1588 Ok(peek_routing_envelope(buf)?.into_owned())
1589}
1590
1591#[derive(Clone, Debug)]
1595pub(crate) struct RoutingEnvelope<'a> {
1596 pub ty: DataType,
1597 bitmap: [u64; EP_BITMAP_BYTES / 8],
1598 endpoint_list: Option<&'a [u8]>,
1599 #[cfg(feature = "std")]
1600 schema_endpoints: Option<Arc<[DataEndpoint]>>,
1601 #[cfg(not(feature = "std"))]
1602 schema_endpoints: Option<&'static [DataEndpoint]>,
1603 pub source_address: u32,
1604 pub timestamp_ms: u64,
1605 pub wire_shape: Option<MessageElement>,
1606 pub is_reliable: bool,
1607 payload_is_empty: bool,
1608 targets: &'a [u8],
1609 has_reliable_header: bool,
1610}
1611#[derive(Clone)]
1612pub(crate) struct RoutingEndpointIter<'a> {
1613 envelope: &'a RoutingEnvelope<'a>,
1614 word: usize,
1615 bits: u64,
1616 list_offset: usize,
1617 previous: Option<DataEndpoint>,
1618 finished: bool,
1619}
1620impl Iterator for RoutingEndpointIter<'_> {
1621 type Item = DataEndpoint;
1622 #[inline(never)]
1623 fn next(&mut self) -> Option<DataEndpoint> {
1624 if self.finished {
1625 return None;
1626 }
1627 if let Some(bytes) = self.envelope.endpoint_list {
1628 let mut reader = ByteReader {
1629 buf: bytes,
1630 off: self.list_offset,
1631 };
1632 if reader.remaining() == 0 {
1633 self.finished = true;
1634 return None;
1635 }
1636 let endpoint = read_uleb128(&mut reader)
1637 .ok()
1638 .and_then(|id| u32::try_from(id).ok())
1639 .map(DataEndpoint);
1640 self.list_offset = reader.off;
1641 return endpoint;
1642 }
1643 if let Some(endpoints) = &self.envelope.schema_endpoints {
1644 let next = endpoints
1645 .iter()
1646 .copied()
1647 .filter(|ep| self.previous.is_none_or(|previous| *ep > previous))
1648 .min();
1649 self.previous = next;
1650 self.finished = next.is_none();
1651 return next;
1652 }
1653 while self.bits == 0 {
1654 self.word += 1;
1655 if self.word == self.envelope.bitmap.len() {
1656 self.finished = true;
1657 return None;
1658 }
1659 self.bits = self.envelope.bitmap[self.word];
1660 }
1661 let bit = self.bits.trailing_zeros();
1662 self.bits &= self.bits - 1;
1663 Some(DataEndpoint((self.word * 64 + bit as usize) as u32))
1664 }
1665}
1666
1667impl RoutingEnvelope<'_> {
1668 pub fn endpoints(&self) -> RoutingEndpointIter<'_> {
1669 RoutingEndpointIter {
1670 envelope: self,
1671 word: 0,
1672 bits: self.bitmap[0],
1673 list_offset: 0,
1674 previous: None,
1675 finished: false,
1676 }
1677 }
1678 pub fn target_senders(&self) -> impl ExactSizeIterator<Item = u64> + '_ {
1679 self.targets
1680 .as_chunks::<8>()
1681 .0
1682 .iter()
1683 .map(|bytes| u64::from_le_bytes(*bytes))
1684 }
1685 fn into_owned(self) -> TelemetryEnvelope {
1686 TelemetryEnvelope {
1687 ty: self.ty,
1688 endpoints: self.endpoints().collect::<Arc<[_]>>(),
1689 sender: sender_name_for_address(self.source_address),
1690 source_address: self.source_address,
1691 timestamp_ms: self.timestamp_ms,
1692 wire_shape: self.wire_shape,
1693 target_senders: self.target_senders().collect::<Arc<[_]>>(),
1694 }
1695 }
1696}
1697
1698fn read_routing_envelope<'a>(r: &mut ByteReader<'a>) -> TelemetryResult<RoutingEnvelope<'a>> {
1699 let flags = r.read_bytes(1)?[0];
1700 let nep = r.read_bytes(1)?[0] as usize;
1701 let ty_v = read_uleb128(r)?;
1702 let payload_is_empty = read_uleb128(r)? == 0;
1703 let timestamp_ms = read_uleb128(r)?;
1704 if flags & FLAG_PACKET_NONCE != 0 {
1705 let _ = read_uleb128(r)?;
1706 }
1707 let source_address = u32::try_from(read_uleb128(r)?)
1708 .map_err(|_| TelemetryError::Unpack("source address too large"))?;
1709 let ty_u32 = data_type_id_from_wire(ty_v)?;
1710 let is_reliable = flags & FLAG_RELIABLE_HEADER_PRESENT != 0;
1711 let mut bitmap = [0u64; EP_BITMAP_BYTES / 8];
1714 let mut endpoint_list = None;
1715 let mut schema_endpoints = None;
1716 if flags & FLAG_ENDPOINT_LIST_PRESENT != 0 {
1717 if flags & FLAG_ENDPOINT_BITMAP_PRESENT != 0 {
1718 return Err(TelemetryError::Unpack("ambiguous endpoint encoding"));
1719 }
1720 let bytes = read_endpoint_list(r, nep)?;
1721 endpoint_list = Some(bytes);
1722 } else if flags & FLAG_ENDPOINT_BITMAP_PRESENT != 0 {
1723 let bytes = r.read_bytes(EP_BITMAP_BYTES)?;
1724 for (word, chunk) in bitmap.iter_mut().zip(bytes.as_chunks::<8>().0) {
1725 *word = u64::from_le_bytes(*chunk);
1726 }
1727 if bitmap
1728 .iter()
1729 .map(|word| word.count_ones() as usize)
1730 .sum::<usize>()
1731 != nep
1732 {
1733 return Err(TelemetryError::Unpack("endpoint count mismatch"));
1734 }
1735 for (index, &word) in bitmap.iter().enumerate() {
1736 let mut bits = word;
1737 while bits != 0 {
1738 let id = index * 64 + bits.trailing_zeros() as usize;
1739 if !crate::config::endpoint_exists(DataEndpoint(id as u32)) {
1740 return Err(TelemetryError::Unpack("bad endpoint bit set"));
1741 }
1742 bits &= bits - 1;
1743 }
1744 }
1745 } else {
1746 let meta = crate::config::try_get_message_meta(DataType(ty_u32))
1747 .ok_or(TelemetryError::InvalidType)?;
1748 if unique_sorted_endpoints(meta.endpoints_ref()).count() != nep {
1749 return Err(TelemetryError::Unpack("endpoint count mismatch"));
1750 }
1751 schema_endpoints = Some(meta.endpoints);
1752 }
1753 let mut shape = None;
1754 let mut targets: &'a [u8] = &[];
1755 let mut has_reliable_header = false;
1756 if flags & FLAG_WIRE_CONTRACT != 0 {
1757 let len = usize::try_from(read_uleb128(r)?)
1758 .map_err(|_| TelemetryError::Unpack("wire contract length"))?;
1759 let mut cr = ByteReader::new(r.read_bytes(len)?);
1760 let contract_flags = cr.read_bytes(1)?[0];
1761 if contract_flags & CONTRACT_FLAG_SHAPE != 0 {
1762 shape = Some(decode_wire_shape(&mut cr)?);
1763 }
1764 if contract_flags & CONTRACT_FLAG_TARGETS != 0 {
1765 let count = usize::try_from(read_uleb128(&mut cr)?)
1766 .map_err(|_| TelemetryError::Unpack("wire contract target count"))?;
1767 let bytes = count
1768 .checked_mul(8)
1769 .ok_or(TelemetryError::Unpack("wire contract target count"))?;
1770 targets = cr.read_bytes(bytes)?;
1771 }
1772 if cr.remaining() != 0 {
1773 return Err(TelemetryError::Unpack("wire contract trailing bytes"));
1774 }
1775 has_reliable_header = contract_flags & CONTRACT_FLAG_RELIABLE_HEADER != 0;
1776 }
1777 if nep == 0 {
1778 if !is_reliable && !has_reliable_header {
1779 return Err(TelemetryError::EmptyEndpoints);
1780 }
1781 let mut header = ByteReader {
1782 buf: r.buf,
1783 off: r.off,
1784 };
1785 let reliable =
1786 read_reliable_header_encoded(&mut header, flags & FLAG_COMPACT_RELIABLE_HEADER != 0)?;
1787 if reliable.flags & RELIABLE_FLAG_ACK_ONLY == 0 {
1788 return Err(TelemetryError::EmptyEndpoints);
1789 }
1790 }
1791 let ty = DataType(ty_u32);
1792 Ok(RoutingEnvelope {
1793 ty,
1794 bitmap,
1795 endpoint_list,
1796 schema_endpoints,
1797 source_address,
1798 timestamp_ms,
1799 wire_shape: shape,
1800 is_reliable,
1801 payload_is_empty,
1802 targets,
1803 has_reliable_header,
1804 })
1805}
1806
1807pub(crate) fn scheduling_type(buf: &[u8]) -> Option<DataType> {
1811 let mut reader = ByteReader::new(buf.get(..buf.len().min(12))?);
1812 reader.read_bytes(2).ok()?;
1813 let id = data_type_id_from_wire(read_uleb128(&mut reader).ok()?).ok()?;
1814 Some(DataType(id))
1818}
1819
1820#[cfg(feature = "discovery")]
1824pub(crate) fn chunked_schema_source(prefix: &[u8]) -> Option<u32> {
1825 let envelope = read_routing_envelope(&mut ByteReader::new(prefix)).ok()?;
1826 (envelope.ty == DataType::DiscoverySchema).then_some(envelope.source_address)
1827}
1828
1829pub(crate) fn peek_routing_envelope(buf: &[u8]) -> TelemetryResult<RoutingEnvelope<'_>> {
1830 let data = verify_crc32(buf)?;
1831 read_routing_envelope(&mut ByteReader::new(data))
1832}
1833
1834pub struct TelemetryFrameInfo {
1836 pub envelope: TelemetryEnvelope,
1837 pub reliable: Option<ReliableHeader>,
1838}
1839
1840impl TelemetryFrameInfo {
1841 #[inline]
1842 pub fn ack_only(&self) -> bool {
1844 self.reliable
1845 .map(|h| (h.flags & RELIABLE_FLAG_ACK_ONLY) != 0)
1846 .unwrap_or(false)
1847 }
1848}
1849
1850#[cfg(feature = "discovery")]
1853#[derive(Clone, Copy)]
1854pub(crate) enum TargetSenderSet<'a> {
1855 Hashes(&'a [u64]),
1856 Encoded(&'a [u8]),
1857}
1858#[cfg(feature = "discovery")]
1859impl<'a> TargetSenderSet<'a> {
1860 pub(crate) fn skip(self, count: usize) -> Self {
1861 match self {
1862 Self::Hashes(v) => Self::Hashes(v.get(count..).unwrap_or(&[])),
1863 Self::Encoded(v) => Self::Encoded(v.get(count.saturating_mul(8)..).unwrap_or(&[])),
1864 }
1865 }
1866 pub(crate) fn iter(self) -> impl Iterator<Item = u64> + 'a {
1867 let (hashes, encoded) = match self {
1868 Self::Hashes(v) => (v, &[][..]),
1869 Self::Encoded(v) => (&[][..], v),
1870 };
1871 hashes.iter().copied().chain(
1872 encoded
1873 .as_chunks::<8>()
1874 .0
1875 .iter()
1876 .map(|v| u64::from_le_bytes(*v)),
1877 )
1878 }
1879 pub(crate) fn is_empty(&self) -> bool {
1880 match self {
1881 Self::Hashes(v) => v.is_empty(),
1882 Self::Encoded(v) => v.is_empty(),
1883 }
1884 }
1885 pub(crate) fn contains(&self, target: u64) -> bool {
1886 match self {
1887 Self::Hashes(v) => v.contains(&target),
1888 Self::Encoded(v) => v
1889 .as_chunks::<8>()
1890 .0
1891 .iter()
1892 .any(|bytes| u64::from_le_bytes(*bytes) == target),
1893 }
1894 }
1895}
1896#[cfg(feature = "discovery")]
1897impl<'a> RoutingEnvelope<'a> {
1898 pub(crate) fn target_sender_set(&self) -> TargetSenderSet<'a> {
1899 TargetSenderSet::Encoded(self.targets)
1900 }
1901}
1902
1903pub(crate) struct RoutingFrameInfo<'a> {
1904 pub envelope: RoutingEnvelope<'a>,
1905 pub reliable: Option<ReliableHeader>,
1906 pub empty_payload: bool,
1907}
1908impl RoutingFrameInfo<'_> {
1909 pub fn ack_only(&self) -> bool {
1910 self.reliable
1911 .is_some_and(|header| header.flags & RELIABLE_FLAG_ACK_ONLY != 0)
1912 }
1913}
1914#[cfg(feature = "discovery")]
1917pub(crate) fn peek_plain_discovery_payload(
1918 buf: &[u8],
1919 expected: DataType,
1920) -> TelemetryResult<Option<(u32, &[u8])>> {
1921 let mut prelude = ByteReader::new(buf);
1924 prelude.read_bytes(2)?;
1925 if data_type_id_from_wire(read_uleb128(&mut prelude)?)? != expected.as_u32() {
1926 return Ok(None);
1927 }
1928 let data = verify_crc32(buf)?;
1929 let mut reader = ByteReader::new(data);
1930 let envelope = read_routing_envelope(&mut reader)?;
1931 if envelope.ty != expected
1932 || data[0] & (FLAG_COMPRESSED_PAYLOAD | 0x10) != 0
1933 || envelope.target_senders().len() != 0
1934 || envelope
1935 .wire_shape
1936 .is_some_and(|shape| shape != crate::message_meta(expected).element)
1937 {
1938 return Ok(None);
1939 }
1940 if envelope.is_reliable || envelope.has_reliable_header {
1941 let reliable =
1942 read_reliable_header_encoded(&mut reader, data[0] & FLAG_COMPACT_RELIABLE_HEADER != 0)?;
1943 if reliable.flags != RELIABLE_FLAG_UNSEQUENCED || reliable.seq != 0 || reliable.ack != 0 {
1946 return Ok(None);
1947 }
1948 }
1949 let mut prelude = ByteReader::new(data);
1952 prelude.read_bytes(2)?;
1953 read_uleb128(&mut prelude)?;
1954 let size = usize::try_from(read_uleb128(&mut prelude)?)
1955 .map_err(|_| TelemetryError::Unpack("discovery payload too large"))?;
1956 if reader.remaining() != size {
1957 return Err(TelemetryError::Unpack("discovery payload length mismatch"));
1958 }
1959 Ok(Some((envelope.source_address, reader.read_bytes(size)?)))
1960}
1961
1962#[cfg(feature = "latest-value-egress")]
1965pub(crate) fn peek_latest_plain_frame_info(buf: &[u8]) -> TelemetryResult<RoutingFrameInfo<'_>> {
1966 if buf.first().is_none_or(|flags| flags & 0x17 != 0) {
1967 return Err(TelemetryError::BadArg);
1968 }
1969 let data = verify_crc32(buf)?;
1970 let mut reader = ByteReader::new(data);
1971 let envelope = read_routing_envelope(&mut reader)?;
1972 if envelope.has_reliable_header || envelope.endpoints().next().is_none() {
1973 return Err(TelemetryError::BadArg);
1974 }
1975 let mut prelude = ByteReader::new(data);
1976 prelude.read_bytes(2)?;
1977 read_uleb128(&mut prelude)?;
1978 let size = usize::try_from(read_uleb128(&mut prelude)?).map_err(|_| TelemetryError::BadArg)?;
1979 if reader.remaining() != size {
1980 return Err(TelemetryError::SizeMismatch {
1981 expected: size,
1982 got: reader.remaining(),
1983 });
1984 }
1985 let meta =
1986 crate::config::try_get_message_meta(envelope.ty).ok_or(TelemetryError::InvalidType)?;
1987 Packet::validate_payload_against_element(meta.element, reader.read_bytes(size)?)?;
1988 Ok(RoutingFrameInfo {
1989 envelope,
1990 reliable: None,
1991 empty_payload: size == 0,
1992 })
1993}
1994
1995pub(crate) fn peek_routing_frame_info(buf: &[u8]) -> TelemetryResult<RoutingFrameInfo<'_>> {
1996 let data = verify_crc32(buf)?;
1997 let mut reader = ByteReader::new(data);
1998 let envelope = read_routing_envelope(&mut reader)?;
1999 let reliable = if envelope.is_reliable || envelope.has_reliable_header {
2000 Some(read_reliable_header_encoded(
2001 &mut reader,
2002 data[0] & FLAG_COMPACT_RELIABLE_HEADER != 0,
2003 )?)
2004 } else {
2005 None
2006 };
2007 let empty_payload = envelope.payload_is_empty && reader.remaining() == 0;
2008 Ok(RoutingFrameInfo {
2009 envelope,
2010 reliable,
2011 empty_payload,
2012 })
2013}
2014
2015fn peek_frame_info_inner(buf: &[u8]) -> TelemetryResult<TelemetryFrameInfo> {
2016 let mut r = ByteReader::new(buf);
2017 let header = read_routing_envelope(&mut r)?;
2018 let reliable = if header.is_reliable || header.has_reliable_header {
2019 Some(read_reliable_header_encoded(
2020 &mut r,
2021 buf[0] & FLAG_COMPACT_RELIABLE_HEADER != 0,
2022 )?)
2023 } else {
2024 None
2025 };
2026 Ok(TelemetryFrameInfo {
2027 envelope: header.into_owned(),
2028 reliable,
2029 })
2030}
2031
2032pub fn peek_frame_info(buf: &[u8]) -> TelemetryResult<TelemetryFrameInfo> {
2038 let data = verify_crc32(buf)?;
2039 peek_frame_info_inner(data)
2040}
2041
2042pub fn peek_frame_info_unchecked(buf: &[u8]) -> TelemetryResult<TelemetryFrameInfo> {
2048 let (data, _crc) = split_crc32(buf)?;
2049 peek_frame_info_inner(data)
2050}
2051
2052pub fn reliable_header_offset(buf: &[u8]) -> TelemetryResult<Option<usize>> {
2067 Ok(reliable_header_span(buf)?.map(|(off, _, _)| off))
2068}
2069
2070pub(crate) fn reliable_header_span(
2071 buf: &[u8],
2072) -> TelemetryResult<Option<(usize, usize, ReliableHeader)>> {
2073 if buf.len() < CRC32_BYTES + 1 {
2074 return Err(TelemetryError::Unpack("short prelude"));
2075 }
2076 let data_len = buf.len().saturating_sub(CRC32_BYTES);
2077 let mut r = ByteReader::new(&buf[..data_len]);
2078
2079 let flags = r.read_bytes(1)?[0];
2080 let endpoint_bitmap_present = (flags & FLAG_ENDPOINT_BITMAP_PRESENT) != 0;
2081 let compact_reliable_header = (flags & FLAG_COMPACT_RELIABLE_HEADER) != 0;
2082
2083 let nep = r.read_bytes(1)?[0] as usize;
2084
2085 let ty_v = read_uleb128(&mut r)?;
2086 let _dsz = read_uleb128(&mut r)? as usize;
2087 let _ts_v = read_uleb128(&mut r)?;
2088 if (flags & FLAG_PACKET_NONCE) != 0 {
2089 let _ = read_uleb128(&mut r)?;
2090 }
2091 let _source_address = u32::try_from(read_uleb128(&mut r)?)
2092 .map_err(|_| TelemetryError::Unpack("source address too large"))?;
2093 let endpoint_bytes = if endpoint_bitmap_present {
2094 EP_BITMAP_BYTES
2095 } else {
2096 0
2097 };
2098
2099 if r.remaining() < endpoint_bytes {
2100 return Err(TelemetryError::Unpack("short buffer"));
2101 }
2102
2103 if flags & FLAG_ENDPOINT_LIST_PRESENT != 0 {
2104 if endpoint_bitmap_present {
2105 return Err(TelemetryError::Unpack("ambiguous endpoint encoding"));
2106 }
2107 read_endpoint_list(&mut r, nep)?;
2108 } else if endpoint_bitmap_present {
2109 r.read_bytes(EP_BITMAP_BYTES)?;
2110 }
2111 let contract = decode_wire_contract(&mut r, (flags & FLAG_WIRE_CONTRACT) != 0)?;
2112 let ty_u32 = data_type_id_from_wire(ty_v)?;
2113 let _ty = DataType(ty_u32);
2114 if flags & FLAG_RELIABLE_HEADER_PRESENT == 0 && !contract.has_reliable_header {
2115 return Ok(None);
2116 }
2117
2118 let off = r.off;
2119 let hdr = read_reliable_header_encoded(&mut r, compact_reliable_header)?;
2120 Ok(Some((off, r.off - off, hdr)))
2121}
2122
2123pub fn rewrite_reliable_header(
2139 buf: &mut [u8],
2140 flags: u8,
2141 seq: u32,
2142 ack: u32,
2143) -> TelemetryResult<bool> {
2144 let Some((off, old_len, _)) = reliable_header_span(buf)? else {
2145 return Ok(false);
2146 };
2147 let hdr = ReliableHeader { flags, seq, ack };
2148 let compact = should_compact_reliable_header(hdr);
2149 if reliable_wire_size(hdr, compact) != old_len {
2150 return Err(TelemetryError::Unpack(
2151 "reliable header rewrite changes wire size",
2152 ));
2153 }
2154 let data_len = buf.len().saturating_sub(CRC32_BYTES);
2155 if data_len.saturating_sub(off) < old_len {
2156 return Err(TelemetryError::Unpack("short buffer"));
2157 }
2158 if compact {
2159 buf[0] |= FLAG_COMPACT_RELIABLE_HEADER;
2160 } else {
2161 buf[0] &= !FLAG_COMPACT_RELIABLE_HEADER;
2162 }
2163 let mut encoded = Vec::with_capacity(old_len);
2164 write_reliable_header_encoded(hdr, compact, &mut encoded);
2165 buf[off..off + old_len].copy_from_slice(&encoded);
2166 if buf.len() < CRC32_BYTES {
2167 return Err(TelemetryError::Unpack("short buffer"));
2168 }
2169 let crc = crc32_bytes(&buf[..data_len]);
2170 buf[data_len..data_len + CRC32_BYTES].copy_from_slice(&crc.to_le_bytes());
2171 Ok(true)
2172}
2173
2174pub(crate) fn rewrite_reliable_header_owned(
2175 buf: &[u8],
2176 flags: u8,
2177 seq: u32,
2178 ack: u32,
2179) -> TelemetryResult<Option<Arc<[u8]>>> {
2180 let Some((off, old_len, _)) = reliable_header_span(buf)? else {
2181 return Ok(None);
2182 };
2183 let data_len = buf.len().saturating_sub(CRC32_BYTES);
2184 if data_len < off + old_len {
2185 return Err(TelemetryError::Unpack("short buffer"));
2186 }
2187 let hdr = ReliableHeader { flags, seq, ack };
2188 let compact = should_compact_reliable_header(hdr);
2189 let mut encoded = Vec::with_capacity(reliable_wire_size(hdr, compact));
2190 write_reliable_header_encoded(hdr, compact, &mut encoded);
2191
2192 let mut out = Vec::with_capacity(data_len - old_len + encoded.len() + CRC32_BYTES);
2193 out.extend_from_slice(&buf[..off]);
2194 if compact {
2195 out[0] |= FLAG_COMPACT_RELIABLE_HEADER;
2196 } else {
2197 out[0] &= !FLAG_COMPACT_RELIABLE_HEADER;
2198 }
2199 out.extend_from_slice(&encoded);
2200 out.extend_from_slice(&buf[off + old_len..data_len]);
2201 let crc = crc32_bytes(&out);
2202 out.extend_from_slice(&crc.to_le_bytes());
2203 Ok(Some(Arc::from(out)))
2204}
2205
2206pub fn header_size_bytes(pkt: &Packet) -> usize {
2222 let prelude = 2; let source_address = sender_address_u32(pkt.sender());
2225
2226 prelude
2227 + uleb128_size(pkt.data_type().as_u32() as u64)
2228 + uleb128_size(pkt.data_size() as u64)
2229 + uleb128_size(pkt.timestamp())
2230 + if pkt.nonce() != 0 {
2231 uleb128_size(pkt.nonce() as u64)
2232 } else {
2233 0
2234 }
2235 + uleb128_size(source_address as u64)
2236}
2237
2238pub fn packet_wire_size(pkt: &Packet) -> usize {
2251 let shape = pkt.wire_shape();
2252 let targets = pkt.wire_target_senders();
2253 let contract_bytes = wire_contract_size(shape, targets);
2254 let contract_len = if shape.is_some() || !targets.is_empty() {
2255 uleb128_size(contract_bytes as u64) + contract_bytes
2256 } else {
2257 0
2258 };
2259 let endpoint_len = if packet_endpoint_bitmap_present(pkt, shape, targets) {
2260 if explicit_endpoint_list(pkt.endpoints()) {
2261 endpoint_list_size(pkt.endpoints())
2262 } else {
2263 EP_BITMAP_BYTES
2264 }
2265 } else {
2266 0
2267 };
2268 let (_, payload_wire) = payload_for_wire(pkt);
2269 let reliable_len = if is_reliable_type(pkt.data_type()) {
2270 let header = ReliableHeader {
2271 flags: RELIABLE_FLAG_UNSEQUENCED,
2272 seq: 0,
2273 ack: 0,
2274 };
2275 reliable_wire_size(header, should_compact_reliable_header(header))
2276 } else {
2277 0
2278 };
2279 header_size_bytes(pkt)
2282 + endpoint_len
2283 + contract_len
2284 + reliable_len
2285 + payload_wire.len()
2286 + CRC32_BYTES
2287}
2288
2289#[inline]
2290pub fn packet_id_from_wire(buf: &[u8]) -> Result<u64, TelemetryError> {
2292 let data = verify_crc32(buf)?;
2293 if data.len() < 2 {
2294 return Err(TelemetryError::Unpack("short prelude"));
2295 }
2296
2297 let mut r = ByteReader::new(data);
2298
2299 let flags = r.read_bytes(1)?[0];
2300 let payload_is_compressed = (flags & FLAG_COMPRESSED_PAYLOAD) != 0;
2301 let compact_reliable_header = (flags & FLAG_COMPACT_RELIABLE_HEADER) != 0;
2302 #[cfg(feature = "cryptography")]
2303 let payload_is_encrypted = (flags & FLAG_E2E_ENCRYPTED_PAYLOAD) != 0;
2304 #[cfg(not(feature = "cryptography"))]
2305 if (flags & 0x10) != 0 {
2306 return Err(TelemetryError::Unpack("e2e crypto unsupported"));
2307 }
2308
2309 let _nep = r.read_bytes(1)?[0] as usize;
2310
2311 let _ = read_uleb128(&mut r)?;
2312 let dsz = read_uleb128(&mut r)? as usize; let ts_v = read_uleb128(&mut r)?;
2314 let nonce = if (flags & FLAG_PACKET_NONCE) != 0 {
2315 u16::try_from(read_uleb128(&mut r)?)
2316 .map_err(|_| TelemetryError::Unpack("packet nonce too large"))?
2317 } else {
2318 0
2319 };
2320 let source_address = u32::try_from(read_uleb128(&mut r)?)
2321 .map_err(|_| TelemetryError::Unpack("source address too large"))?;
2322 let mut envelope_reader = ByteReader::new(data);
2325 let envelope = read_routing_envelope(&mut envelope_reader)?;
2326 r = envelope_reader;
2327 let ty = envelope.ty;
2328
2329 if envelope.is_reliable || envelope.has_reliable_header {
2331 let hdr = read_reliable_header_encoded(&mut r, compact_reliable_header)?;
2332 if (hdr.flags & RELIABLE_FLAG_ACK_ONLY) != 0 {
2333 return Err(TelemetryError::Unpack("reliable control frame"));
2334 }
2335 }
2336
2337 #[cfg(feature = "cryptography")]
2339 let payload_wire_owned;
2340 #[cfg(feature = "cryptography")]
2341 let payload_wire: &[u8] = if payload_is_encrypted {
2342 let aad_end = r.off;
2343 payload_wire_owned =
2344 read_encrypted_payload(&mut r, data.get(..aad_end).unwrap_or(&[]), dsz)?;
2345 &payload_wire_owned
2346 } else if !payload_is_compressed {
2347 if r.remaining() < dsz {
2348 return Err(TelemetryError::Unpack("short buffer"));
2349 }
2350 r.read_bytes(dsz)?
2351 } else {
2352 let comp_len = r.remaining();
2353 if comp_len < 1 {
2354 return Err(TelemetryError::Unpack("short buffer"));
2355 }
2356 r.read_bytes(comp_len)?
2357 };
2358 #[cfg(not(feature = "cryptography"))]
2359 let payload_wire: &[u8] = if !payload_is_compressed {
2360 if r.remaining() < dsz {
2361 return Err(TelemetryError::Unpack("short buffer"));
2362 }
2363 r.read_bytes(dsz)?
2364 } else {
2365 let comp_len = r.remaining();
2366 if comp_len < 1 {
2367 return Err(TelemetryError::Unpack("short buffer"));
2368 }
2369 r.read_bytes(comp_len)?
2370 };
2371 let payload_decompressed;
2372 let payload_bytes: &[u8] = if payload_is_compressed {
2373 payload_decompressed = payload_compression::decompress(payload_wire, dsz)?;
2374 &payload_decompressed
2375 } else {
2376 if payload_wire.len() != dsz {
2377 return Err(TelemetryError::Unpack("payload length mismatch"));
2378 }
2379 payload_wire
2380 };
2381
2382 let mut h: u64 = 0x9E37_79B9_7F4A_7C15;
2384
2385 h = hash_bytes_u64(h, &source_address.to_le_bytes());
2387
2388 h = hash_bytes_u64(h, &ty.0.to_le_bytes());
2390
2391 for ep in envelope.endpoints() {
2393 h = hash_bytes_u64(h, &ep.0.to_le_bytes());
2394 }
2395
2396 h = hash_bytes_u64(h, &ts_v.to_le_bytes());
2398 h = hash_bytes_u64(h, &nonce.to_le_bytes());
2399 h = hash_bytes_u64(h, &(dsz as u64).to_le_bytes());
2400
2401 h = hash_bytes_u64(h, payload_bytes);
2403 Ok(h)
2404}
2405
2406mod payload_compression {
2407 use crate::TelemetryError;
2408 use alloc::borrow::Cow;
2409 #[cfg(feature = "compression")]
2410 use alloc::vec;
2411 use alloc::vec::Vec;
2412
2413 #[cfg(feature = "compression")]
2414 use crate::config::runtime_payload_compress_threshold;
2415 #[cfg(feature = "compression")]
2416 use zstd_safe::CompressionLevel;
2417
2418 #[cfg(feature = "compression")]
2426 pub fn compress_if_beneficial(payload: &'_ [u8]) -> (bool, Cow<'_, [u8]>) {
2427 if payload.len() < runtime_payload_compress_threshold() {
2428 return (false, Cow::Borrowed(payload));
2429 }
2430
2431 let Some(compressed) = compress_to_vec_bounded(payload, payload.len().saturating_sub(2))
2433 else {
2434 return (false, Cow::Borrowed(payload));
2435 };
2436
2437 if compressed.len() + 1 >= payload.len() {
2439 (false, Cow::Borrowed(payload))
2440 } else {
2441 (true, Cow::Owned(compressed))
2442 }
2443 }
2444
2445 #[cfg(feature = "compression")]
2446 fn compress_to_vec_bounded(input: &[u8], max_output: usize) -> Option<Vec<u8>> {
2447 if input.is_empty() || max_output == 0 {
2448 return None;
2449 }
2450
2451 let mut out = vec![0u8; max_output];
2452 let level: CompressionLevel = 1;
2454 let written = zstd_safe::compress(&mut out[..], input, level).ok()?;
2455 out.truncate(written);
2456 Some(out)
2457 }
2458
2459 #[cfg(feature = "compression")]
2469 pub fn decompress(compressed: &[u8], expected_len: usize) -> Result<Vec<u8>, TelemetryError> {
2470 let mut out = vec![0u8; expected_len];
2471 let written = zstd_safe::decompress(&mut out[..], compressed)
2472 .map_err(|_| TelemetryError::Unpack("decompression failed"))?;
2473 if written != expected_len {
2474 return Err(TelemetryError::Unpack("decompressed size mismatch"));
2475 }
2476 Ok(out)
2477 }
2478
2479 #[cfg(not(feature = "compression"))]
2481 pub fn compress_if_beneficial<'a>(payload: &'a [u8]) -> (bool, Cow<'a, [u8]>) {
2483 (false, Cow::Borrowed(payload))
2484 }
2485
2486 #[cfg(not(feature = "compression"))]
2487 pub fn decompress(_compressed: &[u8], _expected_len: usize) -> Result<Vec<u8>, TelemetryError> {
2489 Err(TelemetryError::Unpack(
2490 "compressed payloads not supported (compression feature disabled)",
2491 ))
2492 }
2493}
2494
2495#[cfg(test)]
2496mod endpoint_decode_tests {
2497 use super::*;
2498
2499 #[test]
2500 fn numeric_sender_alias_keeps_packet_wire_and_receipt_identity() {
2501 let ty = DataType::ReliableAck;
2502 let make = |sender: &str| {
2503 Packet::new_with_wire_contract(
2504 ty,
2505 message_meta(ty).endpoints_ref(),
2506 sender,
2507 200,
2508 9,
2509 Arc::from(42u64.to_le_bytes()),
2510 Some(MessageElement::Static(
2511 2,
2512 crate::MessageDataType::UInt32,
2513 crate::MessageClass::Data,
2514 )),
2515 Arc::from([123u64]),
2516 )
2517 .unwrap()
2518 };
2519 let named = make("FC");
2520 let alias = alloc::format!("@addr:{}", sender_address_u32("FC"));
2521 let unresolved = make(&alias);
2522 assert_eq!(unresolved.packet_id(), named.packet_id());
2523 let named_wire = pack_packet(&named);
2524 let alias_wire = pack_packet(&unresolved);
2525 assert_eq!(alias_wire, named_wire);
2526 assert_eq!(packet_id_from_wire(&alias_wire).unwrap(), named.packet_id());
2527 assert_eq!(
2528 unpack_packet(&alias_wire).unwrap().packet_id(),
2529 named.packet_id()
2530 );
2531 }
2532
2533 #[test]
2534 fn scheduling_type_is_bounded_and_never_replaces_frame_validation() {
2535 let pkt = Packet::new_with_wire_contract(
2536 DataType::ReliableAck,
2537 message_meta(DataType::ReliableAck).endpoints_ref(),
2538 "E2EACK:GS",
2539 100,
2540 1,
2541 Arc::from(42u64.to_le_bytes()),
2542 Some(MessageElement::Static(
2543 2,
2544 crate::MessageDataType::UInt32,
2545 crate::MessageClass::Data,
2546 )),
2547 Arc::from([123u64]),
2548 )
2549 .unwrap();
2550 let mut bytes = pack_packet(&pkt).to_vec();
2551 assert_eq!(scheduling_type(&bytes), Some(DataType::ReliableAck));
2552 let last = bytes.len() - 1;
2553 bytes[last] ^= 1;
2554 assert_eq!(scheduling_type(&bytes), Some(DataType::ReliableAck));
2555 assert!(peek_routing_envelope(&bytes).is_err());
2556 assert!(unpack_packet(&bytes).is_err());
2557 assert_eq!(scheduling_type(&[0, 0, 0x80]), None);
2558 assert_eq!(scheduling_type(&[0xff; 1024]), None);
2559 assert_eq!(scheduling_type(&[]), None);
2560 let unknown_prefix = [0, 0, 0xff, 0xff, 0xff, 0xff, 0x0f];
2561 assert_eq!(scheduling_type(&unknown_prefix), Some(DataType(u32::MAX)));
2562 assert!(peek_routing_envelope(&unknown_prefix).is_err());
2563 assert!(unpack_packet(&unknown_prefix).is_err());
2564 }
2565
2566 #[test]
2567 fn receipt_contract_omits_redundant_builtin_endpoint_bitmap() {
2568 for ty in [
2569 DataType::ReliableAck,
2570 DataType::ReliablePartialAck,
2571 DataType::ReliablePacketRequest,
2572 ] {
2573 let shape = Some(MessageElement::Static(
2574 2,
2575 crate::MessageDataType::UInt32,
2576 crate::MessageClass::Data,
2577 ));
2578 let targets: Arc<[u64]> = Arc::from([0xa7fcbb129e92435f, 0x41426f62de53eb28]);
2579 let payload: Arc<[u8]> = Arc::from(0x9c2a63d6d7d44100u64.to_le_bytes());
2580 let make = |eps: &[DataEndpoint]| {
2581 Packet::new_with_wire_contract(
2582 ty,
2583 eps,
2584 "E2EACK:GS",
2585 1791737696510,
2586 1234,
2587 payload.clone(),
2588 shape,
2589 targets.clone(),
2590 )
2591 .unwrap()
2592 };
2593 let packet = make(message_meta(ty).endpoints_ref());
2594 let packed = pack_packet(&packet);
2595 assert_eq!(packed[0] & FLAG_ENDPOINT_BITMAP_PRESENT, 0);
2596 assert_ne!(packed[0] & FLAG_WIRE_CONTRACT, 0);
2597 let routing = peek_routing_envelope(&packed).unwrap();
2598 assert_eq!(routing.endpoints().collect::<Vec<_>>(), packet.endpoints());
2599 assert_eq!(
2600 routing.target_senders().collect::<Vec<_>>(),
2601 targets.as_ref()
2602 );
2603 let decoded = unpack_packet(&packed).unwrap();
2604 assert_eq!(decoded.packet_id(), packet.packet_id());
2605 assert_eq!(decoded.payload(), packet.payload());
2606 assert_eq!(decoded.wire_target_senders(), targets.as_ref());
2607 let explicit = pack_packet(&make(&[DataEndpoint::Discovery]));
2609 assert_ne!(explicit[0] & FLAG_ENDPOINT_LIST_PRESENT, 0);
2610 assert_eq!(
2611 explicit.len(),
2612 packed.len() + uleb128_size(u64::from(DataEndpoint::Discovery.0))
2613 );
2614 assert_eq!(
2615 unpack_packet(&explicit).unwrap().endpoints(),
2616 &[DataEndpoint::Discovery]
2617 );
2618 }
2619 }
2620
2621 #[test]
2622 fn borrowed_routing_header_preserves_targets_and_validates_truncated_contracts() {
2623 let targets: Arc<[u64]> = Arc::from([1, u64::MAX, 42]);
2624 let packet = Packet::new_with_wire_contract(
2625 DataType::DiscoverySchema,
2626 &[DataEndpoint::Discovery],
2627 "PEER",
2628 1234,
2629 7,
2630 Arc::from([1u8, 2, 3]),
2631 Some(MessageElement::Dynamic(
2632 crate::MessageDataType::Binary,
2633 crate::MessageClass::Data,
2634 )),
2635 targets.clone(),
2636 )
2637 .unwrap();
2638 let packed = pack_packet(&packet);
2639 let routing = peek_routing_envelope(&packed).unwrap();
2640 let full = unpack_packet(&packed).unwrap();
2641 assert_eq!(routing.ty, full.data_type());
2642 assert_eq!(routing.endpoints().collect::<Vec<_>>(), full.endpoints());
2643 assert_eq!(
2644 routing.target_senders().collect::<Vec<_>>(),
2645 targets.as_ref()
2646 );
2647 assert_eq!(routing.timestamp_ms, full.timestamp());
2648 let data = verify_crc32(&packed).unwrap();
2650 let mut reader = ByteReader::new(data);
2651 read_routing_envelope(&mut reader).unwrap();
2652 for end in 0..reader.off {
2653 let mut truncated = data[..end].to_vec();
2654 append_crc32(&mut truncated);
2655 assert!(
2656 peek_routing_envelope(&truncated).is_err(),
2657 "accepted cut at {end}"
2658 );
2659 }
2660 }
2661
2662 #[test]
2663 fn large_packet_with_long_prefix_does_not_grow_packing_buffer() {
2664 let payload: Vec<u8> = (0..3500)
2665 .map(|i| ((i * 73 + i / 251) % 256) as u8)
2666 .collect();
2667 let packet = Packet::new(
2668 DataType::DiscoverySchema,
2669 &[DataEndpoint::Discovery],
2670 "LONG_PREFIX_SENDER",
2671 u64::MAX,
2672 payload.into(),
2673 )
2674 .unwrap()
2675 .with_nonce(u16::MAX);
2676 assert!(header_size_bytes(&packet) > 16);
2677 let packed = pack_packet(&packet);
2678 let unpacked = unpack_packet(&packed).unwrap();
2679 assert_eq!(unpacked.payload(), packet.payload());
2680 assert_eq!(unpacked.timestamp(), u64::MAX);
2681 assert_eq!(unpacked.nonce(), u16::MAX);
2682 }
2683
2684 #[test]
2685 fn bitmap_decode_preserves_sorted_unique_destinations() {
2686 let input = [
2687 DataEndpoint::TelemetryError,
2688 DataEndpoint::TimeSync,
2689 DataEndpoint::Discovery,
2690 DataEndpoint::TimeSync,
2691 ];
2692 let decoded = expand_endpoint_bitmap(&build_endpoint_bitmap(&input)).unwrap();
2693 assert_eq!(
2694 &*decoded,
2695 &[
2696 DataEndpoint::TimeSync,
2697 DataEndpoint::Discovery,
2698 DataEndpoint::TelemetryError
2699 ]
2700 );
2701 assert!(
2702 expand_endpoint_bitmap(&[0; EP_BITMAP_BYTES])
2703 .unwrap()
2704 .is_empty()
2705 );
2706 assert!(expand_endpoint_bitmap(&[0; EP_BITMAP_BYTES - 1]).is_err());
2707 }
2708
2709 #[test]
2710 fn bitmap_decode_rejects_unregistered_endpoint_bits() {
2711 if let Some(id) =
2712 (0..=MAX_VALUE_DATA_ENDPOINT).find(|id| DataEndpoint::try_from_u32(*id).is_none())
2713 {
2714 let mut bitmap = [0; EP_BITMAP_BYTES];
2715 bitmap[id as usize / 8] = 1 << (id % 8);
2716 assert!(expand_endpoint_bitmap(&bitmap).is_err());
2717 }
2718 }
2719}
2720
2721#[cfg(all(test, feature = "discovery", feature = "compression"))]
2722mod bootstrap_compression_tests {
2723 use super::*;
2724
2725 #[test]
2726 fn bootstrap_payloads_remain_decodable_without_optional_compression() {
2727 let payload: Arc<[u8]> = Arc::from(alloc::vec![b'A'; 4096]);
2728 assert!(payload_compression::compress_if_beneficial(&payload).0);
2729 for ty in [
2730 DataType::DiscoverySchema,
2731 DataType::DiscoveryTopology,
2732 DataType::ManagedVariableValue,
2733 ] {
2734 let packet =
2735 Packet::new(ty, &[DataEndpoint::Discovery], "HOST", 123, payload.clone()).unwrap();
2736 let wire = pack_packet(&packet);
2737 assert_eq!(wire[0] & FLAG_COMPRESSED_PAYLOAD, 0, "{ty:?}");
2738 assert_eq!(packet_wire_size(&packet), wire.len());
2739 assert_eq!(unpack_packet(&wire).unwrap().payload(), payload.as_ref());
2740 }
2741 }
2742
2743 #[test]
2744 fn ordinary_data_keeps_optional_payload_compression() {
2745 let payload: Arc<[u8]> = Arc::from(alloc::vec![b'A'; 4096]);
2746 let packet = Packet::new(
2747 DataType::TelemetryError,
2748 &[DataEndpoint::TelemetryError],
2749 "HOST",
2750 123,
2751 payload.clone(),
2752 )
2753 .unwrap();
2754 let wire = pack_packet(&packet);
2755 assert_ne!(wire[0] & FLAG_COMPRESSED_PAYLOAD, 0);
2756 assert_eq!(packet_wire_size(&packet), wire.len());
2757 assert_eq!(unpack_packet(&wire).unwrap().payload(), payload.as_ref());
2758 }
2759}