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

actordb_protocol/
lib.rs
