macula_rust/
frame.rs

1//! macula 12's frames, as macula_frame and macula-go build and read them: the
2//! requests, replies, relay errors, publications and stream frames that carry
3//! signed objects, the control frames a pq_hybrid link neighbour-signs, the
4//! decoding rule's payload bounds, and the length-prefixed wire codec.
5//!
6//! A wire frame is `<Length:4 bytes big-endian><Cbor>`, the deterministic
7//! encoding of one map with `version` and `frame_type`. No frame carries a
8//! frame-level signature: what is signed is the signed object a frame holds,
9//! and, in pq_hybrid, a control frame's neighbour signature.
10
11mod check_payload;
12mod neighbour;
13mod publication;
14mod reply;
15mod request;
16mod stream;
17
18pub use check_payload::{
19    check_frame, check_payload, FRAME_RESERVED_ELEMENTS, MAX_PAYLOAD_ELEMENTS, MAX_PAYLOAD_NESTING,
20};
21pub use neighbour::{
22    advertise_frame, goodbye_frame, neighbour_signed, sign_neighbour, subscribe_frame,
23    unadvertise_frame, unsubscribe_frame, verify_neighbour, NeighbourLink, NeighbourPeer,
24};
25pub use publication::{sign_publish, verify_publication, PublicationSpec, VerifiedPublication};
26pub use reply::{
27    claimed_reply_ids, sign_provider_error, sign_relay_error, sign_result, verify_relay_error,
28    verify_reply, RelayErrorSpec, RelayErrorType, ReplyType, VerifiedRelayError, VerifiedReply,
29};
30pub use request::{
31    request_fields_accepted, sign_call, sign_stream_open, verify_request, RequestSpec, RequestType,
32    VerifiedRequest, MAX_PROOFS, MAX_PROOFS_BYTES,
33};
34pub use stream::{
35    open_stream, sign_caller_stream, sign_provider_stream, verify_caller_stream,
36    verify_provider_stream, StreamEncoding, StreamFields, StreamMode, StreamRole, StreamState,
37    VerifiedStreamFrame,
38};
39
40use std::fmt;
41
42use crate::cbor::{self, Value};
43use crate::node_key::{KeyError, NodeKey, Purpose};
44use crate::signed_object::ObjectError;
45
46/// The version field every frame carries.
47pub const PROTOCOL_VERSION: i64 = 2;
48
49/// The CBOR payload size cap: 16 MiB minus one byte, as macula's.
50pub const MAX_FRAME_BYTES: usize = 0x00FF_FFFF;
51
52/// The labels of the signed objects frames carry (D25, D17).
53const REQUEST_LABEL: &str = "MACULA-PQ-REQUEST-V1";
54const REPLY_LABEL: &str = "MACULA-PQ-REPLY-V1";
55const RELAY_ERROR_LABEL: &str = "MACULA-PQ-RELAY-ERROR-V1";
56const STREAM_LABEL: &str = "MACULA-PQ-STREAM-V1";
57const CALLER_STREAM_LABEL: &str = "MACULA-PQ-CALLER-STREAM-V1";
58const PUBLICATION_LABEL: &str = "MACULA-PQ-PUBLICATION-V1";
59
60/// A protocol integer stays below 2^53; a procedure name is at most 512
61/// bytes, an error code at most 64, and an error's text at most 256.
62const MAX_PROTOCOL_INT: u64 = 1 << 53;
63const MAX_PROCEDURE_BYTES: usize = 512;
64const MAX_ERROR_CODE_BYTES: usize = 64;
65const MAX_ERROR_TEXT_BYTES: usize = 256;
66const MAX_TOPIC_BYTES: usize = 512;
67
68/// The refusals of a frame, named as macula_frame names them.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub enum FrameError {
71    /// An encoding longer than [`MAX_FRAME_BYTES`], or a header claiming one.
72    TooLarge(usize),
73    /// A frame, or the signed object it carries, without exactly the shape and
74    /// fields of its type.
75    Malformed,
76    /// A payload the decoding rule would refuse where it arrives, and where.
77    Payload(String),
78    /// A whole frame the decoding rule would refuse, and where.
79    BreaksDecodingRule(String),
80    /// A request's delegation chain proofs outside their bound.
81    ProofsOutOfBound,
82    /// A signed object whose signer is not the key it verified with.
83    KeyIdMismatch,
84    /// A reply, relay error or stream frame naming another request.
85    RequestMismatch,
86    /// A reply, or a provider's first stream frame, from a node other than its
87    /// request's target.
88    NotTheTarget,
89    /// A relay error from another station than the connection's.
90    NotTheConnection,
91    /// A key that cannot sign this frame.
92    Unsignable,
93    /// A text longer than its bound, naming the field.
94    TextTooLong(String),
95    /// A text that is not valid UTF-8, naming the field.
96    InvalidText(String),
97    /// A relay error code outside its closed set.
98    RelayCodeOutsideItsSet,
99    /// A field outside its range, and which.
100    OutOfRange(String),
101    /// A stream frame its side does not send, and which.
102    NotAllowed(String),
103    /// A stream frame out of its side's order.
104    SeqMismatch,
105    /// A stream frame after its side's STREAM_END.
106    StreamEnded,
107    /// A frame given to sign that already carries a neighbour signature.
108    NeighbourSigned,
109    /// A publication published too far ahead, by how many milliseconds.
110    NotYetValid(i64),
111    /// A publication past its expiry, by how many milliseconds.
112    Expired(i64),
113    /// A signed object's signature that does not verify.
114    SignatureInvalid,
115    /// A key that could not sign.
116    Key(KeyError),
117}
118
119impl fmt::Display for FrameError {
120    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
121        match self {
122            FrameError::TooLarge(n) => write!(
123                f,
124                "a frame of {n} bytes, over the {MAX_FRAME_BYTES}-byte cap"
125            ),
126            FrameError::Malformed => f.write_str("malformed frame"),
127            FrameError::Payload(why) | FrameError::BreaksDecodingRule(why) => f.write_str(why),
128            FrameError::ProofsOutOfBound => {
129                f.write_str("the request's proofs are outside the bound")
130            }
131            FrameError::KeyIdMismatch => {
132                f.write_str("the signer the frame names is not the key it verified with")
133            }
134            FrameError::RequestMismatch => f.write_str("the frame names another request"),
135            FrameError::NotTheTarget => f.write_str("the frame is not from the request's target"),
136            FrameError::NotTheConnection => {
137                f.write_str("the relay error is not from the connection's station")
138            }
139            FrameError::Unsignable => f.write_str("the key cannot sign this frame"),
140            FrameError::TextTooLong(what) => write!(f, "text longer than its bound: {what}"),
141            FrameError::InvalidText(what) => write!(f, "text that is not valid UTF-8: {what}"),
142            FrameError::RelayCodeOutsideItsSet => {
143                f.write_str("a relay error code outside its closed set")
144            }
145            FrameError::OutOfRange(what) => write!(f, "a field outside its range: {what}"),
146            FrameError::NotAllowed(what) => {
147                write!(f, "a stream frame its side does not send: {what}")
148            }
149            FrameError::SeqMismatch => f.write_str("a stream frame out of its side's order"),
150            FrameError::StreamEnded => f.write_str("a stream frame after its side's STREAM_END"),
151            FrameError::NeighbourSigned => {
152                f.write_str("the frame already carries a neighbour signature")
153            }
154            FrameError::NotYetValid(ms) => write!(f, "a publication not yet valid, by {ms} ms"),
155            FrameError::Expired(ms) => write!(f, "a publication past its expiry, by {ms} ms"),
156            FrameError::SignatureInvalid => {
157                f.write_str("the signed object's signature does not verify")
158            }
159            FrameError::Key(e) => write!(f, "{e}"),
160        }
161    }
162}
163
164impl std::error::Error for FrameError {}
165
166/// The refusal of a frame whose signed object did not verify: a signature
167/// that does not verify as it is, anything else as [`FrameError::Malformed`].
168fn object_refusal(e: ObjectError) -> FrameError {
169    match e {
170        ObjectError::SignatureInvalid => FrameError::SignatureInvalid,
171        ObjectError::Key(k) => FrameError::Key(k),
172        _ => FrameError::Malformed,
173    }
174}
175
176/// Wraps `frame` as `<Length:4 bytes big-endian><Cbor>`, refusing one over
177/// the frame cap.
178pub fn encode(frame: &Value) -> Result<Vec<u8>, FrameError> {
179    let payload = cbor::encode(frame).map_err(|e| FrameError::Payload(e.to_string()))?;
180    if payload.len() > MAX_FRAME_BYTES {
181        return Err(FrameError::TooLarge(payload.len()));
182    }
183    let mut out = Vec::with_capacity(4 + payload.len());
184    out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
185    out.extend_from_slice(&payload);
186    Ok(out)
187}
188
189/// What decoding the head of a buffer found.
190#[derive(Debug, Clone, PartialEq)]
191pub enum Decoded {
192    /// A whole frame, and how many bytes of the buffer it took.
193    Complete { frame: Value, consumed: usize },
194    /// At least this many more bytes are needed before trying again.
195    NeedMore(usize),
196}
197
198/// Decodes one length-prefixed frame from the head of `buf`, under the
199/// decoding rule.
200pub fn decode(buf: &[u8]) -> Result<Decoded, FrameError> {
201    let Some((header, rest)) = buf.split_first_chunk::<4>() else {
202        return Ok(Decoded::NeedMore(4 - buf.len()));
203    };
204    let length = u32::from_be_bytes(*header) as usize;
205    if length > MAX_FRAME_BYTES {
206        return Err(FrameError::TooLarge(length));
207    }
208    if rest.len() < length {
209        return Ok(Decoded::NeedMore(length - rest.len()));
210    }
211    let frame = cbor::decode(&rest[..length]).map_err(|_| FrameError::Malformed)?;
212    Ok(Decoded::Complete {
213        frame,
214        consumed: 4 + length,
215    })
216}
217
218/// A field's rule, as a frame's field table names it.
219#[derive(Debug, Clone, Copy)]
220enum Rule {
221    Any,
222    AnyBytes,
223    BytesOf(usize),
224    TextWithin(usize),
225    TextIn(&'static [&'static str]),
226    ProtocolUint,
227    ProtocolVersion,
228    CarriedObject,
229    HeldObject,
230    StreamObject,
231    Proofs,
232}
233
234impl Rule {
235    fn accepts(self, v: &Value) -> bool {
236        match self {
237            Rule::Any => true,
238            Rule::AnyBytes => matches!(v, Value::Bytes(_)),
239            Rule::BytesOf(n) => matches!(v, Value::Bytes(b) if b.len() == n),
240            Rule::TextWithin(n) => matches!(v, Value::Text(t) if t.len() <= n),
241            Rule::TextIn(names) => matches!(v, Value::Text(t) if names.contains(&t.as_str())),
242            Rule::ProtocolUint => protocol_uint(v).is_some(),
243            Rule::ProtocolVersion => {
244                matches!(v, Value::Int(n) if *n == i128::from(PROTOCOL_VERSION))
245            }
246            Rule::CarriedObject => crate::signed_object::Object::from_value(v).is_ok(),
247            Rule::HeldObject => crate::signed_object::HeldObject::from_value(v).is_ok(),
248            Rule::StreamObject => Rule::CarriedObject.accepts(v) || Rule::HeldObject.accepts(v),
249            Rule::Proofs => request::proofs_within_bound(v),
250        }
251    }
252}
253
254/// A frame's or signed object's fields by name.
255type Fields = std::collections::HashMap<String, Value>;
256
257/// A map read through its table, as macula_frame's read_fields does: every
258/// key text, named in the table and there once, with a value its rule
259/// accepts.
260fn read_fields(v: &Value, table: &[(&str, Rule)]) -> Option<Fields> {
261    let Value::Map(pairs) = v else {
262        return None;
263    };
264    let mut fields = Fields::with_capacity(pairs.len());
265    for (key, value) in pairs {
266        let Value::Text(name) = key else {
267            return None;
268        };
269        let rule = table.iter().find(|(n, _)| *n == name)?.1;
270        if fields.contains_key(name) || !rule.accepts(value) {
271            return None;
272        }
273        fields.insert(name.clone(), value.clone());
274    }
275    Some(fields)
276}
277
278fn has_fields(fields: &Fields, names: &[&str]) -> bool {
279    names.iter().all(|n| fields.contains_key(*n))
280}
281
282/// A received frame that carries its fields in one signed object: exactly
283/// version, frame_type, the object under `object_name` and the routing fields
284/// `routes` names; the protocol's version; a frame type of `types`; the object
285/// in a shape `object_rule` accepts; and each routing field of its rule.
286fn received_frame(
287    v: &Value,
288    object_name: &str,
289    object_rule: Rule,
290    routes: &[(&str, Rule)],
291    types: &'static [&'static str],
292) -> Option<(String, Value)> {
293    let mut table = vec![
294        ("version", Rule::ProtocolVersion),
295        ("frame_type", Rule::TextIn(types)),
296        (object_name, object_rule),
297    ];
298    table.extend_from_slice(routes);
299    let fields = read_fields(v, &table)?;
300    if !has_fields(&fields, &["version", "frame_type", object_name]) {
301        return None;
302    }
303    Some((text_of(&fields["frame_type"]), fields[object_name].clone()))
304}
305
306/// Refuses text longer than `max` bytes, judged first, then text that is not
307/// UTF-8, naming the field.
308fn bounded_text(field: &str, text: &[u8], max: usize) -> Result<(), FrameError> {
309    if text.len() > max {
310        return Err(FrameError::TextTooLong(format!(
311            "a {field} of {} bytes, over {max}",
312            text.len()
313        )));
314    }
315    if std::str::from_utf8(text).is_err() {
316        return Err(FrameError::InvalidText(format!("the {field}")));
317    }
318    Ok(())
319}
320
321/// Refuses a key that is not an identity key.
322fn identity_signer(key: &NodeKey) -> Result<(), FrameError> {
323    if key.purpose() != Purpose::Identity {
324        return Err(FrameError::Unsignable);
325    }
326    Ok(())
327}
328
329fn protocol_uint(v: &Value) -> Option<u64> {
330    match v {
331        Value::Int(n) if *n >= 0 && *n < i128::from(MAX_PROTOCOL_INT) => Some(*n as u64),
332        _ => None,
333    }
334}
335
336fn text_of(v: &Value) -> String {
337    match v {
338        Value::Text(t) => t.clone(),
339        _ => String::new(),
340    }
341}
342
343fn bytes_of(v: &Value) -> Vec<u8> {
344    match v {
345        Value::Bytes(b) => b.clone(),
346        _ => Vec::new(),
347    }
348}
349
350fn fixed<const N: usize>(v: &Value) -> [u8; N] {
351    let mut out = [0u8; N];
352    if let Value::Bytes(b) = v {
353        if b.len() == N {
354            out.copy_from_slice(b);
355        }
356    }
357    out
358}
359
360fn entry(name: &str, value: Value) -> (Value, Value) {
361    (Value::text(name), value)
362}
363
364fn uint(n: u64) -> Value {
365    Value::Int(i128::from(n))
366}
367
368/// Whether a reply's, relay error's or stream frame's request_id and
369/// request_hash are `request`'s.
370fn names_request(fields: &Fields, request: &VerifiedRequest) -> bool {
371    fields.get("request_id") == Some(&Value::Bytes(request.request_id.to_vec()))
372        && fields.get("request_hash") == Some(&Value::Bytes(request.request_hash.to_vec()))
373}
374
375/// The envelope a control frame carries, as macula_frame's base/2: version,
376/// frame_type, a fresh frame_id (UUID v7), sent_at_ms, capabilities, and the
377/// null realm, call_id and source_route.
378fn base(frame_type: &str) -> Vec<(Value, Value)> {
379    vec![
380        entry("version", Value::Int(i128::from(PROTOCOL_VERSION))),
381        entry("frame_type", Value::text(frame_type)),
382        entry("frame_id", Value::Bytes(crate::uuid_v7::new().to_vec())),
383        entry("sent_at_ms", uint(crate::uuid_v7::now_ms())),
384        entry("capabilities", uint(0)),
385        entry("realm", Value::Null),
386        entry("call_id", Value::Null),
387        entry("source_route", Value::Null),
388    ]
389}
390
391/// Replaces `fields`' entry for `key`, or appends one: a raw push over a base
392/// field would put two entries under one key.
393fn with_field(mut fields: Vec<(Value, Value)>, key: &str, value: Value) -> Vec<(Value, Value)> {
394    match fields.iter_mut().find(|(k, _)| *k == Value::text(key)) {
395        Some(slot) => slot.1 = value,
396        None => fields.push(entry(key, value)),
397    }
398    fields
399}
macula_rust/frame.rs

macula_rust/
frame.rs
