actordb_protocol/
lib.rs

1//! Committed Prost bindings and stable limits for peer and application protocols.
2//!
3//! Regenerate with `cargo run -p actordb-protocol --features codegen --bin generate`.
4//! Append `-- --check` to verify schema/binding agreement without writing files.
5pub mod meta {
6    include!("generated/meta.rs");
7}
8#[allow(clippy::large_enum_variant)]
9pub mod client {
10    include!("generated/client.rs");
11}
12
13mod version;
14/// Per-write acknowledgement policy and progress receipts.
15pub mod write_concern;
16pub use version::RecordVersion;
17
18impl client::Value {
19    /// Physical/logical timestamp; compare writer identity after this pair.
20    pub fn version(&self) -> RecordVersion {
21        RecordVersion::new(self.timestamp_ms, self.logical_counter)
22    }
23}
24
25pub const CLIENT_PROTOCOL_VERSION: u32 = 4;
26pub const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
27pub const MAX_BATCH_ITEMS: usize = 4096;
28pub const MAX_HANDSHAKE_BYTES: usize = 4096;
29
30/// Canonical challenge proof. Client credentials use their own domain and never
31/// authenticate a peer. Both identities, offers, and fresh challenges are bound.
32pub fn client_auth_proof(
33    secret: &[u8],
34    dialer: bool,
35    client: &client::Hello,
36    server: &client::Hello,
37) -> Vec<u8> {
38    use hmac::{KeyInit, Mac};
39    let mut mac =
40        hmac::Hmac::<sha2::Sha256>::new_from_slice(secret).expect("HMAC accepts any key size");
41    mac.update(b"actordb-client-proof-v4\0");
42    mac.update(if dialer { b"client" } else { b"server" });
43    for hello in [client, server] {
44        mac.update(&(hello.identity.len() as u64).to_be_bytes());
45        mac.update(hello.identity.as_bytes());
46        mac.update(&hello.protocol_version.to_be_bytes());
47        mac.update(&(hello.challenge.len() as u64).to_be_bytes());
48        mac.update(&hello.challenge);
49    }
50    mac.finalize().into_bytes().to_vec()
51}
52
53pub fn verify_client_auth(
54    secret: Option<&[u8]>,
55    dialer: bool,
56    client: &client::Hello,
57    server: &client::Hello,
58    proof: &[u8],
59) -> bool {
60    match secret {
61        None => proof.is_empty(),
62        Some(secret) => {
63            // Compare using a MAC verification operation, avoiding data-dependent equality.
64            use hmac::{KeyInit, Mac};
65            let expected = client_auth_proof(secret, dialer, client, server);
66            let mut expected_mac = hmac::Hmac::<sha2::Sha256>::new_from_slice(secret)
67                .expect("HMAC accepts any key size");
68            expected_mac.update(&expected);
69            let tag = expected_mac.finalize().into_bytes();
70            let mut supplied_mac = hmac::Hmac::<sha2::Sha256>::new_from_slice(secret)
71                .expect("HMAC accepts any key size");
72            supplied_mac.update(proof);
73            supplied_mac.verify_slice(&tag).is_ok() && proof.len() == 32
74        }
75    }
76}
77
78pub fn valid_client_hello(hello: &client::Hello) -> bool {
79    !hello.identity.is_empty()
80        && hello.identity.len() <= 255
81        && hello.challenge.len() == 32
82        && hello.challenge.iter().any(|byte| *byte != 0)
83        && hello.protocol_version == CLIENT_PROTOCOL_VERSION
84}
85
86/// Allocation-free protobuf preflight. Bounds repeated fields before Prost
87/// materializes empty nested messages. Rejects duplicate envelope/operation
88/// variants rather than accepting protobuf's last-one-wins interpretation.
89pub fn preflight_client_frame(bytes: &[u8]) -> Result<(), &'static str> {
90    #[derive(Clone, Copy, PartialEq, Eq)]
91    enum Context {
92        Frame,
93        Request,
94        BatchGet,
95        BatchPut,
96        Query,
97        Response,
98        QueryResult,
99        Row,
100        Inspection,
101        Health,
102        Placement,
103        Cell,
104        Put,
105        Delete,
106        MutationResult,
107    }
108    fn varint(bytes: &[u8], pos: &mut usize) -> Result<u64, &'static str> {
109        let mut result = 0;
110        for shift in (0..70).step_by(7) {
111            let byte = *bytes.get(*pos).ok_or("truncated varint")?;
112            *pos += 1;
113            if shift == 63 && byte > 1 {
114                return Err("overflowing varint");
115            }
116            result |= u64::from(byte & 127) << shift;
117            if byte & 128 == 0 {
118                return Ok(result);
119            }
120        }
121        Err("overflowing varint")
122    }
123    fn walk(bytes: &[u8], context: Context) -> Result<(), &'static str> {
124        use Context::*;
125        let mut pos = 0;
126        let mut counts = [0usize; 10];
127        let mut variants = 0;
128        while pos < bytes.len() {
129            let tag = varint(bytes, &mut pos)?;
130            let field = tag >> 3;
131            if field == 0 {
132                return Err("invalid field");
133            }
134            if (context == Frame && (1..=4).contains(&field))
135                || (context == Request && (2..=9).contains(&field))
136                || (context == Cell && (1..=4).contains(&field))
137            {
138                variants += 1;
139                if variants > 1 || (context != Cell && tag & 7 != 2) {
140                    return Err("duplicate or invalid message");
141                }
142            }
143            let limit = match (context, field) {
144                (BatchGet | BatchPut, 1) | (Query, 2) | (Response, 3 | 4) => Some(MAX_BATCH_ITEMS),
145                (Response, 7) | (QueryResult, 2) | (Inspection, 4 | 5) => Some(100_000),
146                (QueryResult, 1) | (Row, 2) => Some(32),
147                (Health, 3 | 4) | (Placement, 2) => Some(MAX_BATCH_ITEMS),
148                _ => None,
149            };
150            if let Some(limit) = limit {
151                let counter = counts
152                    .get_mut(field as usize)
153                    .ok_or("invalid repeated field")?;
154                *counter += 1;
155                if *counter > limit || tag & 7 != 2 {
156                    return Err("invalid or excessive repeated field");
157                }
158            }
159            if context == Response && [5, 6].contains(&field) {
160                counts[field as usize] += 1;
161                if counts[field as usize] > 1 {
162                    return Err("duplicate structured response");
163                }
164            }
165            match tag & 7 {
166                0 => {
167                    varint(bytes, &mut pos)?;
168                }
169                1 => {
170                    pos = pos.checked_add(8).ok_or("overflow")?;
171                }
172                5 => {
173                    pos = pos.checked_add(4).ok_or("overflow")?;
174                }
175                2 => {
176                    let length =
177                        usize::try_from(varint(bytes, &mut pos)?).map_err(|_| "overflow")?;
178                    let end = pos.checked_add(length).ok_or("overflow")?;
179                    let inner = bytes.get(pos..end).ok_or("truncated field")?;
180                    if matches!((context, field), (Put, 5) | (Delete, 3))
181                        && inner.len() > write_concern::MAX_WRITE_CONCERN_POLICY_BYTES
182                    {
183                        return Err("write concern policy exceeds byte limit");
184                    }
185                    if matches!((context, field), (MutationResult, 3)) {
186                        write_concern::preflight_result(inner)
187                            .map_err(|_| "invalid write concern receipt")?;
188                    }
189                    let nested = match (context, field) {
190                        (Frame, 3) => Some(Request),
191                        (Frame, 4) => Some(Response),
192                        (Request, 5) => Some(BatchGet),
193                        (Request, 3) | (BatchPut, 1) => Some(Put),
194                        (Request, 4) => Some(Delete),
195                        (Response, 4) => Some(MutationResult),
196                        (Request, 6) => Some(BatchPut),
197                        (Request, 7) => Some(Query),
198                        (Query, 2) | (Row, 2) => Some(Cell),
199                        (Response, 5) => Some(QueryResult),
200                        (Response, 6) => Some(Inspection),
201                        (QueryResult, 2) => Some(Row),
202                        (Inspection, 4) => Some(Health),
203                        (Inspection, 5) => Some(Placement),
204                        _ => None,
205                    };
206                    if let Some(nested) = nested {
207                        walk(inner, nested)?;
208                    }
209                    pos = end;
210                }
211                _ => return Err("invalid wire type"),
212            }
213            if pos > bytes.len() {
214                return Err("truncated field");
215            }
216        }
217        if context == Frame && variants != 1 {
218            return Err("missing message");
219        }
220        Ok(())
221    }
222    walk(bytes, Context::Frame)
223}
actordb_protocol/lib.rs

actordb_protocol/
lib.rs
