core_ipc/
framing.rs

1//! Postcard serialization and length-prefixed wire framing.
2//!
3//! Two layers:
4//! - **Serialization:** `encode_frame` / `decode_frame` convert between typed values
5//!   and postcard byte payloads. These are symmetric: encode produces what decode consumes.
6//! - **Wire I/O:** `write_frame` / `read_frame` add/strip a 4-byte big-endian length
7//!   prefix for socket transport. The length prefix is a wire concern only — internal
8//!   channels (bus routing, `BusServer::publish`, subscriber mpsc channels) carry raw
9//!   postcard payloads without it.
10//!
11//! Wire format on the socket: `[4-byte BE length][postcard payload]`
12
13use serde::{Serialize, de::DeserializeOwned};
14use tokio::io::{AsyncReadExt, AsyncWriteExt};
15
16/// Maximum frame payload size (16 MiB). Prevents OOM from malformed length prefixes.
17pub const MAX_FRAME_SIZE: u32 = 16 * 1024 * 1024;
18
19/// Serialize a value to postcard bytes.
20///
21/// Symmetric with [`decode_frame`]: `decode_frame(encode_frame(v)) == v`.
22///
23/// The returned bytes are suitable for `BusServer::publish()`, internal channel
24/// transport, and as input to `write_frame()` for socket I/O.
25///
26/// # Errors
27///
28/// Returns an error if postcard serialization fails.
29pub fn encode_frame<T: Serialize>(value: &T) -> core_types::Result<Vec<u8>> {
30    postcard::to_allocvec(value)
31        .map_err(|e| core_types::Error::Ipc(format!("serialization failed: {e}")))
32}
33
34/// Deserialize a value from postcard bytes.
35///
36/// Symmetric with [`encode_frame`]: `decode_frame(encode_frame(v)) == v`.
37///
38/// # Errors
39///
40/// Returns an error if postcard deserialization fails.
41pub fn decode_frame<T: DeserializeOwned>(payload: &[u8]) -> core_types::Result<T> {
42    postcard::from_bytes(payload)
43        .map_err(|e| core_types::Error::Ipc(format!("deserialization failed: {e}")))
44}
45
46/// Read a single length-prefixed frame from an async reader.
47///
48/// # Errors
49///
50/// Returns an error on I/O failure or if the frame size exceeds `MAX_FRAME_SIZE`.
51pub async fn read_frame<R: tokio::io::AsyncRead + Unpin>(
52    reader: &mut R,
53) -> core_types::Result<Vec<u8>> {
54    let mut len_buf = [0u8; 4];
55    reader.read_exact(&mut len_buf).await?;
56    let len = u32::from_be_bytes(len_buf);
57    if len > MAX_FRAME_SIZE {
58        return Err(core_types::Error::Ipc(format!(
59            "frame size {len} exceeds maximum {MAX_FRAME_SIZE}"
60        )));
61    }
62    let mut payload = vec![0u8; len as usize];
63    reader.read_exact(&mut payload).await?;
64    Ok(payload)
65}
66
67/// Write a single length-prefixed frame to an async writer.
68///
69/// # Errors
70///
71/// Returns an error on I/O failure.
72pub async fn write_frame<W: tokio::io::AsyncWrite + Unpin>(
73    writer: &mut W,
74    payload: &[u8],
75) -> core_types::Result<()> {
76    let len = u32::try_from(payload.len())
77        .map_err(|_| core_types::Error::Ipc("payload exceeds u32 length".into()))?;
78    writer.write_all(&len.to_be_bytes()).await?;
79    writer.write_all(payload).await?;
80    writer.flush().await?;
81    Ok(())
82}
83
84#[cfg(test)]
85mod tests {
86    use super::*;
87    use core_types::{DaemonId, EventKind, SecurityLevel};
88    use uuid::Uuid;
89
90    #[test]
91    fn encode_decode_roundtrip() {
92        let event = EventKind::DaemonStarted {
93            daemon_id: DaemonId::from_uuid(Uuid::from_u128(1)),
94            version: "0.1.0".into(),
95            capabilities: vec!["test".into()],
96        };
97        let ctx = crate::message::MessageContext::new(DaemonId::from_uuid(Uuid::from_u128(1)));
98        let msg = crate::message::Message::new(
99            &ctx,
100            event,
101            SecurityLevel::Internal,
102            std::time::Instant::now(),
103        );
104        let payload = encode_frame(&msg).unwrap();
105        let decoded: crate::message::Message<EventKind> = decode_frame(&payload).unwrap();
106        assert!(matches!(decoded.payload, EventKind::DaemonStarted { .. }));
107    }
108
109    #[tokio::test]
110    async fn async_read_write_roundtrip() {
111        let payload = b"hello postcard";
112        let mut buf = Vec::new();
113        write_frame(&mut buf, payload).await.unwrap();
114        let mut cursor = &buf[..];
115        let decoded = read_frame(&mut cursor).await.unwrap();
116        assert_eq!(decoded, payload);
117    }
118
119    // SECURITY INVARIANT: Frames exceeding MAX_FRAME_SIZE must be rejected on
120    // read to prevent OOM from malformed or malicious length prefixes.
121    #[tokio::test]
122    async fn oversized_frame_rejected_on_read() {
123        let oversized_len: u32 = MAX_FRAME_SIZE + 1;
124        let mut buf = Vec::new();
125        buf.extend_from_slice(&oversized_len.to_be_bytes());
126        // Don't need actual payload bytes — read_frame should reject at the length check.
127        buf.extend(std::iter::repeat_n(0u8, 64));
128        let mut cursor = &buf[..];
129        let result = read_frame(&mut cursor).await;
130        assert!(
131            result.is_err(),
132            "frame exceeding MAX_FRAME_SIZE must be rejected"
133        );
134    }
135
136    // SECURITY INVARIANT: Zero-length frames must roundtrip correctly —
137    // edge case that must not panic or produce garbage.
138    #[tokio::test]
139    async fn zero_length_frame_roundtrips() {
140        let payload: &[u8] = &[];
141        let mut buf = Vec::new();
142        write_frame(&mut buf, payload).await.unwrap();
143        let mut cursor = &buf[..];
144        let decoded = read_frame(&mut cursor).await.unwrap();
145        assert!(decoded.is_empty());
146    }
147}
core_ipc/framing.rs

core_ipc/
framing.rs
