dig_peer/
rpc.rs

1//! The peer-stream RPC framing — how a typed JSON-RPC call rides one mux stream.
2//!
3//! `dig-rpc` is an HTTP JSON-RPC *server*; the peer surface instead carries JSON-RPC over `dig-nat`'s
4//! multiplexed mTLS streams. dig-peer defines that on-stream framing here and is the CLIENT that
5//! speaks it. Each call opens ONE fresh logical stream, writes a single length-prefixed request body,
6//! reads a single length-prefixed response body, and lets the stream close — a clean request/response
7//! per stream, with concurrency provided by the mux (open many streams).
8//!
9//! ## Framing (normative)
10//!
11//! A body is a `u32` big-endian length prefix followed by that many bytes — the SAME uniform framing
12//! `dig-nat`'s control messages use, so the two never disagree. For an **unsealed** (public-read)
13//! call the body is the JSON of a `JsonRpcRequest`/`JsonRpcResponse`. For a **directed** (sealed)
14//! call the body is the byte-serialized sealed [`dig_message`] envelope wrapping that JSON (§5.4).
15//! The [`MAX_BODY`] bound guards against a malicious length prefix forcing a huge allocation.
16
17use serde::de::DeserializeOwned;
18use serde::Serialize;
19use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
20
21use crate::error::{DigPeerError, Result};
22
23/// Maximum length-prefixed body dig-peer will read — guards against a hostile length prefix. Matches
24/// dig-nat's control-frame bound (64 KiB) for small control/RPC messages.
25pub const MAX_BODY: usize = 64 * 1024;
26
27/// Write a length-prefixed body (`u32` big-endian length + bytes) to `w`.
28pub async fn write_framed<W: AsyncWrite + Unpin>(w: &mut W, body: &[u8]) -> Result<()> {
29    if body.len() > MAX_BODY {
30        return Err(DigPeerError::Codec(format!(
31            "outbound body {} exceeds the {MAX_BODY}-byte bound",
32            body.len()
33        )));
34    }
35    w.write_all(&(body.len() as u32).to_be_bytes()).await?;
36    w.write_all(body).await?;
37    w.flush().await?;
38    Ok(())
39}
40
41/// Read one length-prefixed body (`u32` big-endian length + bytes) from `r`, bounded by [`MAX_BODY`].
42pub async fn read_framed<R: AsyncRead + Unpin>(r: &mut R) -> Result<Vec<u8>> {
43    let mut len_buf = [0u8; 4];
44    r.read_exact(&mut len_buf).await?;
45    let len = u32::from_be_bytes(len_buf) as usize;
46    if len > MAX_BODY {
47        return Err(DigPeerError::Codec(format!(
48            "inbound body length {len} exceeds the {MAX_BODY}-byte bound"
49        )));
50    }
51    let mut body = vec![0u8; len];
52    r.read_exact(&mut body).await?;
53    Ok(body)
54}
55
56/// Serialize a value to a JSON body for the wire.
57pub fn to_json<T: Serialize>(value: &T) -> Result<Vec<u8>> {
58    serde_json::to_vec(value).map_err(|e| DigPeerError::Codec(e.to_string()))
59}
60
61/// Deserialize a JSON body from the wire into a typed value.
62pub fn from_json<T: DeserializeOwned>(bytes: &[u8]) -> Result<T> {
63    serde_json::from_slice(bytes).map_err(|e| DigPeerError::Codec(e.to_string()))
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69
70    /// **Proves:** a body written by [`write_framed`] is read back byte-identically by
71    /// [`read_framed`] — the length-prefixed framing is self-consistent.
72    #[tokio::test]
73    async fn framed_body_round_trips() {
74        let body = b"{\"jsonrpc\":\"2.0\"}".to_vec();
75        let mut buf = Vec::new();
76        write_framed(&mut buf, &body).await.expect("write");
77        let mut cursor = std::io::Cursor::new(buf);
78        let read = read_framed(&mut cursor).await.expect("read");
79        assert_eq!(read, body);
80    }
81
82    /// **Proves:** an empty body frames + unframes cleanly (a zero-length prefix is valid).
83    #[tokio::test]
84    async fn empty_body_round_trips() {
85        let mut buf = Vec::new();
86        write_framed(&mut buf, &[]).await.expect("write");
87        let mut cursor = std::io::Cursor::new(buf);
88        assert!(read_framed(&mut cursor).await.expect("read").is_empty());
89    }
90
91    /// **Proves:** writing a body over the [`MAX_BODY`] bound is refused, guarding against a caller
92    /// serializing an oversized payload.
93    #[tokio::test]
94    async fn oversized_write_is_refused() {
95        let big = vec![0u8; MAX_BODY + 1];
96        let mut buf = Vec::new();
97        let result = write_framed(&mut buf, &big).await;
98        assert!(matches!(result, Err(DigPeerError::Codec(_))));
99    }
100
101    /// **Proves:** a length prefix over [`MAX_BODY`] is rejected before allocating, guarding against a
102    /// hostile prefix forcing a huge allocation.
103    #[tokio::test]
104    async fn oversized_length_prefix_is_rejected() {
105        let mut framed = ((MAX_BODY + 1) as u32).to_be_bytes().to_vec();
106        framed.extend_from_slice(&[0u8; 8]);
107        let mut cursor = std::io::Cursor::new(framed);
108        assert!(matches!(
109            read_framed(&mut cursor).await,
110            Err(DigPeerError::Codec(_))
111        ));
112    }
113
114    /// **Proves:** [`to_json`]/[`from_json`] round-trip a value; a malformed body is a `Codec` error.
115    #[test]
116    fn json_round_trips_and_rejects_garbage() {
117        let value = serde_json::json!({"a": 1});
118        let bytes = to_json(&value).expect("to_json");
119        let back: serde_json::Value = from_json(&bytes).expect("from_json");
120        assert_eq!(value, back);
121        assert!(matches!(
122            from_json::<serde_json::Value>(b"not json"),
123            Err(DigPeerError::Codec(_))
124        ));
125    }
126}
dig_peer/rpc.rs

dig_peer/
rpc.rs
