yaiba_sync/
proto.rs

1//! Wire protocol for a sync round.
2//!
3//! One bidirectional stream carries a three-message exchange that leaves
4//! both sides holding the union of each other's writes:
5//!
6//! ```text
7//!   dialer  ──── Hello { room, vv_a } ────▶  listener
8//!           ◀─── Offer { vv_b, entries } ──          (what A is missing)
9//!           ──── Push  { entries } ───────▶          (what B is missing)
10//! ```
11//!
12//! Both directions travel in a single round trip because each side sends
13//! its version vector before the other has to decide what to transmit.
14
15use anyhow::{Context, Result, bail};
16use serde::de::DeserializeOwned;
17use serde::{Deserialize, Serialize};
18use yaiba_core::{Entry, VersionVector};
19
20/// Frames larger than this are refused rather than allocated. A full
21/// dataset of a few thousand tasks is well under a megabyte of JSON;
22/// anything near the cap is a bug or a hostile peer.
23const MAX_FRAME: usize = 32 * 1024 * 1024;
24
25#[derive(Debug, Serialize, Deserialize)]
26pub struct Hello {
27    /// Hex-encoded room key. Peers that don't share it are dropped
28    /// before any data is exchanged.
29    pub room: String,
30    pub vv: VersionVector,
31}
32
33#[derive(Debug, Serialize, Deserialize)]
34pub struct Offer {
35    pub vv: VersionVector,
36    pub entries: Vec<Entry>,
37}
38
39#[derive(Debug, Serialize, Deserialize)]
40pub struct Push {
41    pub entries: Vec<Entry>,
42}
43
44/// Length-prefixed JSON. JSON rather than a compact codec because sync
45/// traffic is tiny and being able to read a capture during development
46/// is worth more than the bytes.
47pub async fn write_frame<T, W>(writer: &mut W, value: &T) -> Result<()>
48where
49    T: Serialize,
50    W: tokio::io::AsyncWrite + Unpin,
51{
52    use tokio::io::AsyncWriteExt;
53    let bytes = serde_json::to_vec(value).context("encode frame")?;
54    if bytes.len() > MAX_FRAME {
55        bail!("frame too large to send: {} bytes", bytes.len());
56    }
57    writer
58        .write_all(&(bytes.len() as u32).to_be_bytes())
59        .await?;
60    writer.write_all(&bytes).await?;
61    writer.flush().await?;
62    Ok(())
63}
64
65pub async fn read_frame<T, R>(reader: &mut R) -> Result<T>
66where
67    T: DeserializeOwned,
68    R: tokio::io::AsyncRead + Unpin,
69{
70    use tokio::io::AsyncReadExt;
71    let mut len = [0u8; 4];
72    reader.read_exact(&mut len).await.context("read length")?;
73    let len = u32::from_be_bytes(len) as usize;
74    if len > MAX_FRAME {
75        bail!("peer announced an oversized frame: {len} bytes");
76    }
77    let mut buf = vec![0u8; len];
78    reader.read_exact(&mut buf).await.context("read body")?;
79    serde_json::from_slice(&buf).context("decode frame")
80}
81
82/// Constant-time-ish comparison of the shared room key.
83///
84/// The room key is the only authorisation in the protocol, so the
85/// comparison must not leak where two keys diverge. `subtle` would be
86/// the rigorous answer; for a 32-byte key over a network round trip this
87/// fold is enough to remove the timing signal.
88pub fn room_matches(a: &str, b: &str) -> bool {
89    if a.len() != b.len() {
90        return false;
91    }
92    a.bytes()
93        .zip(b.bytes())
94        .fold(0u8, |acc, (x, y)| acc | (x ^ y))
95        == 0
96}
97
98#[cfg(test)]
99mod tests {
100    use super::*;
101
102    #[tokio::test]
103    async fn frames_round_trip() {
104        let hello = Hello {
105            room: "abc".into(),
106            vv: VersionVector::new(),
107        };
108        let mut buf = Vec::new();
109        write_frame(&mut buf, &hello).await.unwrap();
110
111        let mut cursor = std::io::Cursor::new(buf);
112        let back: Hello = read_frame(&mut cursor).await.unwrap();
113        assert_eq!(back.room, "abc");
114    }
115
116    #[tokio::test]
117    async fn an_oversized_length_prefix_is_refused_before_allocating() {
118        let mut framed = Vec::new();
119        framed.extend_from_slice(&u32::MAX.to_be_bytes());
120        let mut cursor = std::io::Cursor::new(framed);
121        let err = read_frame::<Hello, _>(&mut cursor).await.unwrap_err();
122        assert!(err.to_string().contains("oversized"));
123    }
124
125    #[test]
126    fn room_comparison_rejects_mismatches_and_length_changes() {
127        assert!(room_matches("deadbeef", "deadbeef"));
128        assert!(!room_matches("deadbeef", "deadbeee"));
129        assert!(!room_matches("deadbeef", "deadbee"));
130    }
131}
yaiba_sync/proto.rs

yaiba_sync/
proto.rs
