tf_types/
packet.rs

1#![allow(clippy::unusual_byte_groupings)]
2//! Packet mode (TF-0011) — Rust mirror of
3//! `tools/tf-types-ts/src/core/packet.ts`.
4
5use base64::engine::general_purpose::STANDARD;
6use base64::Engine;
7use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
8use flate2::read::DeflateDecoder;
9use flate2::write::DeflateEncoder;
10use flate2::Compression;
11use serde::{Deserialize, Serialize};
12use serde_json::Value;
13use sha2::{Digest, Sha256};
14use std::io::{Read, Write};
15
16use crate::canonicalize;
17use crate::expiration::{is_within_window, Window};
18
19#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
20pub struct PacketSignatureEnvelope {
21    pub algorithm: String,
22    pub signer: String,
23    pub signature: String,
24}
25
26#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
27pub struct PacketFragmentHeader {
28    pub fragment_id: String,
29    pub index: u32,
30    pub count: u32,
31    pub total_payload_bytes: u32,
32    pub payload_digest: String,
33}
34
35#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
36pub struct Packet {
37    pub packet_version: String,
38    pub packet_id: String,
39    pub source: String,
40    pub destination: String,
41    pub priority: String,
42    #[serde(skip_serializing_if = "Option::is_none", default)]
43    pub emergency: Option<bool>,
44    pub created_at: String,
45    #[serde(skip_serializing_if = "Option::is_none", default)]
46    pub expires_at: Option<String>,
47    #[serde(skip_serializing_if = "Option::is_none", default)]
48    pub ttl_hops: Option<u32>,
49    #[serde(skip_serializing_if = "Option::is_none", default)]
50    pub route_constraints: Option<Vec<String>>,
51    #[serde(skip_serializing_if = "Option::is_none", default)]
52    pub encoding: Option<String>,
53    #[serde(skip_serializing_if = "Option::is_none", default)]
54    pub compression: Option<String>,
55    pub payload: String,
56    #[serde(skip_serializing_if = "Option::is_none", default)]
57    pub session_ref: Option<String>,
58    #[serde(skip_serializing_if = "Option::is_none", default)]
59    pub fragment: Option<PacketFragmentHeader>,
60    pub signature: PacketSignatureEnvelope,
61}
62
63#[derive(Clone, Debug)]
64pub struct SignPacketArgs<'a> {
65    pub packet_id: String,
66    pub source: String,
67    pub destination: String,
68    pub priority: String,
69    pub payload: &'a [u8],
70    pub encoding: Option<String>,
71    pub compression: Option<String>,
72    pub emergency: bool,
73    pub expires_at: Option<String>,
74    pub ttl_hops: Option<u32>,
75    pub route_constraints: Option<Vec<String>>,
76    pub session_ref: Option<String>,
77    pub private_key: [u8; 32],
78    pub signer: String,
79    pub created_at: Option<String>,
80}
81
82pub fn packet_signing_bytes(p: &Packet) -> [u8; 32] {
83    let mut value = serde_json::to_value(p).unwrap_or(Value::Null);
84    if let Value::Object(map) = &mut value {
85        map.remove("signature");
86    }
87    let canonical = canonicalize(&value).unwrap_or_default();
88    Sha256::digest(canonical.as_bytes()).into()
89}
90
91pub fn sign_packet(args: SignPacketArgs<'_>) -> Result<Packet, String> {
92    if args.priority == "P0" && !args.emergency {
93        return Err("P0 priority is reserved for emergency packets".into());
94    }
95    let encoding = args.encoding.unwrap_or_else(|| "cbor".to_string());
96    let compression = args.compression.unwrap_or_else(|| "none".to_string());
97
98    // Wrap payload in canonical envelope before encoding.
99    let payload_bytes: Vec<u8> = if encoding == "cbor" {
100        serde_cbor_envelope(args.payload)
101    } else {
102        let canonical = canonicalize(&serde_json::json!({
103            "raw": STANDARD.encode(args.payload),
104        }))
105        .map_err(|e| format!("canonicalize: {}", e))?;
106        canonical.into_bytes()
107    };
108
109    let final_bytes = if compression == "deflate" {
110        let mut enc = DeflateEncoder::new(Vec::new(), Compression::default());
111        enc.write_all(&payload_bytes).map_err(|e| e.to_string())?;
112        enc.finish().map_err(|e| e.to_string())?
113    } else {
114        payload_bytes
115    };
116
117    let mut draft = Packet {
118        packet_version: "1".into(),
119        packet_id: args.packet_id,
120        source: args.source.clone(),
121        destination: args.destination,
122        priority: args.priority,
123        emergency: if args.emergency { Some(true) } else { None },
124        created_at: args.created_at.unwrap_or_else(now_iso8601),
125        expires_at: args.expires_at,
126        ttl_hops: args.ttl_hops,
127        route_constraints: args.route_constraints.filter(|r| !r.is_empty()),
128        encoding: Some(encoding),
129        compression: Some(compression),
130        payload: STANDARD.encode(&final_bytes),
131        session_ref: args.session_ref,
132        fragment: None,
133        signature: PacketSignatureEnvelope {
134            algorithm: "ed25519".into(),
135            signer: args.signer.clone(),
136            signature: String::new(),
137        },
138    };
139    let digest = packet_signing_bytes(&draft);
140    let signing = SigningKey::from_bytes(&args.private_key);
141    let sig: Signature = signing.sign(&digest);
142    draft.signature.signature = STANDARD.encode(sig.to_bytes());
143    Ok(draft)
144}
145
146#[derive(Debug)]
147pub struct VerifyPacketResult {
148    pub ok: bool,
149    pub reason: Option<String>,
150    pub payload: Option<Vec<u8>>,
151}
152
153pub fn verify_packet(packet: &Packet, public_key: &[u8; 32], now: &str) -> VerifyPacketResult {
154    let rejected = |r: &str| VerifyPacketResult {
155        ok: false,
156        reason: Some(r.to_string()),
157        payload: None,
158    };
159    if packet.packet_version != "1" {
160        return rejected(&format!(
161            "unsupported packet_version {}",
162            packet.packet_version
163        ));
164    }
165    if packet.signature.signer != packet.source {
166        return rejected("signature signer does not match source");
167    }
168    if packet.priority == "P0" && packet.emergency != Some(true) {
169        return rejected("P0 reserved for emergency packets");
170    }
171    if let Some(expires) = &packet.expires_at {
172        let window = Window {
173            valid_until: Some(expires.as_str()),
174            ..Window::default()
175        };
176        if !is_within_window(&window, now) {
177            return rejected("packet expired");
178        }
179    }
180    let digest = packet_signing_bytes(packet);
181    let sig_bytes = match STANDARD.decode(&packet.signature.signature) {
182        Ok(b) => b,
183        Err(e) => return rejected(&format!("signature base64 decode: {}", e)),
184    };
185    let sig = match Signature::from_slice(&sig_bytes) {
186        Ok(s) => s,
187        Err(e) => return rejected(&format!("signature parse: {}", e)),
188    };
189    let vk = match VerifyingKey::from_bytes(public_key) {
190        Ok(v) => v,
191        Err(e) => return rejected(&format!("verifying key: {}", e)),
192    };
193    if vk.verify(&digest, &sig).is_err() {
194        return rejected("signature verification failed");
195    }
196    let wire = match STANDARD.decode(&packet.payload) {
197        Ok(b) => b,
198        Err(e) => return rejected(&format!("payload base64: {}", e)),
199    };
200    let decompressed = if packet.compression.as_deref() == Some("deflate") {
201        let mut dec = DeflateDecoder::new(&wire[..]);
202        let mut out = Vec::new();
203        if dec.read_to_end(&mut out).is_err() {
204            return rejected("deflate inflate failed");
205        }
206        out
207    } else {
208        wire
209    };
210    let payload = match packet.encoding.as_deref() {
211        Some("json") | None => match serde_json::from_slice::<Value>(&decompressed) {
212            Ok(v) => match v.get("raw").and_then(|r| r.as_str()) {
213                Some(b64) => match STANDARD.decode(b64) {
214                    Ok(b) => b,
215                    Err(_) => return rejected("payload base64 inner"),
216                },
217                None => return rejected("json envelope missing raw"),
218            },
219            Err(e) => return rejected(&format!("json decode: {}", e)),
220        },
221        Some("cbor") => match decode_cbor_envelope(&decompressed) {
222            Ok(b) => b,
223            Err(e) => return rejected(&format!("cbor decode: {}", e)),
224        },
225        Some(other) => return rejected(&format!("unknown encoding {}", other)),
226    };
227    VerifyPacketResult {
228        ok: true,
229        reason: None,
230        payload: Some(payload),
231    }
232}
233
234#[derive(Clone, Debug, Default)]
235pub struct FragmentOptions {
236    pub mtu: Option<usize>,
237}
238
239pub fn fragment_packet(
240    source: &Packet,
241    private_key: &[u8; 32],
242    opts: FragmentOptions,
243) -> Vec<Packet> {
244    let mtu = opts.mtu.unwrap_or(256);
245    let original = STANDARD.decode(&source.payload).unwrap_or_default();
246    let total_bytes = original.len();
247    if total_bytes <= mtu {
248        return vec![source.clone()];
249    }
250    let count = total_bytes.div_ceil(mtu);
251    let digest_hex = sha256_hex(&original);
252    let payload_digest = format!("sha256:{}", digest_hex);
253    let fragment_id = format!("frag-{}", source.packet_id);
254    let mut out = Vec::with_capacity(count);
255    for i in 0..count {
256        let start = i * mtu;
257        let end = (start + mtu).min(total_bytes);
258        let slice = &original[start..end];
259        let mut draft = Packet {
260            packet_version: "1".into(),
261            packet_id: format!("{}-{}", source.packet_id, i),
262            source: source.source.clone(),
263            destination: source.destination.clone(),
264            priority: source.priority.clone(),
265            emergency: source.emergency,
266            created_at: source.created_at.clone(),
267            expires_at: source.expires_at.clone(),
268            ttl_hops: source.ttl_hops,
269            route_constraints: source.route_constraints.clone(),
270            encoding: source.encoding.clone(),
271            compression: source.compression.clone(),
272            payload: STANDARD.encode(slice),
273            session_ref: source.session_ref.clone(),
274            fragment: Some(PacketFragmentHeader {
275                fragment_id: fragment_id.clone(),
276                index: i as u32,
277                count: count as u32,
278                total_payload_bytes: total_bytes as u32,
279                payload_digest: payload_digest.clone(),
280            }),
281            signature: PacketSignatureEnvelope {
282                algorithm: "ed25519".into(),
283                signer: source.source.clone(),
284                signature: String::new(),
285            },
286        };
287        let digest = packet_signing_bytes(&draft);
288        let signing = SigningKey::from_bytes(private_key);
289        let sig: Signature = signing.sign(&digest);
290        draft.signature.signature = STANDARD.encode(sig.to_bytes());
291        out.push(draft);
292    }
293    out
294}
295
296#[derive(Debug)]
297pub struct ReassembleResult {
298    pub ok: bool,
299    pub reason: Option<String>,
300    pub packet_id: Option<String>,
301    pub payload: Option<Vec<u8>>,
302}
303
304pub fn reassemble_fragments(fragments: &[Packet]) -> ReassembleResult {
305    let rejected = |r: &str| ReassembleResult {
306        ok: false,
307        reason: Some(r.to_string()),
308        packet_id: None,
309        payload: None,
310    };
311    if fragments.is_empty() {
312        return rejected("no fragments");
313    }
314    let header = match fragments[0].fragment.as_ref() {
315        Some(h) => h.clone(),
316        None => return rejected("first fragment missing fragment header"),
317    };
318    if fragments.len() != header.count as usize {
319        return rejected(&format!(
320            "expected {} fragments, got {}",
321            header.count,
322            fragments.len()
323        ));
324    }
325    let mut ordered: Vec<Option<&Packet>> = vec![None; header.count as usize];
326    for f in fragments {
327        let h = match f.fragment.as_ref() {
328            Some(h) => h,
329            None => return rejected("fragment missing header"),
330        };
331        if h.fragment_id != header.fragment_id {
332            return rejected("mismatched fragment_id");
333        }
334        if h.count != header.count {
335            return rejected("mismatched fragment count");
336        }
337        if ordered[h.index as usize].is_some() {
338            return rejected(&format!("duplicate fragment index {}", h.index));
339        }
340        ordered[h.index as usize] = Some(f);
341    }
342    let mut out = Vec::with_capacity(header.total_payload_bytes as usize);
343    for slot in ordered {
344        match slot {
345            Some(p) => {
346                let bytes = match STANDARD.decode(&p.payload) {
347                    Ok(b) => b,
348                    Err(e) => return rejected(&format!("base64: {}", e)),
349                };
350                out.extend_from_slice(&bytes);
351            }
352            None => return rejected("missing fragment slot"),
353        }
354    }
355    if out.len() != header.total_payload_bytes as usize {
356        return rejected(&format!(
357            "assembled {} bytes, expected {}",
358            out.len(),
359            header.total_payload_bytes
360        ));
361    }
362    let computed = format!("sha256:{}", sha256_hex(&out));
363    if computed != header.payload_digest {
364        return rejected("reassembled payload digest mismatch");
365    }
366    ReassembleResult {
367        ok: true,
368        reason: None,
369        packet_id: Some(header.fragment_id),
370        payload: Some(out),
371    }
372}
373
374fn serde_cbor_envelope(raw: &[u8]) -> Vec<u8> {
375    let value = ciborium::value::Value::Map(vec![(
376        ciborium::value::Value::Text("raw".into()),
377        ciborium::value::Value::Bytes(raw.to_vec()),
378    )]);
379    let mut buf = Vec::new();
380    ciborium::ser::into_writer(&value, &mut buf).expect("cbor encode");
381    buf
382}
383
384fn decode_cbor_envelope(bytes: &[u8]) -> Result<Vec<u8>, String> {
385    let value: ciborium::value::Value =
386        ciborium::de::from_reader(bytes).map_err(|e| format!("cbor: {}", e))?;
387    match value {
388        ciborium::value::Value::Map(entries) => {
389            for (k, v) in entries {
390                if matches!(k, ciborium::value::Value::Text(ref s) if s == "raw") {
391                    if let ciborium::value::Value::Bytes(b) = v {
392                        return Ok(b);
393                    }
394                }
395            }
396            Err("cbor envelope missing raw".into())
397        }
398        _ => Err("cbor envelope not a map".into()),
399    }
400}
401
402fn sha256_hex(b: &[u8]) -> String {
403    let digest = Sha256::digest(b);
404    digest.iter().map(|byte| format!("{:02x}", byte)).collect()
405}
406
407fn now_iso8601() -> String {
408    let secs = std::time::SystemTime::now()
409        .duration_since(std::time::UNIX_EPOCH)
410        .unwrap_or_default()
411        .as_secs() as i64;
412    let (y, m, d, h, mi, s) = secs_to_ymdhms(secs);
413    format!("{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z", y, m, d, h, mi, s)
414}
415
416fn secs_to_ymdhms(secs: i64) -> (i32, u32, u32, u32, u32, u32) {
417    let days = secs.div_euclid(86_400);
418    let time = secs.rem_euclid(86_400);
419    let hour = (time / 3600) as u32;
420    let minute = ((time % 3600) / 60) as u32;
421    let second = (time % 60) as u32;
422    let z = days + 719_468;
423    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
424    let doe = (z - era * 146_097) as u64;
425    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
426    let y = yoe as i64 + era * 400;
427    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
428    let mp = (5 * doy + 2) / 153;
429    let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
430    let m = if mp < 10 {
431        (mp + 3) as u32
432    } else {
433        (mp - 9) as u32
434    };
435    let year = if m <= 2 { y + 1 } else { y };
436    (year as i32, m, d, hour, minute, second)
437}
438
439/* ----------------------------------------------------------------------- */
440/*  LoRa-style channel simulation (mirror of TS `simulateLora`)            */
441/* ----------------------------------------------------------------------- */
442
443#[derive(Clone, Debug, Default)]
444pub struct LoraSimOptions {
445    /// Per-packet drop probability ∈ [0, 1]. Default 0 (lossless).
446    pub packet_loss: Option<f64>,
447    /// Bandwidth in bytes/sec. Default 250.
448    pub bandwidth_bps: Option<f64>,
449    /// Base latency in ms. Default 5000.
450    pub base_latency_ms: Option<f64>,
451}
452
453#[derive(Debug, Default)]
454pub struct LoraSimResult {
455    pub delivered: Vec<Packet>,
456    pub dropped: Vec<Packet>,
457    /// Cumulative simulated latency, ms.
458    pub total_latency_ms: f64,
459}
460
461/// Walk a list of packets through a one-way LoRa-style channel. Drops
462/// packets per `packet_loss`, accumulates latency proportional to size /
463/// `bandwidth_bps`. Pure simulation — no IO. The optional `rng_seed`
464/// argument makes the result deterministic for tests.
465pub fn simulate_lora(
466    packets: &[Packet],
467    opts: LoraSimOptions,
468    rng_seed: Option<u64>,
469) -> LoraSimResult {
470    let loss = opts.packet_loss.unwrap_or(0.0);
471    let bw = opts.bandwidth_bps.unwrap_or(250.0);
472    let base = opts.base_latency_ms.unwrap_or(5000.0);
473    let mut state = rng_seed.unwrap_or(0xdeadbeef_dead_beefu64);
474    let mut next = move || {
475        // xorshift64* — small deterministic PRNG, values in (0,1).
476        state ^= state << 13;
477        state ^= state >> 7;
478        state ^= state << 17;
479
480        (state.wrapping_mul(0x2545_F491_4F6C_DD1Du64) >> 11) as f64 / (1u64 << 53) as f64
481    };
482    let mut delivered = Vec::with_capacity(packets.len());
483    let mut dropped: Vec<Packet> = Vec::new();
484    let mut total_latency_ms = 0.0_f64;
485    for p in packets {
486        let canonical = serde_json::to_string(p).unwrap_or_default();
487        let size_bytes = canonical.len() as f64;
488        let tx_ms = (size_bytes / bw) * 1000.0;
489        total_latency_ms += base + tx_ms;
490        if next() < loss {
491            dropped.push(p.clone());
492        } else {
493            delivered.push(p.clone());
494        }
495    }
496    LoraSimResult {
497        delivered,
498        dropped,
499        total_latency_ms,
500    }
501}
502
503#[cfg(test)]
504mod lora_tests {
505    use super::*;
506    use rand::rngs::OsRng;
507
508    fn fixture(packet_id: &str) -> Packet {
509        let signer = "tf:actor:agent:example.com/x";
510        let mut signer_seed = [0u8; 32];
511        rand::RngCore::fill_bytes(&mut OsRng, &mut signer_seed);
512        sign_packet(SignPacketArgs {
513            packet_id: packet_id.into(),
514            source: signer.into(),
515            destination: "tf:actor:service:example.com/d".into(),
516            priority: "P3".into(),
517            payload: b"hi",
518            encoding: None,
519            compression: None,
520            emergency: false,
521            expires_at: None,
522            ttl_hops: None,
523            route_constraints: None,
524            session_ref: None,
525            private_key: signer_seed,
526            signer: signer.into(),
527            created_at: Some("2026-04-24T12:00:00Z".into()),
528        })
529        .expect("sign")
530    }
531
532    #[test]
533    fn lossless_channel_delivers_everything() {
534        let packets = vec![fixture("a"), fixture("b"), fixture("c")];
535        let r = simulate_lora(
536            &packets,
537            LoraSimOptions {
538                packet_loss: Some(0.0),
539                bandwidth_bps: Some(250.0),
540                base_latency_ms: Some(5000.0),
541            },
542            Some(1),
543        );
544        assert_eq!(r.delivered.len(), 3);
545        assert_eq!(r.dropped.len(), 0);
546        assert!(r.total_latency_ms > 15_000.0);
547    }
548
549    #[test]
550    fn full_loss_drops_everything() {
551        let packets = vec![fixture("a"), fixture("b")];
552        let r = simulate_lora(
553            &packets,
554            LoraSimOptions {
555                packet_loss: Some(1.0),
556                ..Default::default()
557            },
558            Some(42),
559        );
560        assert_eq!(r.delivered.len(), 0);
561        assert_eq!(r.dropped.len(), 2);
562    }
563
564    #[test]
565    fn deterministic_with_seed() {
566        let packets: Vec<Packet> = (0..10).map(|i| fixture(&format!("pkt-{}", i))).collect();
567        let r1 = simulate_lora(
568            &packets,
569            LoraSimOptions {
570                packet_loss: Some(0.5),
571                ..Default::default()
572            },
573            Some(99),
574        );
575        let r2 = simulate_lora(
576            &packets,
577            LoraSimOptions {
578                packet_loss: Some(0.5),
579                ..Default::default()
580            },
581            Some(99),
582        );
583        assert_eq!(r1.delivered.len(), r2.delivered.len());
584        assert_eq!(r1.dropped.len(), r2.dropped.len());
585    }
586}
tf_types/packet.rs

tf_types/
packet.rs
