native_ipc/
control.rs

1use crate::session::HARD_MAX_CONTROL_PAYLOAD_BYTES;
2use core::cell::Cell;
3use core::marker::PhantomData;
4
5const MAGIC: [u8; 8] = *b"NIPCAPP1";
6const VERSION_MAJOR: u16 = 1;
7const VERSION_MINOR: u16 = 0;
8pub(crate) const CONTROL_HEADER_LEN: usize = 72;
9/// Lowest application-owned control kind; smaller values are library-reserved.
10pub const APPLICATION_CONTROL_KIND_MIN: u32 = 0x8000_0000;
11
12pub(crate) const fn control_wire_len(payload_len: usize) -> Option<usize> {
13    CONTROL_HEADER_LEN.checked_add(payload_len)
14}
15
16/// One bounded opaque application-control record received from the peer.
17#[derive(Debug, Eq, PartialEq)]
18pub struct ControlFrame {
19    pub(crate) kind: u32,
20    pub(crate) payload: Vec<u8>,
21}
22
23impl ControlFrame {
24    /// Application-owned record kind.
25    pub const fn kind(&self) -> u32 {
26        self.kind
27    }
28
29    /// Opaque record payload.
30    pub fn payload(&self) -> &[u8] {
31        &self.payload
32    }
33
34    /// Consumes the frame and returns its opaque payload allocation.
35    pub fn into_payload(self) -> Vec<u8> {
36        self.payload
37    }
38}
39
40pub(crate) struct ControlState {
41    nonce: [u8; 32],
42    maximum_payload: u32,
43    next_send: u64,
44    next_receive: u64,
45    transaction_open: bool,
46    poisoned: bool,
47    pending_receive: bool,
48    not_sync: PhantomData<Cell<()>>,
49}
50
51impl ControlState {
52    pub(crate) fn new(nonce: [u8; 32], maximum_payload: u32) -> Option<Self> {
53        if nonce == [0; 32]
54            || maximum_payload == 0
55            || maximum_payload > HARD_MAX_CONTROL_PAYLOAD_BYTES
56        {
57            return None;
58        }
59        Some(Self {
60            nonce,
61            maximum_payload,
62            next_send: 1,
63            next_receive: 1,
64            transaction_open: false,
65            poisoned: false,
66            pending_receive: false,
67            not_sync: PhantomData,
68        })
69    }
70
71    #[cfg(test)]
72    pub(crate) fn encoded_len(&self, frame: &ControlFrame) -> Result<usize, ControlError> {
73        self.encoded_len_parts(frame.kind, frame.payload.len())
74    }
75
76    pub(crate) fn encoded_len_parts(
77        &self,
78        kind: u32,
79        payload_len: usize,
80    ) -> Result<usize, ControlError> {
81        self.check_local_parts(kind, payload_len)?;
82        control_wire_len(payload_len).ok_or(ControlError::LengthOverflow)
83    }
84
85    #[cfg(test)]
86    pub(crate) fn encode_into(
87        &mut self,
88        frame: &ControlFrame,
89        destination: &mut [u8],
90    ) -> Result<usize, ControlError> {
91        self.encode_parts_into(frame.kind, &frame.payload, destination)
92    }
93
94    pub(crate) fn encode_parts_into(
95        &mut self,
96        kind: u32,
97        payload: &[u8],
98        destination: &mut [u8],
99    ) -> Result<usize, ControlError> {
100        if self.transaction_open || self.pending_receive {
101            self.poisoned = true;
102            return Err(ControlError::TransactionConflict);
103        }
104        if self.next_send == u64::MAX {
105            self.poisoned = true;
106            return Err(ControlError::SequenceExhausted);
107        }
108        let required = self.encoded_len_parts(kind, payload.len())?;
109        if destination.len() < required {
110            return Err(ControlError::DestinationTooSmall);
111        }
112        // The transport preallocates the exact wire length, but this encoder also
113        // accepts reused caller buffers. Every byte is written below except the
114        // canonical reserved field, which must therefore be explicitly cleared.
115        destination[28..32].fill(0);
116        destination[..8].copy_from_slice(&MAGIC);
117        put_u16(destination, 8, VERSION_MAJOR);
118        put_u16(destination, 10, VERSION_MINOR);
119        put_u32(destination, 12, CONTROL_HEADER_LEN as u32);
120        put_u32(
121            destination,
122            16,
123            u32::try_from(required).map_err(|_| ControlError::LengthOverflow)?,
124        );
125        put_u32(
126            destination,
127            20,
128            u32::try_from(payload.len()).map_err(|_| ControlError::LengthOverflow)?,
129        );
130        put_u32(destination, 24, kind);
131        destination[32..64].copy_from_slice(&self.nonce);
132        put_u64(destination, 64, self.next_send);
133        destination[CONTROL_HEADER_LEN..required].copy_from_slice(payload);
134        self.next_send = self
135            .next_send
136            .checked_add(1)
137            .ok_or(ControlError::SequenceExhausted)?;
138        Ok(required)
139    }
140
141    #[cfg(test)]
142    pub(crate) fn decode(&mut self, source: &[u8]) -> Result<ControlFrame, ControlError> {
143        if source.len() < CONTROL_HEADER_LEN {
144            self.poisoned = true;
145            return Err(ControlError::Truncated);
146        }
147        let header = self.validate_header(&source[..CONTROL_HEADER_LEN])?;
148        let expected =
149            control_wire_len(header.payload_len()).ok_or(ControlError::LengthOverflow)?;
150        if source.len() != expected {
151            return Err(ControlError::NonCanonical);
152        }
153        header.finish(&source[CONTROL_HEADER_LEN..])
154    }
155
156    pub(crate) fn decode_owned(
157        &mut self,
158        mut source: Vec<u8>,
159    ) -> Result<ControlFrame, ControlError> {
160        if source.len() < CONTROL_HEADER_LEN {
161            self.poisoned = true;
162            return Err(ControlError::Truncated);
163        }
164        let header = self.validate_header(&source[..CONTROL_HEADER_LEN])?;
165        let payload_len = header.payload_len();
166        let expected = control_wire_len(payload_len).ok_or(ControlError::LengthOverflow)?;
167        if source.len() != expected {
168            return Err(ControlError::NonCanonical);
169        }
170        source.copy_within(CONTROL_HEADER_LEN..expected, 0);
171        source.truncate(payload_len);
172        header.finish_owned(source)
173    }
174
175    pub(crate) fn validate_header(
176        &mut self,
177        source: &[u8],
178    ) -> Result<PendingControlReceive<'_>, ControlError> {
179        let result = self.validate_header_inner(source);
180        if result.is_err() {
181            self.poisoned = true;
182        } else {
183            self.pending_receive = true;
184        }
185        result.map(|header| PendingControlReceive {
186            state: self,
187            kind: header.kind,
188            payload_len: header.payload_len,
189            committed: false,
190        })
191    }
192
193    pub(crate) fn begin_transaction(&mut self) -> Result<(), ControlError> {
194        if self.poisoned {
195            return Err(ControlError::Poisoned);
196        }
197        if self.transaction_open || self.pending_receive {
198            self.poisoned = true;
199            return Err(ControlError::TransactionConflict);
200        }
201        self.transaction_open = true;
202        Ok(())
203    }
204
205    pub(crate) fn end_transaction(&mut self) -> Result<(), ControlError> {
206        if self.poisoned {
207            return Err(ControlError::Poisoned);
208        }
209        if !self.transaction_open {
210            self.poisoned = true;
211            return Err(ControlError::TransactionConflict);
212        }
213        self.transaction_open = false;
214        Ok(())
215    }
216
217    pub(crate) fn poison(&mut self) {
218        self.poisoned = true;
219    }
220
221    pub(crate) const fn is_poisoned(&self) -> bool {
222        self.poisoned
223    }
224
225    pub(crate) const fn is_transaction_open(&self) -> bool {
226        self.transaction_open
227    }
228
229    fn check_local_parts(&self, kind: u32, payload_len: usize) -> Result<(), ControlError> {
230        if self.poisoned {
231            return Err(ControlError::Poisoned);
232        }
233        if self.transaction_open {
234            return Err(ControlError::TransactionConflict);
235        }
236        if kind < APPLICATION_CONTROL_KIND_MIN {
237            return Err(ControlError::ReservedKind);
238        }
239        let payload_len = u32::try_from(payload_len).map_err(|_| ControlError::LengthOverflow)?;
240        if payload_len > self.maximum_payload || payload_len > HARD_MAX_CONTROL_PAYLOAD_BYTES {
241            return Err(ControlError::PayloadTooLarge);
242        }
243        if self.next_send == u64::MAX {
244            return Err(ControlError::SequenceExhausted);
245        }
246        Ok(())
247    }
248
249    fn validate_header_inner(&self, source: &[u8]) -> Result<ValidatedControlHeader, ControlError> {
250        if self.poisoned {
251            return Err(ControlError::Poisoned);
252        }
253        if self.transaction_open {
254            return Err(ControlError::TransactionConflict);
255        }
256        if self.pending_receive {
257            return Err(ControlError::ReplayOrReorder);
258        }
259        if source.len() != CONTROL_HEADER_LEN {
260            return Err(ControlError::Truncated);
261        }
262        if source[..8] != MAGIC {
263            return Err(ControlError::BadMagic);
264        }
265        if get_u16(source, 8) != VERSION_MAJOR || get_u16(source, 10) != VERSION_MINOR {
266            return Err(ControlError::BadVersion);
267        }
268        if get_u32(source, 12) != CONTROL_HEADER_LEN as u32 || get_u32(source, 28) != 0 {
269            return Err(ControlError::NonCanonical);
270        }
271        let frame_len = get_u32(source, 16) as usize;
272        let payload_len = get_u32(source, 20);
273        let expected_len =
274            control_wire_len(payload_len as usize).ok_or(ControlError::LengthOverflow)?;
275        if frame_len != expected_len {
276            return Err(ControlError::NonCanonical);
277        }
278        if payload_len > self.maximum_payload || payload_len > HARD_MAX_CONTROL_PAYLOAD_BYTES {
279            return Err(ControlError::PayloadTooLarge);
280        }
281        let kind = get_u32(source, 24);
282        if kind < APPLICATION_CONTROL_KIND_MIN {
283            return Err(ControlError::ReservedKind);
284        }
285        if source[32..64] != self.nonce {
286            return Err(ControlError::WrongSession);
287        }
288        if get_u64(source, 64) != self.next_receive {
289            return Err(ControlError::ReplayOrReorder);
290        }
291        if self.next_receive == u64::MAX {
292            return Err(ControlError::SequenceExhausted);
293        }
294        Ok(ValidatedControlHeader { kind, payload_len })
295    }
296}
297
298pub(crate) struct ValidatedControlHeader {
299    kind: u32,
300    payload_len: u32,
301}
302
303pub(crate) struct PendingControlReceive<'a> {
304    state: &'a mut ControlState,
305    kind: u32,
306    payload_len: u32,
307    committed: bool,
308}
309
310impl PendingControlReceive<'_> {
311    pub(crate) const fn payload_len(&self) -> usize {
312        self.payload_len as usize
313    }
314
315    #[cfg(test)]
316    pub(crate) fn finish(self, payload_source: &[u8]) -> Result<ControlFrame, ControlError> {
317        if payload_source.len() != self.payload_len as usize {
318            return Err(ControlError::NonCanonical);
319        }
320        let mut payload = Vec::new();
321        payload
322            .try_reserve_exact(self.payload_len as usize)
323            .map_err(|_| ControlError::AllocationFailed)?;
324        payload.extend_from_slice(payload_source);
325        self.finish_owned(payload)
326    }
327
328    pub(crate) fn finish_owned(mut self, payload: Vec<u8>) -> Result<ControlFrame, ControlError> {
329        if payload.len() != self.payload_len as usize {
330            return Err(ControlError::NonCanonical);
331        }
332        self.state.next_receive = self
333            .state
334            .next_receive
335            .checked_add(1)
336            .ok_or(ControlError::SequenceExhausted)?;
337        self.state.pending_receive = false;
338        self.committed = true;
339        Ok(ControlFrame {
340            kind: self.kind,
341            payload,
342        })
343    }
344}
345
346impl Drop for PendingControlReceive<'_> {
347    fn drop(&mut self) {
348        if !self.committed {
349            self.state.pending_receive = false;
350            self.state.poisoned = true;
351        }
352    }
353}
354
355/// Bounded application-control validation or state failure.
356#[derive(Clone, Copy, Debug, Eq, PartialEq)]
357pub enum ControlError {
358    /// The session is terminally poisoned.
359    Poisoned,
360    /// The peer record ended before its canonical length.
361    Truncated,
362    /// The peer record did not use the application-control magic.
363    BadMagic,
364    /// The peer selected an unsupported control wire version.
365    BadVersion,
366    /// The peer record was not the one canonical encoding.
367    NonCanonical,
368    /// The peer record carried another session nonce.
369    WrongSession,
370    /// The local or peer kind is reserved for library framing.
371    ReservedKind,
372    /// The payload exceeds the negotiated finite maximum.
373    PayloadTooLarge,
374    /// The peer sequence was stale, future, replayed, or reordered.
375    ReplayOrReorder,
376    /// Application control conflicted with an open native transaction.
377    TransactionConflict,
378    /// A direction exhausted its monotonic sequence space.
379    SequenceExhausted,
380    /// A checked wire-length computation overflowed.
381    LengthOverflow,
382    /// The bounded receive or encode allocation failed.
383    AllocationFailed,
384    /// An internal caller supplied less than the precomputed wire length.
385    DestinationTooSmall,
386}
387
388impl core::fmt::Display for ControlError {
389    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
390        write!(formatter, "application control failed: {self:?}")
391    }
392}
393
394impl std::error::Error for ControlError {}
395
396fn put_u16(bytes: &mut [u8], offset: usize, value: u16) {
397    bytes[offset..offset + 2].copy_from_slice(&value.to_le_bytes());
398}
399
400fn put_u32(bytes: &mut [u8], offset: usize, value: u32) {
401    bytes[offset..offset + 4].copy_from_slice(&value.to_le_bytes());
402}
403
404fn put_u64(bytes: &mut [u8], offset: usize, value: u64) {
405    bytes[offset..offset + 8].copy_from_slice(&value.to_le_bytes());
406}
407
408fn get_u16(bytes: &[u8], offset: usize) -> u16 {
409    u16::from_le_bytes(
410        bytes[offset..offset + 2]
411            .try_into()
412            .expect("fixed checked range"),
413    )
414}
415
416fn get_u32(bytes: &[u8], offset: usize) -> u32 {
417    u32::from_le_bytes(
418        bytes[offset..offset + 4]
419            .try_into()
420            .expect("fixed checked range"),
421    )
422}
423
424fn get_u64(bytes: &[u8], offset: usize) -> u64 {
425    u64::from_le_bytes(
426        bytes[offset..offset + 8]
427            .try_into()
428            .expect("fixed checked range"),
429    )
430}
431
432#[cfg(test)]
433#[path = "control_test.rs"]
434mod tests;
native_ipc/control.rs

native_ipc/
control.rs
