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
30mod client_decode;
31pub use client_decode::{
32    preflight_client_frame, preflight_client_frame_kind, preflight_client_frame_with_budget,
33    preflight_client_response, ClientDecodeReport, ClientFrameKind, ClientResponseShape,
34    DEFAULT_MAX_DECODED_RESPONSE_BYTES, PORTABLE_CELL_HEADER_CHARGE,
35};
36
37/// Canonical challenge proof. Client credentials use their own domain and never
38/// authenticate a peer. Both identities, offers, and fresh challenges are bound.
39pub fn client_auth_proof(
40    secret: &[u8],
41    dialer: bool,
42    client: &client::Hello,
43    server: &client::Hello,
44) -> Vec<u8> {
45    use hmac::{KeyInit, Mac};
46    let mut mac =
47        hmac::Hmac::<sha2::Sha256>::new_from_slice(secret).expect("HMAC accepts any key size");
48    mac.update(b"actordb-client-proof-v4\0");
49    mac.update(if dialer { b"client" } else { b"server" });
50    for hello in [client, server] {
51        mac.update(&(hello.identity.len() as u64).to_be_bytes());
52        mac.update(hello.identity.as_bytes());
53        mac.update(&hello.protocol_version.to_be_bytes());
54        mac.update(&(hello.challenge.len() as u64).to_be_bytes());
55        mac.update(&hello.challenge);
56    }
57    mac.finalize().into_bytes().to_vec()
58}
59
60pub fn verify_client_auth(
61    secret: Option<&[u8]>,
62    dialer: bool,
63    client: &client::Hello,
64    server: &client::Hello,
65    proof: &[u8],
66) -> bool {
67    match secret {
68        None => proof.is_empty(),
69        Some(secret) => {
70            // Compare using a MAC verification operation, avoiding data-dependent equality.
71            use hmac::{KeyInit, Mac};
72            let expected = client_auth_proof(secret, dialer, client, server);
73            let mut expected_mac = hmac::Hmac::<sha2::Sha256>::new_from_slice(secret)
74                .expect("HMAC accepts any key size");
75            expected_mac.update(&expected);
76            let tag = expected_mac.finalize().into_bytes();
77            let mut supplied_mac = hmac::Hmac::<sha2::Sha256>::new_from_slice(secret)
78                .expect("HMAC accepts any key size");
79            supplied_mac.update(proof);
80            supplied_mac.verify_slice(&tag).is_ok() && proof.len() == 32
81        }
82    }
83}
84
85pub fn valid_client_hello(hello: &client::Hello) -> bool {
86    !hello.identity.is_empty()
87        && hello.identity.len() <= 255
88        && hello.challenge.len() == 32
89        && hello.challenge.iter().any(|byte| *byte != 0)
90        && hello.protocol_version == CLIENT_PROTOCOL_VERSION
91}
92
93/// Structural protobuf preflight. Bounds repeated fields before Prost
94/// materializes empty nested messages. Rejects duplicate envelope/operation
95/// variants rather than accepting protobuf's last-one-wins interpretation.
96/// Malformed typed receipts can allocate bounded error metadata on rejection.
97fn validate_client_frame_structure(bytes: &[u8]) -> Result<(), &'static str> {
98    #[derive(Clone, Copy, PartialEq, Eq)]
99    enum Context {
100        Frame,
101        Request,
102        BatchGet,
103        BatchPut,
104        Query,
105        Response,
106        QueryResult,
107        Row,
108        Inspection,
109        Health,
110        Placement,
111        Cell,
112        Put,
113        Delete,
114        MutationResult,
115    }
116    fn varint(bytes: &[u8], pos: &mut usize) -> Result<u64, &'static str> {
117        let mut result = 0;
118        for shift in (0..70).step_by(7) {
119            let byte = *bytes.get(*pos).ok_or("truncated varint")?;
120            *pos += 1;
121            if shift == 63 && byte > 1 {
122                return Err("overflowing varint");
123            }
124            result |= u64::from(byte & 127) << shift;
125            if byte & 128 == 0 {
126                return Ok(result);
127            }
128        }
129        Err("overflowing varint")
130    }
131    fn walk(bytes: &[u8], context: Context) -> Result<(), &'static str> {
132        use Context::*;
133        let mut pos = 0;
134        let mut counts = [0usize; 10];
135        let mut variants = 0;
136        while pos < bytes.len() {
137            let tag = varint(bytes, &mut pos)?;
138            let field = tag >> 3;
139            if field == 0 {
140                return Err("invalid field");
141            }
142            if (context == Frame && (1..=4).contains(&field))
143                || (context == Request && (2..=9).contains(&field))
144                || (context == Cell && (1..=4).contains(&field))
145            {
146                variants += 1;
147                if variants > 1 || (context != Cell && tag & 7 != 2) {
148                    return Err("duplicate or invalid message");
149                }
150            }
151            let limit = match (context, field) {
152                (BatchGet | BatchPut, 1) | (Query, 2) | (Response, 3 | 4) => Some(MAX_BATCH_ITEMS),
153                (Response, 7) | (QueryResult, 2) | (Inspection, 4 | 5) => Some(100_000),
154                (QueryResult, 1) | (Row, 2) => Some(32),
155                (Health, 3 | 4) | (Placement, 2) => Some(MAX_BATCH_ITEMS),
156                _ => None,
157            };
158            if let Some(limit) = limit {
159                let counter = counts
160                    .get_mut(field as usize)
161                    .ok_or("invalid repeated field")?;
162                *counter += 1;
163                if *counter > limit || tag & 7 != 2 {
164                    return Err("invalid or excessive repeated field");
165                }
166            }
167            if context == Response && [5, 6].contains(&field) {
168                counts[field as usize] += 1;
169                if counts[field as usize] > 1 {
170                    return Err("duplicate structured response");
171                }
172            }
173            match tag & 7 {
174                0 => {
175                    varint(bytes, &mut pos)?;
176                }
177                1 => {
178                    pos = pos.checked_add(8).ok_or("overflow")?;
179                }
180                5 => {
181                    pos = pos.checked_add(4).ok_or("overflow")?;
182                }
183                2 => {
184                    let length =
185                        usize::try_from(varint(bytes, &mut pos)?).map_err(|_| "overflow")?;
186                    let end = pos.checked_add(length).ok_or("overflow")?;
187                    let inner = bytes.get(pos..end).ok_or("truncated field")?;
188                    if matches!((context, field), (Put, 5) | (Delete, 3))
189                        && inner.len() > write_concern::MAX_WRITE_CONCERN_POLICY_BYTES
190                    {
191                        return Err("write concern policy exceeds byte limit");
192                    }
193                    if matches!((context, field), (MutationResult, 3)) {
194                        write_concern::preflight_result(inner)
195                            .map_err(|_| "invalid write concern receipt")?;
196                    }
197                    let nested = match (context, field) {
198                        (Frame, 3) => Some(Request),
199                        (Frame, 4) => Some(Response),
200                        (Request, 5) => Some(BatchGet),
201                        (Request, 3) | (BatchPut, 1) => Some(Put),
202                        (Request, 4) => Some(Delete),
203                        (Response, 4) => Some(MutationResult),
204                        (Request, 6) => Some(BatchPut),
205                        (Request, 7) => Some(Query),
206                        (Query, 2) | (Row, 2) => Some(Cell),
207                        (Response, 5) => Some(QueryResult),
208                        (Response, 6) => Some(Inspection),
209                        (QueryResult, 2) => Some(Row),
210                        (Inspection, 4) => Some(Health),
211                        (Inspection, 5) => Some(Placement),
212                        _ => None,
213                    };
214                    if let Some(nested) = nested {
215                        walk(inner, nested)?;
216                    }
217                    pos = end;
218                }
219                _ => return Err("invalid wire type"),
220            }
221            if pos > bytes.len() {
222                return Err("truncated field");
223            }
224        }
225        if context == Frame && variants != 1 {
226            return Err("missing message");
227        }
228        Ok(())
229    }
230    walk(bytes, Context::Frame)
231}
actordb_protocol/lib.rs

actordb_protocol/
lib.rs
