buntrail_protocol/
frame.rs

1use crate::types::{Context, Header};
2pub const MAJOR: u32 = 1;
3pub const MINOR: u32 = 0;
4pub const MAX_HEADER: usize = 65536;
5pub const MAX_PAYLOAD: usize = 1048576;
6#[derive(Debug)]
7pub struct Frame {
8    pub header: Header,
9    pub payload: Vec<u8>,
10}
11#[derive(Default)]
12pub struct Decoder {
13    prefix: Vec<u8>,
14    header: Vec<u8>,
15    payload: Vec<u8>,
16    lengths: Option<(usize, usize)>,
17    parsed: Option<(Header, bool)>,
18    failed: bool,
19}
20
21impl Decoder {
22    pub fn buffered_payload_bytes(&self) -> usize {
23        self.payload.capacity()
24    }
25    pub fn push(
26        &mut self,
27        bytes: &[u8],
28        context: Option<&Context>,
29    ) -> Result<Vec<Frame>, &'static str> {
30        self.push_authorized(bytes, |candidate| Some(candidate) == context)
31    }
32    /// Authorize only contexts registered by the authenticated connection owner, never a header's own claims.
33    pub fn push_authorized(
34        &mut self,
35        bytes: &[u8],
36        authorize: impl Fn(&Context) -> bool,
37    ) -> Result<Vec<Frame>, &'static str> {
38        self.push_validated(bytes, |header, payload| {
39            let expected = header.context.as_ref().filter(|context| authorize(context));
40            header.validate(payload, expected)?;
41            Ok(true)
42        })
43    }
44    /// Helper requests: retired controls retain only the pinned bridge identity, never new-work authority.
45    pub fn push_helper_requests(
46        &mut self,
47        bytes: &[u8],
48        authorize: impl Fn(&Context) -> bool,
49        bridge: Option<&Context>,
50    ) -> Result<Vec<Frame>, &'static str> {
51        use crate::types::Message;
52        self.push_validated(bytes, |header, payload| {
53            let control = matches!(
54                header.message,
55                Message::Credit { .. } | Message::Cancel { .. }
56            );
57            let expected = header.context.as_ref().filter(|context| {
58                authorize(context)
59                    || (control
60                        && bridge.is_some_and(|bridge| {
61                            bridge.client_id == context.client_id
62                                && bridge.target_binding == context.target_binding
63                                && bridge.runtime_generation == context.runtime_generation
64                        }))
65            });
66            header.validate(payload, expected)?;
67            if control
68                && (header.sequence.is_some()
69                    || (matches!(header.message, Message::Cancel { .. })
70                        && (header.operation_id.is_some() || header.stream_id.is_some())))
71            {
72                return Err("invalid helper control identity");
73            }
74            Ok(true)
75        })
76    }
77    /// Client replies only: unknown request IDs are structurally checked and discarded,
78    /// never delivered or treated as authenticated state. Request decoders do not use this.
79    pub fn push_replies(
80        &mut self,
81        bytes: &[u8],
82        owner: impl Fn(u64) -> Option<(Context, Option<String>)>,
83    ) -> Result<Vec<Frame>, &'static str> {
84        use crate::types::Message;
85        self.push_validated(bytes, |header, payload| {
86            if !matches!(
87                header.message,
88                Message::ProofLookupReady { .. }
89                    | Message::ProofReady { .. }
90                    | Message::ProofOutputBegin { .. }
91                    | Message::ProofOutputChunk { .. }
92                    | Message::ProofOutputEnd { .. }
93                    | Message::ProofPrepared { .. }
94                    | Message::ProofExisting { .. }
95                    | Message::ProofMissing { .. }
96                    | Message::Ack { .. }
97                    | Message::Error { .. }
98                    | Message::Credit { .. }
99                    | Message::Receipt { .. }
100                    | Message::Cancellation { .. }
101                    | Message::ServiceEvent { .. }
102                    | Message::ReadBegin { .. }
103                    | Message::ReadChunk { .. }
104                    | Message::ReadEnd { .. }
105                    | Message::TerminalOutput { .. }
106            ) {
107                return Err("unexpected client reply direction");
108            }
109            if let Some((context, operation)) = owner(header.request_id) {
110                header.validate(payload, Some(&context))?;
111                if header.operation_id != operation {
112                    return Err("reply operation mismatch");
113                }
114                Ok(true)
115            } else {
116                header.validate(payload, header.context.as_ref())?;
117                Ok(false)
118            }
119        })
120    }
121    fn push_validated(
122        &mut self,
123        mut bytes: &[u8],
124        validate: impl Fn(&Header, usize) -> Result<bool, &'static str>,
125    ) -> Result<Vec<Frame>, &'static str> {
126        if bytes.len() > MAX_HEADER + MAX_PAYLOAD + 8 {
127            self.failed = true;
128            return Err("input window exceeds frame bound");
129        }
130        if self.failed {
131            return Err("decoder failed; close connection");
132        }
133        let result = (|| {
134            let mut frames = Vec::new();
135            while !bytes.is_empty() {
136                if self.lengths.is_none() {
137                    take(&mut self.prefix, &mut bytes, 8);
138                    if self.prefix.len() < 8 {
139                        break;
140                    }
141                    let h = u32::from_be_bytes(self.prefix[..4].try_into().unwrap()) as usize;
142                    let p = u32::from_be_bytes(self.prefix[4..].try_into().unwrap()) as usize;
143                    if h == 0 || h > MAX_HEADER || p > MAX_PAYLOAD {
144                        return Err("frame length exceeds limit");
145                    }
146                    self.lengths = Some((h, p));
147                }
148                let (h, p) = self.lengths.unwrap();
149                if self.parsed.is_none() {
150                    take(&mut self.header, &mut bytes, h);
151                    if self.header.len() < h {
152                        break;
153                    }
154                    let header: Header = serde_json::from_slice(&self.header)
155                        .map_err(|_| "invalid header schema")?;
156                    let deliver = validate(&header, p)?;
157                    self.parsed = Some((header, deliver));
158                }
159                take(&mut self.payload, &mut bytes, p);
160                if self.payload.len() < p {
161                    break;
162                }
163                let (header, deliver) = self.parsed.take().unwrap();
164                if deliver {
165                    frames.push(Frame {
166                        header,
167                        payload: std::mem::take(&mut self.payload),
168                    });
169                } else {
170                    self.payload.clear();
171                }
172                self.prefix.clear();
173                self.header.clear();
174                self.lengths = None;
175            }
176            Ok(frames)
177        })();
178        if result.is_err() {
179            self.failed = true;
180        }
181        result
182    }
183    pub fn finish(&self) -> Result<(), &'static str> {
184        if self.failed || !self.prefix.is_empty() {
185            Err("incomplete or invalid frame")
186        } else {
187            Ok(())
188        }
189    }
190}
191fn take(destination: &mut Vec<u8>, input: &mut &[u8], total: usize) {
192    let n = (total - destination.len()).min(input.len());
193    destination.extend_from_slice(&input[..n]);
194    *input = &input[n..];
195}
196
197/// Transfer integrity only: staging and durable commit remain the file service's responsibility.
198pub struct TransferVerifier {
199    id: String,
200    total: u64,
201    offset: u64,
202    hash: sha2::Sha256,
203    ended: bool,
204}
205impl TransferVerifier {
206    pub fn new(id: String, total: u64) -> Result<Self, &'static str> {
207        use sha2::Digest;
208        if !crate::types::uuid(&id) {
209            return Err("invalid transfer ID");
210        }
211        Ok(Self {
212            id,
213            total,
214            offset: 0,
215            hash: sha2::Sha256::new(),
216            ended: false,
217        })
218    }
219    pub fn chunk(&mut self, id: &str, offset: u64, bytes: &[u8]) -> Result<(), &'static str> {
220        use sha2::Digest;
221        if self.ended
222            || id != self.id
223            || offset != self.offset
224            || bytes.len() > 65536
225            || bytes.is_empty()
226            || bytes.len() as u64 > self.total - self.offset
227        {
228            return Err("invalid transfer chunk");
229        }
230        self.hash.update(bytes);
231        self.offset += bytes.len() as u64;
232        Ok(())
233    }
234    pub fn end(&mut self, id: &str, digest: [u8; 32]) -> Result<(), &'static str> {
235        use sha2::Digest;
236        if self.ended
237            || id != self.id
238            || self.offset != self.total
239            || self.hash.clone().finalize().as_slice() != digest
240        {
241            return Err("invalid transfer completion");
242        }
243        self.ended = true;
244        Ok(())
245    }
246}
247
248impl Frame {
249    pub fn encode(&self) -> Result<Vec<u8>, &'static str> {
250        let header = serde_json::to_vec(&self.header).map_err(|_| "invalid header")?;
251        if header.is_empty() || header.len() > MAX_HEADER || self.payload.len() > MAX_PAYLOAD {
252            return Err("frame length exceeds limit");
253        }
254        let mut bytes = Vec::with_capacity(8 + header.len() + self.payload.len());
255        bytes.extend((header.len() as u32).to_be_bytes());
256        bytes.extend((self.payload.len() as u32).to_be_bytes());
257        bytes.extend(header);
258        bytes.extend(&self.payload);
259        Ok(bytes)
260    }
261}
buntrail_protocol/frame.rs

buntrail_protocol/
frame.rs
