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

macula_rust/
frame.rs
