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 = 2;
26pub const CLIENT_CAPABILITY_QUERY: u64 = 1;
27pub const CLIENT_CAPABILITY_SCAN: u64 = 2;
28pub const CLIENT_CAPABILITY_WRITE_CONCERNS: u64 = 4;
29pub const CLIENT_CAPABILITIES: u64 =
30    CLIENT_CAPABILITY_QUERY | CLIENT_CAPABILITY_SCAN | CLIENT_CAPABILITY_WRITE_CONCERNS;
31pub const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
32pub const MAX_BATCH_ITEMS: usize = 4096;
33pub const MAX_HANDSHAKE_BYTES: usize = 4096;
34
35/// Canonical challenge proof. Client credentials use their own domain and never
36/// authenticate a peer. Both identities, offers, and fresh challenges are bound.
37pub fn client_auth_proof(
38    secret: &[u8],
39    dialer: bool,
40    client: &client::Hello,
41    server: &client::Hello,
42) -> Vec<u8> {
43    use hmac::{KeyInit, Mac};
44    let mut mac =
45        hmac::Hmac::<sha2::Sha256>::new_from_slice(secret).expect("HMAC accepts any key size");
46    mac.update(b"actordb-client-proof-v1\0");
47    mac.update(if dialer { b"client" } else { b"server" });
48    for hello in [client, server] {
49        mac.update(&(hello.identity.len() as u64).to_be_bytes());
50        mac.update(hello.identity.as_bytes());
51        mac.update(&hello.min_version.to_be_bytes());
52        mac.update(&hello.max_version.to_be_bytes());
53        mac.update(&hello.capabilities.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.min_version != 0
91        && hello.min_version <= CLIENT_PROTOCOL_VERSION
92        && hello.max_version >= CLIENT_PROTOCOL_VERSION
93}
94
95/// Allocation-free protobuf preflight. Bounds repeated fields before Prost
96/// materializes empty nested messages. Rejects duplicate envelope/operation
97/// variants rather than accepting protobuf's last-one-wins interpretation.
98pub fn preflight_client_frame(bytes: &[u8]) -> Result<(), &'static str> {
99    #[derive(Clone, Copy, PartialEq, Eq)]
100    enum Context {
101        Frame,
102        Request,
103        BatchGet,
104        BatchPut,
105        Query,
106        Response,
107        QueryResult,
108        Row,
109        Inspection,
110        Health,
111        Placement,
112        Cell,
113        Put,
114        Delete,
115        MutationResult,
116    }
117    fn varint(bytes: &[u8], pos: &mut usize) -> Result<u64, &'static str> {
118        let mut result = 0;
119        for shift in (0..70).step_by(7) {
120            let byte = *bytes.get(*pos).ok_or("truncated varint")?;
121            *pos += 1;
122            if shift == 63 && byte > 1 {
123                return Err("overflowing varint");
124            }
125            result |= u64::from(byte & 127) << shift;
126            if byte & 128 == 0 {
127                return Ok(result);
128            }
129        }
130        Err("overflowing varint")
131    }
132    fn walk(bytes: &[u8], context: Context) -> Result<(), &'static str> {
133        use Context::*;
134        let mut pos = 0;
135        let mut counts = [0usize; 10];
136        let mut variants = 0;
137        while pos < bytes.len() {
138            let tag = varint(bytes, &mut pos)?;
139            let field = tag >> 3;
140            if field == 0 {
141                return Err("invalid field");
142            }
143            if (context == Frame && (1..=4).contains(&field))
144                || (context == Request && (2..=9).contains(&field))
145                || (context == Cell && (1..=4).contains(&field))
146            {
147                variants += 1;
148                if variants > 1 || (context != Cell && tag & 7 != 2) {
149                    return Err("duplicate or invalid message");
150                }
151            }
152            let limit = match (context, field) {
153                (BatchGet | BatchPut, 1) | (Query, 2) | (Response, 3 | 4) => Some(MAX_BATCH_ITEMS),
154                (Response, 7) | (QueryResult, 2) | (Inspection, 4 | 5) => Some(100_000),
155                (QueryResult, 1) | (Row, 2) => Some(32),
156                (Health, 3 | 4) | (Placement, 2) => Some(MAX_BATCH_ITEMS),
157                _ => None,
158            };
159            if let Some(limit) = limit {
160                let counter = counts
161                    .get_mut(field as usize)
162                    .ok_or("invalid repeated field")?;
163                *counter += 1;
164                if *counter > limit || tag & 7 != 2 {
165                    return Err("invalid or excessive repeated field");
166                }
167            }
168            if context == Response && [5, 6].contains(&field) {
169                counts[field as usize] += 1;
170                if counts[field as usize] > 1 {
171                    return Err("duplicate structured response");
172                }
173            }
174            match tag & 7 {
175                0 => {
176                    varint(bytes, &mut pos)?;
177                }
178                1 => {
179                    pos = pos.checked_add(8).ok_or("overflow")?;
180                }
181                5 => {
182                    pos = pos.checked_add(4).ok_or("overflow")?;
183                }
184                2 => {
185                    let length =
186                        usize::try_from(varint(bytes, &mut pos)?).map_err(|_| "overflow")?;
187                    let end = pos.checked_add(length).ok_or("overflow")?;
188                    let inner = bytes.get(pos..end).ok_or("truncated field")?;
189                    if matches!((context, field), (Put, 5) | (Delete, 3))
190                        && inner.len() > write_concern::MAX_WRITE_CONCERN_POLICY_BYTES
191                    {
192                        return Err("write concern policy exceeds byte limit");
193                    }
194                    if matches!((context, field), (MutationResult, 3)) {
195                        write_concern::preflight_result(inner)
196                            .map_err(|_| "invalid write concern receipt")?;
197                    }
198                    let nested = match (context, field) {
199                        (Frame, 3) => Some(Request),
200                        (Frame, 4) => Some(Response),
201                        (Request, 5) => Some(BatchGet),
202                        (Request, 3) | (BatchPut, 1) => Some(Put),
203                        (Request, 4) => Some(Delete),
204                        (Response, 4) => Some(MutationResult),
205                        (Request, 6) => Some(BatchPut),
206                        (Request, 7) => Some(Query),
207                        (Query, 2) | (Row, 2) => Some(Cell),
208                        (Response, 5) => Some(QueryResult),
209                        (Response, 6) => Some(Inspection),
210                        (QueryResult, 2) => Some(Row),
211                        (Inspection, 4) => Some(Health),
212                        (Inspection, 5) => Some(Placement),
213                        _ => None,
214                    };
215                    if let Some(nested) = nested {
216                        walk(inner, nested)?;
217                    }
218                    pos = end;
219                }
220                _ => return Err("invalid wire type"),
221            }
222            if pos > bytes.len() {
223                return Err("truncated field");
224            }
225        }
226        if context == Frame && variants != 1 {
227            return Err("missing message");
228        }
229        Ok(())
230    }
231    walk(bytes, Context::Frame)
232}
actordb_protocol/lib.rs

actordb_protocol/
lib.rs
