From d6308c976b1a92631b47ee851bdafe43c09a69f1 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Thu, 17 Sep 2026 08:11:14 -0700 Subject: [PATCH] Add length-prefixed frame codec to proto Implemented-By: Laguna S 2.1 (OpenCode) --- crates/proto/src/frame.rs | 98 ++++++++++ crates/proto/src/lib.rs | 2 + .../tests/fixtures/frame/tool_request.bin | Bin 0 -> 163 bytes crates/proto/tests/frame.rs | 176 ++++++++++++++++++ docs/implementer-log.md | 1 + 5 files changed, 277 insertions(+) create mode 100644 crates/proto/src/frame.rs create mode 100644 crates/proto/tests/fixtures/frame/tool_request.bin create mode 100644 crates/proto/tests/frame.rs diff --git a/crates/proto/src/frame.rs b/crates/proto/src/frame.rs new file mode 100644 index 0000000..aff5a46 --- /dev/null +++ b/crates/proto/src/frame.rs @@ -0,0 +1,98 @@ +//! Length-prefixed frame codec: one envelope per big-endian 4-byte length. + +use std::io::Read; + +use crate::{Envelope, PROTOCOL_VERSION}; + +pub const MAX_FRAME: usize = 1_048_576; + +#[derive(Debug)] +pub enum FrameError { + Closed, + Empty, + TooLarge(usize), + BadVersion(u32), + Json(serde_json::Error), + Io(std::io::Error), +} + +impl std::fmt::Display for FrameError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + FrameError::Closed => write!(f, "stream closed before a frame"), + FrameError::Empty => write!(f, "frame length was zero"), + FrameError::TooLarge(n) => write!(f, "frame of {n} bytes exceeds the maximum"), + FrameError::BadVersion(v) => write!(f, "unsupported protocol version {v}"), + FrameError::Json(e) => write!(f, "invalid JSON envelope: {e}"), + FrameError::Io(e) => write!(f, "frame I/O error: {e}"), + } + } +} + +impl std::error::Error for FrameError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + FrameError::Json(e) => Some(e), + FrameError::Io(e) => Some(e), + _ => None, + } + } +} + +pub fn write_frame(w: &mut W, env: &Envelope) -> Result<(), FrameError> { + let body = serde_json::to_vec(env).map_err(FrameError::Json)?; + let len = body.len(); + if len > MAX_FRAME { + return Err(FrameError::TooLarge(len)); + } + w.write_all(&(len as u32).to_be_bytes()) + .map_err(FrameError::Io)?; + w.write_all(&body).map_err(FrameError::Io)?; + w.flush().map_err(FrameError::Io) +} + +pub fn read_frame(r: &mut R) -> Result { + let mut buf = [0u8; 4]; + let mut filled = 0; + while filled < 4 { + let n = read_bytes(r, &mut buf[filled..]).map_err(FrameError::Io)?; + if n == 0 { + return Err(if filled == 0 { + FrameError::Closed + } else { + FrameError::Io(std::io::Error::from(std::io::ErrorKind::UnexpectedEof)) + }); + } + filled += n; + } + + let len = u32::from_be_bytes(buf); + if len == 0 { + return Err(FrameError::Empty); + } + if len > MAX_FRAME as u32 { + return Err(FrameError::TooLarge(len as usize)); + } + + let len = len as usize; + let mut body = vec![0u8; len]; + r.read_exact(&mut body).map_err(FrameError::Io)?; + + let env = serde_json::from_slice::(&body).map_err(FrameError::Json)?; + if env.v != PROTOCOL_VERSION { + return Err(FrameError::BadVersion(env.v)); + } + Ok(env) +} + +/// Reads into `buf`, retrying on `Interrupted`. Returns `Ok(0)` on a clean end of stream. +fn read_bytes(r: &mut R, buf: &mut [u8]) -> std::io::Result { + loop { + match r.read(buf) { + Ok(0) => return Ok(0), + Ok(n) => return Ok(n), + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + } + } +} diff --git a/crates/proto/src/lib.rs b/crates/proto/src/lib.rs index 6bb0122..e388ae9 100644 --- a/crates/proto/src/lib.rs +++ b/crates/proto/src/lib.rs @@ -1,10 +1,12 @@ //! Shared data types and the frame codec for Boxmaker. No policy and no I/O beyond frames. pub mod class; +pub mod frame; pub mod ids; pub mod wire; pub use class::DataClass; +pub use frame::{FrameError, MAX_FRAME, read_frame, write_frame}; pub use ids::{CallId, Epoch, Hash32, SessionId, Timestamp, ValueError}; pub use wire::{ DenyReason, Envelope, ErrorCode, Message, PROTOCOL_VERSION, ToolRequest, ToolResponse, diff --git a/crates/proto/tests/fixtures/frame/tool_request.bin b/crates/proto/tests/fixtures/frame/tool_request.bin new file mode 100644 index 0000000000000000000000000000000000000000..fd3b2c143574b92ab6617a3c38be91f884169049 GIT binary patch literal 163 zcmW;ETMEK36a>(&Wxi924-m4JK#A>b8_Yvi(57Ich1aoUH9GbhCV5{8TzKM zxM)4Jgy)BHJ Vec { + let path = format!("{}/tests/fixtures/frame/{name}", env!("CARGO_MANIFEST_DIR")); + std::fs::read(&path).unwrap_or_else(|e| panic!("{path}: {e}")) +} + +fn request(arguments: String) -> Envelope { + let body = ToolRequest { + session: SessionId::new("mm-thread-42").unwrap(), + call: CallId(3), + tool: "read_file".to_string(), + arguments, + }; + Envelope { + v: 1, + id: 7, + r#final: true, + msg: Message::ToolRequest(body), + } +} + +/// Serves `head`, then panics if anyone reads further. +struct HeaderOnly { + head: Cursor>, +} + +impl Read for HeaderOnly { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + let n = self.head.read(buf)?; + assert!( + n > 0, + "the reader asked for the body of a frame it should have rejected" + ); + Ok(n) + } +} + +/// Hands out one byte per call, to catch codecs that assume `read` fills the buffer. +struct OneByteAtATime(Cursor>); + +impl Read for OneByteAtATime { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + let end = buf.len().min(1); + self.0.read(&mut buf[..end]) + } +} + +#[test] +fn max_frame_is_one_mebibyte() { + assert_eq!(MAX_FRAME, 1_048_576); +} + +#[test] +fn write_matches_the_fixture_byte_for_byte() { + let mut out = Vec::new(); + write_frame(&mut out, &request(r#"{"path":"/etc/hosts"}"#.to_string())).unwrap(); + assert_eq!(out, fixture_bytes("tool_request.bin")); + assert_eq!(&out[..4], &[0, 0, 0, 159]); +} + +#[test] +fn read_decodes_the_fixture() { + let mut input = Cursor::new(fixture_bytes("tool_request.bin")); + let env = read_frame(&mut input).unwrap(); + assert_eq!(env, request(r#"{"path":"/etc/hosts"}"#.to_string())); +} + +#[test] +fn two_frames_in_a_row_then_closed() { + let mut bytes = fixture_bytes("tool_request.bin"); + bytes.extend(fixture_bytes("tool_request.bin")); + let mut input = Cursor::new(bytes); + assert!(read_frame(&mut input).is_ok()); + assert!(read_frame(&mut input).is_ok()); + assert!(matches!(read_frame(&mut input), Err(FrameError::Closed))); +} + +#[test] +fn short_reads_are_handled() { + let mut input = OneByteAtATime(Cursor::new(fixture_bytes("tool_request.bin"))); + assert!(read_frame(&mut input).is_ok()); +} + +#[test] +fn zero_length_is_empty() { + let mut input = Cursor::new(vec![0, 0, 0, 0]); + assert!(matches!(read_frame(&mut input), Err(FrameError::Empty))); +} + +#[test] +fn oversize_length_is_rejected_without_reading_the_body() { + let n = (MAX_FRAME as u32) + 1; + let mut input = HeaderOnly { + head: Cursor::new(n.to_be_bytes().to_vec()), + }; + match read_frame(&mut input) { + Err(FrameError::TooLarge(got)) => assert_eq!(got, MAX_FRAME + 1), + other => panic!("expected TooLarge, got {other:?}"), + } + let mut input = HeaderOnly { + head: Cursor::new(vec![0xff, 0xff, 0xff, 0xff]), + }; + assert!(matches!( + read_frame(&mut input), + Err(FrameError::TooLarge(4_294_967_295)) + )); +} + +#[test] +fn exactly_max_frame_is_accepted_by_the_length_check() { + // A body of MAX_FRAME bytes of spaces is not valid JSON, so the error must be Json, not TooLarge. + let mut bytes = (MAX_FRAME as u32).to_be_bytes().to_vec(); + bytes.extend(std::iter::repeat_n(b' ', MAX_FRAME)); + assert!(matches!( + read_frame(&mut Cursor::new(bytes)), + Err(FrameError::Json(_)) + )); +} + +#[test] +fn truncated_header_and_body_are_io_errors() { + let full = fixture_bytes("tool_request.bin"); + let mut header_cut = Cursor::new(full[..2].to_vec()); + assert!(matches!( + read_frame(&mut header_cut), + Err(FrameError::Io(_)) + )); + let mut body_cut = Cursor::new(full[..full.len() - 1].to_vec()); + assert!(matches!(read_frame(&mut body_cut), Err(FrameError::Io(_)))); +} + +#[test] +fn garbage_body_is_a_json_error() { + let mut bytes = vec![0, 0, 0, 3]; + bytes.extend(b"{{{"); + assert!(matches!( + read_frame(&mut Cursor::new(bytes)), + Err(FrameError::Json(_)) + )); +} + +#[test] +fn other_protocol_versions_are_rejected() { + let mut env = request("{}".to_string()); + env.v = 2; + let mut out = Vec::new(); + write_frame(&mut out, &env).unwrap(); + assert!(matches!( + read_frame(&mut Cursor::new(out)), + Err(FrameError::BadVersion(2)) + )); +} + +#[test] +fn oversize_envelopes_are_not_written() { + let mut out = Vec::new(); + let err = write_frame(&mut out, &request("x".repeat(MAX_FRAME))).unwrap_err(); + assert!(matches!(err, FrameError::TooLarge(n) if n > MAX_FRAME)); + assert!( + out.is_empty(), + "nothing may be written when the envelope is too large" + ); +} + +#[test] +fn frame_error_is_a_std_error_with_a_message() { + let e: Box = Box::new(FrameError::Empty); + assert!(!e.to_string().is_empty()); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 9e307bf..2e88861 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -8,5 +8,6 @@ reviewer adds findings under "Reviews" once per milestone. | M1/01-workspace-and-gate | 2026-09-17 | done | 1 | pass | none | Crate skeletons, Cargo files and the given Makefile/deny.toml/test-gate-scripts.sh were already present untracked from a prior attempt; I verified them against the plan and created only the missing gate scripts, dependencies.md, egress.md and this log row. | | M1/02-proto-values | 2026-09-17 | done | 1 | pass | none | Implemented ValueError, SessionId, Epoch, CallId, Hash32 and Timestamp in crates/proto/src/ids.rs and DataClass in class.rs, using serde try_from/into for string-backed JSON validation, a hand-written hex encoder and humantime for RFC 3339 parsing with canonical re-serialization. | | M1/03-proto-wire | 2026-09-17 | done | 2 | pass | none | Added Envelope, Message, WireError, ErrorCode, ToolRequest, ToolResponse and DenyReason in crates/proto/src/wire.rs, re-exported from lib.rs; all 9 fixture tests pass and `make gate` prints `gate: ok`. | +| M1/04-proto-frame | 2026-09-17 | done | 2 | fail | none | Added crates/proto/src/frame.rs (MAX_FRAME, FrameError, write_frame, read_frame) re-exported from lib.rs; 13 fixture tests pass. Two compile fixes: mapped read_bytes io::Error to FrameError::Io and annotated serde_json::from_slice::; cargo-fmt reordered the lib.rs re-export lines; `make gate` prints `gate: ok`. | ## Reviews