hatchery_core/
multiplex.rs

1//! Binary multiplex framing for the C2 byte restreamer (A2).
2//!
3//! C2 sees only the envelope: `seq`, destination, channel kind and size.
4//! Payload bytes are opaque (inner ciphertext). No g4a / m4a parsing here.
5
6use serde::{Deserialize, Serialize};
7use std::fmt;
8
9/// Wire version for multiplex frames.
10pub const MULTIPLEX_VERSION: u8 = 1;
11
12/// Maximum DATA payload length (1 MiB). Larger payloads are split by the sender.
13pub const MAX_DATA_PAYLOAD: usize = 1024 * 1024;
14
15/// Frame header size before the payload: version, type, channel_id, seq, length.
16pub const FRAME_HEADER_LEN: usize = 1 + 1 + 4 + 8 + 4;
17
18#[repr(u8)]
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum FrameType {
21    Open = 1,
22    Data = 2,
23    Close = 3,
24    Window = 4,
25    Ping = 5,
26    Pong = 6,
27    GoAway = 7,
28    Ack = 8,
29    Resume = 9,
30    /// C2 → node: request journal entries with seq > N.
31    JournalReq = 10,
32    /// node → C2: journal entries (payload = JournalReplyBody).
33    JournalReply = 11,
34}
35
36impl FrameType {
37    pub fn from_u8(v: u8) -> Option<Self> {
38        Some(match v {
39            1 => Self::Open,
40            2 => Self::Data,
41            3 => Self::Close,
42            4 => Self::Window,
43            5 => Self::Ping,
44            6 => Self::Pong,
45            7 => Self::GoAway,
46            8 => Self::Ack,
47            9 => Self::Resume,
48            10 => Self::JournalReq,
49            11 => Self::JournalReply,
50            _ => return None,
51        })
52    }
53}
54
55/// Logical channel kinds C2 routes without inspecting contents.
56#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
57#[serde(rename_all = "snake_case")]
58pub enum ChannelKind {
59    G4aControl,
60    G4aPty,
61    G4aEvents,
62    M4a,
63    Vault,
64    Observation,
65}
66
67impl ChannelKind {
68    pub fn as_str(self) -> &'static str {
69        match self {
70            Self::G4aControl => "g4a.control",
71            Self::G4aPty => "g4a.pty",
72            Self::G4aEvents => "g4a.events",
73            Self::M4a => "m4a",
74            Self::Vault => "vault",
75            Self::Observation => "observation",
76        }
77    }
78
79    pub fn parse(s: &str) -> Option<Self> {
80        Some(match s {
81            "g4a.control" => Self::G4aControl,
82            "g4a.pty" => Self::G4aPty,
83            "g4a.events" => Self::G4aEvents,
84            "m4a" => Self::M4a,
85            "vault" => Self::Vault,
86            "observation" => Self::Observation,
87            _ => return None,
88        })
89    }
90}
91
92impl fmt::Display for ChannelKind {
93    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94        f.write_str(self.as_str())
95    }
96}
97
98/// Payload of an OPEN frame. No secrets.
99#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
100pub struct OpenPayload {
101    pub channel_kind: ChannelKind,
102    /// Opaque node identifier assigned by C2; core never sees node addresses.
103    pub node_id: String,
104}
105
106/// One multiplex frame.
107#[derive(Clone, Debug, PartialEq, Eq)]
108pub struct Frame {
109    pub version: u8,
110    pub frame_type: FrameType,
111    pub channel_id: u32,
112    /// For DATA: event sequence. For ACK/RESUME: the acknowledged / resume-after number.
113    pub seq: u64,
114    pub payload: Vec<u8>,
115}
116
117#[derive(Clone, Debug, PartialEq, Eq)]
118pub enum FrameError {
119    Truncated,
120    BadVersion(u8),
121    BadType(u8),
122    PayloadTooLarge(usize),
123    Io(String),
124}
125
126impl fmt::Display for FrameError {
127    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
128        match self {
129            Self::Truncated => write!(f, "truncated frame"),
130            Self::BadVersion(v) => write!(f, "unsupported multiplex version {v}"),
131            Self::BadType(t) => write!(f, "unknown frame type {t}"),
132            Self::PayloadTooLarge(n) => write!(f, "payload length {n} exceeds limit"),
133            Self::Io(msg) => write!(f, "{msg}"),
134        }
135    }
136}
137
138impl std::error::Error for FrameError {}
139
140impl Frame {
141    pub fn open(channel_id: u32, payload: &OpenPayload) -> Result<Self, FrameError> {
142        let body = serde_json::to_vec(payload).map_err(|e| FrameError::Io(e.to_string()))?;
143        Ok(Self {
144            version: MULTIPLEX_VERSION,
145            frame_type: FrameType::Open,
146            channel_id,
147            seq: 0,
148            payload: body,
149        })
150    }
151
152    pub fn data(channel_id: u32, seq: u64, payload: Vec<u8>) -> Result<Self, FrameError> {
153        if payload.len() > MAX_DATA_PAYLOAD {
154            return Err(FrameError::PayloadTooLarge(payload.len()));
155        }
156        Ok(Self {
157            version: MULTIPLEX_VERSION,
158            frame_type: FrameType::Data,
159            channel_id,
160            seq,
161            payload,
162        })
163    }
164
165    pub fn ack(seq: u64) -> Self {
166        Self {
167            version: MULTIPLEX_VERSION,
168            frame_type: FrameType::Ack,
169            channel_id: 0,
170            seq,
171            payload: Vec::new(),
172        }
173    }
174
175    pub fn resume(after: u64) -> Self {
176        Self {
177            version: MULTIPLEX_VERSION,
178            frame_type: FrameType::Resume,
179            channel_id: 0,
180            seq: after,
181            payload: Vec::new(),
182        }
183    }
184
185    pub fn window(channel_id: u32, credit: u32) -> Self {
186        Self {
187            version: MULTIPLEX_VERSION,
188            frame_type: FrameType::Window,
189            channel_id,
190            seq: u64::from(credit),
191            payload: Vec::new(),
192        }
193    }
194
195    pub fn ping(nonce: u64) -> Self {
196        Self {
197            version: MULTIPLEX_VERSION,
198            frame_type: FrameType::Ping,
199            channel_id: 0,
200            seq: nonce,
201            payload: Vec::new(),
202        }
203    }
204
205    pub fn pong(nonce: u64) -> Self {
206        Self {
207            version: MULTIPLEX_VERSION,
208            frame_type: FrameType::Pong,
209            channel_id: 0,
210            seq: nonce,
211            payload: Vec::new(),
212        }
213    }
214
215    pub fn goaway(code: u32) -> Self {
216        Self {
217            version: MULTIPLEX_VERSION,
218            frame_type: FrameType::GoAway,
219            channel_id: 0,
220            seq: u64::from(code),
221            payload: Vec::new(),
222        }
223    }
224
225    pub fn close(channel_id: u32) -> Self {
226        Self {
227            version: MULTIPLEX_VERSION,
228            frame_type: FrameType::Close,
229            channel_id,
230            seq: 0,
231            payload: Vec::new(),
232        }
233    }
234
235    /// Encode big-endian wire bytes.
236    pub fn encode(&self) -> Result<Vec<u8>, FrameError> {
237        if self.payload.len() > MAX_DATA_PAYLOAD {
238            return Err(FrameError::PayloadTooLarge(self.payload.len()));
239        }
240        let mut out = Vec::with_capacity(FRAME_HEADER_LEN + self.payload.len());
241        out.push(self.version);
242        out.push(self.frame_type as u8);
243        out.extend_from_slice(&self.channel_id.to_be_bytes());
244        out.extend_from_slice(&self.seq.to_be_bytes());
245        out.extend_from_slice(&(self.payload.len() as u32).to_be_bytes());
246        out.extend_from_slice(&self.payload);
247        Ok(out)
248    }
249
250    /// Decode one frame from a complete buffer. Returns `(frame, bytes_consumed)`.
251    pub fn decode(buf: &[u8]) -> Result<(Self, usize), FrameError> {
252        if buf.len() < FRAME_HEADER_LEN {
253            return Err(FrameError::Truncated);
254        }
255        let version = buf[0];
256        if version != MULTIPLEX_VERSION {
257            return Err(FrameError::BadVersion(version));
258        }
259        let frame_type = FrameType::from_u8(buf[1]).ok_or(FrameError::BadType(buf[1]))?;
260        let channel_id = u32::from_be_bytes([buf[2], buf[3], buf[4], buf[5]]);
261        let seq = u64::from_be_bytes([
262            buf[6], buf[7], buf[8], buf[9], buf[10], buf[11], buf[12], buf[13],
263        ]);
264        let length = u32::from_be_bytes([buf[14], buf[15], buf[16], buf[17]]) as usize;
265        if length > MAX_DATA_PAYLOAD {
266            return Err(FrameError::PayloadTooLarge(length));
267        }
268        let total = FRAME_HEADER_LEN + length;
269        if buf.len() < total {
270            return Err(FrameError::Truncated);
271        }
272        Ok((
273            Self {
274                version,
275                frame_type,
276                channel_id,
277                seq,
278                payload: buf[FRAME_HEADER_LEN..total].to_vec(),
279            },
280            total,
281        ))
282    }
283
284    pub fn parse_open(&self) -> Result<OpenPayload, FrameError> {
285        if self.frame_type != FrameType::Open {
286            return Err(FrameError::Io("not an OPEN frame".into()));
287        }
288        serde_json::from_slice(&self.payload).map_err(|e| FrameError::Io(e.to_string()))
289    }
290
291    /// Journal request: `seq` = after, payload = u32 limit (big-endian).
292    pub fn journal_req(channel_id: u32, after: u64, limit: u32) -> Self {
293        Self {
294            version: MULTIPLEX_VERSION,
295            frame_type: FrameType::JournalReq,
296            channel_id,
297            seq: after,
298            payload: limit.to_be_bytes().to_vec(),
299        }
300    }
301
302    pub fn journal_req_limit(&self) -> Option<u32> {
303        if self.frame_type != FrameType::JournalReq || self.payload.len() != 4 {
304            return None;
305        }
306        Some(u32::from_be_bytes([
307            self.payload[0],
308            self.payload[1],
309            self.payload[2],
310            self.payload[3],
311        ]))
312    }
313
314    pub fn journal_reply(channel_id: u32, body: &JournalReplyBody) -> Result<Self, FrameError> {
315        let payload = body.encode()?;
316        if payload.len() > MAX_DATA_PAYLOAD {
317            return Err(FrameError::PayloadTooLarge(payload.len()));
318        }
319        Ok(Self {
320            version: MULTIPLEX_VERSION,
321            frame_type: FrameType::JournalReply,
322            channel_id,
323            seq: body.after,
324            payload,
325        })
326    }
327}
328
329/// One journal entry on the wire (opaque payload).
330#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
331pub struct JournalWireEntry {
332    pub seq: u64,
333    pub payload: Vec<u8>,
334}
335
336/// Body of JournalReply: entries with seq > after for one channel.
337#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
338pub struct JournalReplyBody {
339    pub after: u64,
340    pub entries: Vec<JournalWireEntry>,
341}
342
343impl JournalReplyBody {
344    pub fn encode(&self) -> Result<Vec<u8>, FrameError> {
345        serde_json::to_vec(self).map_err(|e| FrameError::Io(e.to_string()))
346    }
347
348    pub fn decode(bytes: &[u8]) -> Result<Self, FrameError> {
349        serde_json::from_slice(bytes).map_err(|e| FrameError::Io(e.to_string()))
350    }
351}
352
353/// Drain-stream envelope: C2 prefixes each uplink DATA with an opaque node id for core.
354#[derive(Clone, Debug, PartialEq, Eq)]
355pub struct DrainEnvelope {
356    /// Opaque destination / source node id. Not an address.
357    pub node_id: String,
358    pub frame: Frame,
359}
360
361impl DrainEnvelope {
362    pub fn encode(&self) -> Result<Vec<u8>, FrameError> {
363        let id = self.node_id.as_bytes();
364        if id.len() > 255 {
365            return Err(FrameError::Io("node_id too long".into()));
366        }
367        let frame_bytes = self.frame.encode()?;
368        let mut out = Vec::with_capacity(1 + id.len() + frame_bytes.len());
369        out.push(id.len() as u8);
370        out.extend_from_slice(id);
371        out.extend_from_slice(&frame_bytes);
372        Ok(out)
373    }
374
375    pub fn decode(buf: &[u8]) -> Result<(Self, usize), FrameError> {
376        if buf.is_empty() {
377            return Err(FrameError::Truncated);
378        }
379        let id_len = buf[0] as usize;
380        if buf.len() < 1 + id_len {
381            return Err(FrameError::Truncated);
382        }
383        let node_id = String::from_utf8(buf[1..1 + id_len].to_vec())
384            .map_err(|_| FrameError::Io("node_id is not utf-8".into()))?;
385        let (frame, n) = Frame::decode(&buf[1 + id_len..])?;
386        Ok((Self { node_id, frame }, 1 + id_len + n))
387    }
388}
389
390#[cfg(test)]
391mod tests {
392    use super::*;
393
394    #[test]
395    fn roundtrip_data_ack_resume() {
396        let data = Frame::data(7, 42, b"opaque-ciphertext".to_vec()).unwrap();
397        let bytes = data.encode().unwrap();
398        let (decoded, n) = Frame::decode(&bytes).unwrap();
399        assert_eq!(n, bytes.len());
400        assert_eq!(decoded, data);
401
402        let ack = Frame::ack(42);
403        let (d, _) = Frame::decode(&ack.encode().unwrap()).unwrap();
404        assert_eq!(d.frame_type, FrameType::Ack);
405        assert_eq!(d.seq, 42);
406
407        let resume = Frame::resume(41);
408        let (d, _) = Frame::decode(&resume.encode().unwrap()).unwrap();
409        assert_eq!(d.frame_type, FrameType::Resume);
410        assert_eq!(d.seq, 41);
411    }
412
413    #[test]
414    fn open_payload_roundtrip() {
415        let open = Frame::open(
416            1,
417            &OpenPayload {
418                channel_kind: ChannelKind::M4a,
419                node_id: "node-a".into(),
420            },
421        )
422        .unwrap();
423        let (decoded, _) = Frame::decode(&open.encode().unwrap()).unwrap();
424        let payload = decoded.parse_open().unwrap();
425        assert_eq!(payload.channel_kind, ChannelKind::M4a);
426        assert_eq!(payload.node_id, "node-a");
427    }
428
429    #[test]
430    fn rejects_oversized_payload() {
431        let big = vec![0u8; MAX_DATA_PAYLOAD + 1];
432        assert!(matches!(
433            Frame::data(1, 1, big),
434            Err(FrameError::PayloadTooLarge(_))
435        ));
436    }
437
438    #[test]
439    fn drain_envelope_prefixes_opaque_node_id() {
440        let env = DrainEnvelope {
441            node_id: "node-b".into(),
442            frame: Frame::data(3, 9, b"inner".to_vec()).unwrap(),
443        };
444        let bytes = env.encode().unwrap();
445        let (decoded, n) = DrainEnvelope::decode(&bytes).unwrap();
446        assert_eq!(n, bytes.len());
447        assert_eq!(decoded.node_id, "node-b");
448        assert_eq!(decoded.frame.seq, 9);
449        assert_eq!(decoded.frame.payload, b"inner");
450    }
451
452    #[test]
453    fn journal_req_roundtrip() {
454        let f = Frame::journal_req(3, 10, 64);
455        let (d, _) = Frame::decode(&f.encode().unwrap()).unwrap();
456        assert_eq!(d.frame_type, FrameType::JournalReq);
457        assert_eq!(d.seq, 10);
458        assert_eq!(d.journal_req_limit(), Some(64));
459        let body = JournalReplyBody {
460            after: 10,
461            entries: vec![JournalWireEntry {
462                seq: 11,
463                payload: b"x".to_vec(),
464            }],
465        };
466        let reply = Frame::journal_reply(3, &body).unwrap();
467        let (d, _) = Frame::decode(&reply.encode().unwrap()).unwrap();
468        let decoded = JournalReplyBody::decode(&d.payload).unwrap();
469        assert_eq!(decoded.entries[0].seq, 11);
470    }
471
472    #[test]
473    fn truncated_frame_errors() {
474        assert!(matches!(Frame::decode(&[1, 2, 3]), Err(FrameError::Truncated)));
475        let mut partial = Frame::data(1, 1, b"hello".to_vec()).unwrap().encode().unwrap();
476        partial.pop();
477        assert!(matches!(Frame::decode(&partial), Err(FrameError::Truncated)));
478    }
479}
hatchery_core/multiplex.rs

hatchery_core/
multiplex.rs
