iroh_db_sync/
lib.rs

1//! Bounded reconciliation protocol layered over iroh and iroh-blobs.
2
3#![forbid(unsafe_code)]
4
5use std::collections::BTreeSet;
6
7use iroh_db_core::{AuthorId, CommitId, DomainId};
8
9/// The default and accepted maximum encoded control-frame size.
10pub const MAX_CONTROL_FRAME: usize = 1024 * 1024;
11/// Maximum immutable references carried by one control frame.
12pub const MAX_REFERENCES: usize = 4_096;
13/// Maximum aggregate immutable references accepted during one reconciliation round.
14pub const MAX_SESSION_REFERENCES: usize = 65_536;
15const WIRE_FORMAT_VERSION: u16 = 4;
16/// Oldest commit/sync protocol understood by this release.
17pub const MIN_PROTOCOL_VERSION: u16 = 4;
18/// Newest commit/sync protocol understood by this release.
19pub const MAX_PROTOCOL_VERSION: u16 = 4;
20const GOSSIP_VERSION: u16 = 1;
21/// Maximum encoded gossip announcement size.
22pub const MAX_GOSSIP_HINT: usize = 512;
23/// Maximum capability objects carried by one authenticated authority message.
24pub const MAX_CAPABILITY_CHAIN: usize = 9;
25/// Maximum objects in the exact successor chain carried by a control grant.
26pub const MAX_CONTROL_CAPABILITIES: usize = MAX_CAPABILITY_CHAIN;
27/// Maximum canonical capability-certificate size.
28pub const MAX_CAPABILITY_BYTES: usize = 4_096;
29
30/// One signed, content-free commit availability hint sent through iroh-gossip.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct GossipHint {
33    author: AuthorId,
34    commit: CommitId,
35    frontier_digest: [u8; 32],
36    counter: u64,
37    signature: [u8; 64],
38}
39
40impl GossipHint {
41    /// Constructs a hint after the caller signs [`Self::signing_bytes`].
42    pub const fn new(
43        author: AuthorId,
44        commit: CommitId,
45        frontier_digest: [u8; 32],
46        counter: u64,
47        signature: [u8; 64],
48    ) -> Self {
49        Self {
50            author,
51            commit,
52            frontier_digest,
53            counter,
54            signature,
55        }
56    }
57
58    /// Returns the announcing device identity.
59    pub const fn author(&self) -> AuthorId {
60        self.author
61    }
62
63    /// Returns one immutable commit advertised as available.
64    pub const fn commit(&self) -> CommitId {
65        self.commit
66    }
67
68    /// Returns the digest of the announcer's complete frontier.
69    pub const fn frontier_digest(&self) -> [u8; 32] {
70        self.frontier_digest
71    }
72
73    /// Returns the per-process monotonic replay counter.
74    pub const fn counter(&self) -> u64 {
75        self.counter
76    }
77
78    /// Returns the detached endpoint signature.
79    pub const fn signature(&self) -> &[u8; 64] {
80        &self.signature
81    }
82
83    /// Encodes the exact bytes covered by the gossip-specific signature context.
84    pub fn signing_bytes(&self) -> Result<Vec<u8>, SyncFrameError> {
85        encode_hint_content(self.author, self.commit, self.frontier_digest, self.counter)
86    }
87
88    /// Encodes the complete strict announcement.
89    pub fn encode_canonical(&self) -> Result<Vec<u8>, SyncFrameError> {
90        let bytes = minicbor::to_vec(GossipWire {
91            content: self.signing_bytes()?,
92            signature: self.signature,
93        })
94        .map_err(codec_error)?;
95        if bytes.len() > MAX_GOSSIP_HINT {
96            return Err(SyncFrameError::GossipHintTooLarge(bytes.len()));
97        }
98        Ok(bytes)
99    }
100
101    /// Strictly decodes one signed announcement.
102    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, SyncFrameError> {
103        if bytes.len() > MAX_GOSSIP_HINT {
104            return Err(SyncFrameError::GossipHintTooLarge(bytes.len()));
105        }
106        let mut decoder = minicbor::Decoder::new(bytes);
107        let wire: GossipWire = decoder.decode().map_err(codec_error)?;
108        if decoder.position() != bytes.len() {
109            return Err(SyncFrameError::NonCanonical);
110        }
111        let content: GossipContent = decode_exact(&wire.content)?;
112        if content.version != GOSSIP_VERSION {
113            return Err(SyncFrameError::UnsupportedVersion(content.version));
114        }
115        let hint = Self::new(
116            content.author,
117            content.commit,
118            content.frontier_digest.into(),
119            content.counter,
120            wire.signature,
121        );
122        if hint.signing_bytes()? != wire.content || hint.encode_canonical()? != bytes {
123            return Err(SyncFrameError::NonCanonical);
124        }
125        Ok(hint)
126    }
127}
128
129/// One canonical `/iroh-db/sync/1` control-plane message.
130#[derive(Clone, PartialEq, Eq)]
131pub enum SyncFrame {
132    /// Negotiates the protocol and binds a fresh session nonce.
133    Hello {
134        /// Random session nonce.
135        nonce: [u8; 32],
136        /// Largest control frame accepted by the sender.
137        max_frame: u32,
138        /// Oldest object protocol the sender can safely accept.
139        min_protocol: u16,
140        /// Newest object protocol the sender can safely accept.
141        max_protocol: u16,
142    },
143    /// Proves possession of authorization material for exactly one domain.
144    DomainRequest {
145        /// Requested security domain.
146        domain_id: DomainId,
147        /// Historical epoch selected for the nonce-bound proof.
148        epoch: u64,
149        /// Nonce-bound capability/key proof.
150        proof: [u8; 32],
151        /// Root-to-leaf capability evidence for the authenticated endpoint.
152        capability_chain: Vec<Vec<u8>>,
153    },
154    /// Advertises the durable authority position before data reconciliation.
155    AuthorityState {
156        /// Active domain epoch.
157        epoch: u64,
158        /// Accepted control sequence, equal to the epoch for control format v3.
159        sequence: u64,
160        /// Canonical control-head ID, absent at genesis.
161        head: Option<[u8; 32]>,
162    },
163    /// Delivers one signed successor and digest-bound key over authenticated QUIC.
164    ControlGrant {
165        /// Canonical signed [`iroh_db_security::ControlTransition`] bytes.
166        transition: Vec<u8>,
167        /// Symmetric key for the transition's new epoch.
168        epoch_key: [u8; 32],
169        /// Capability evidence required to validate transition authority.
170        capability_chain: Vec<Vec<u8>>,
171    },
172    /// Exchanges a signed control head as equivocation evidence without key material.
173    ControlEvidence {
174        /// Canonical signed [`iroh_db_security::ControlTransition`] bytes.
175        transition: Vec<u8>,
176    },
177    /// Advertises the bounded causal frontier used as dependency-discovery roots.
178    Frontier {
179        /// Canonically sorted unique commit IDs.
180        commits: Vec<CommitId>,
181    },
182    /// Requests missing immutable commit objects.
183    Need {
184        /// Canonically sorted unique commit IDs.
185        commits: Vec<CommitId>,
186    },
187    /// Announces immutable hashes available through iroh-blobs.
188    Refs {
189        /// Canonically sorted unique commit IDs.
190        commits: Vec<CommitId>,
191    },
192    /// Acknowledges durable acceptance of one commit.
193    Ack {
194        /// Accepted commit ID.
195        commit: CommitId,
196    },
197    /// Completes a successful reconciliation round.
198    Done,
199    /// Reports a bounded protocol error without sensitive context.
200    Error {
201        /// Stable protocol error code.
202        code: u16,
203    },
204    /// Requests an endpoint-targeted invitation before domain membership exists.
205    InvitationRequest,
206    /// Delivers signed invitation bytes inside the authenticated QUIC session.
207    Invitation {
208        /// Canonical signed invitation payload.
209        payload: Vec<u8>,
210    },
211}
212
213impl std::fmt::Debug for SyncFrame {
214    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215        match self {
216            Self::Hello { max_frame, .. } => formatter
217                .debug_struct("Hello")
218                .field("max_frame", max_frame)
219                .finish_non_exhaustive(),
220            Self::DomainRequest {
221                domain_id,
222                epoch,
223                capability_chain,
224                ..
225            } => formatter
226                .debug_struct("DomainRequest")
227                .field("domain_id", domain_id)
228                .field("epoch", epoch)
229                .field("capability_count", &capability_chain.len())
230                .finish_non_exhaustive(),
231            Self::AuthorityState {
232                epoch,
233                sequence,
234                head,
235            } => formatter
236                .debug_struct("AuthorityState")
237                .field("epoch", epoch)
238                .field("sequence", sequence)
239                .field("head", head)
240                .finish(),
241            Self::ControlGrant {
242                capability_chain, ..
243            } => formatter
244                .debug_struct("ControlGrant")
245                .field("capability_count", &capability_chain.len())
246                .field("key_material", &"[redacted]")
247                .finish_non_exhaustive(),
248            Self::ControlEvidence { transition } => formatter
249                .debug_struct("ControlEvidence")
250                .field("transition_len", &transition.len())
251                .finish(),
252            Self::Frontier { commits } => debug_references(formatter, "Frontier", commits),
253            Self::Need { commits } => debug_references(formatter, "Need", commits),
254            Self::Refs { commits } => debug_references(formatter, "Refs", commits),
255            Self::Ack { commit } => formatter
256                .debug_struct("Ack")
257                .field("commit", commit)
258                .finish(),
259            Self::Done => formatter.write_str("Done"),
260            Self::Error { code } => formatter.debug_struct("Error").field("code", code).finish(),
261            Self::InvitationRequest => formatter.write_str("InvitationRequest"),
262            Self::Invitation { payload } => formatter
263                .debug_struct("Invitation")
264                .field("payload_len", &payload.len())
265                .field("payload", &"[redacted]")
266                .finish(),
267        }
268    }
269}
270
271fn debug_references(
272    formatter: &mut std::fmt::Formatter<'_>,
273    name: &str,
274    commits: &[CommitId],
275) -> std::fmt::Result {
276    formatter
277        .debug_struct(name)
278        .field("commit_count", &commits.len())
279        .finish()
280}
281
282impl SyncFrame {
283    /// Encodes a strict, deterministic control frame.
284    pub fn encode_canonical(&self) -> Result<Vec<u8>, SyncFrameError> {
285        validate(self)?;
286        let wire = WireFrame::from(self);
287        let bytes = minicbor::to_vec(wire).map_err(codec_error)?;
288        if bytes.len() > MAX_CONTROL_FRAME {
289            return Err(SyncFrameError::FrameTooLarge(bytes.len()));
290        }
291        Ok(bytes)
292    }
293
294    /// Strictly decodes one bounded canonical control frame.
295    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, SyncFrameError> {
296        if bytes.len() > MAX_CONTROL_FRAME {
297            return Err(SyncFrameError::FrameTooLarge(bytes.len()));
298        }
299        let mut decoder = minicbor::Decoder::new(bytes);
300        let wire: WireFrame = decoder.decode().map_err(codec_error)?;
301        if decoder.position() != bytes.len() {
302            return Err(SyncFrameError::NonCanonical);
303        }
304        let frame = Self::try_from(wire)?;
305        if frame.encode_canonical()? != bytes {
306            return Err(SyncFrameError::NonCanonical);
307        }
308        Ok(frame)
309    }
310}
311
312/// Returns commit IDs accepted remotely but absent locally.
313pub fn missing_commits(
314    local: &[CommitId],
315    remote: &[CommitId],
316) -> Result<Vec<CommitId>, SyncFrameError> {
317    if local.len() > MAX_SESSION_REFERENCES {
318        return Err(SyncFrameError::TooManyReferences(local.len()));
319    }
320    if remote.len() > MAX_SESSION_REFERENCES {
321        return Err(SyncFrameError::TooManyReferences(remote.len()));
322    }
323    let local = local.iter().copied().collect::<BTreeSet<_>>();
324    Ok(remote
325        .iter()
326        .copied()
327        .collect::<BTreeSet<_>>()
328        .difference(&local)
329        .copied()
330        .collect())
331}
332
333/// Selects the newest protocol supported by both endpoints.
334pub fn negotiate_protocol(
335    local_min: u16,
336    local_max: u16,
337    remote_min: u16,
338    remote_max: u16,
339) -> Result<u16, SyncFrameError> {
340    if local_min > local_max {
341        return Err(SyncFrameError::InvalidProtocolRange {
342            min: local_min,
343            max: local_max,
344        });
345    }
346    if remote_min > remote_max {
347        return Err(SyncFrameError::InvalidProtocolRange {
348            min: remote_min,
349            max: remote_max,
350        });
351    }
352    let minimum = local_min.max(remote_min);
353    let maximum = local_max.min(remote_max);
354    if minimum > maximum {
355        return Err(SyncFrameError::NoCommonProtocol {
356            local_min,
357            local_max,
358            remote_min,
359            remote_max,
360        });
361    }
362    Ok(maximum)
363}
364
365/// A rejected control frame or reconciliation plan.
366#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
367pub enum SyncFrameError {
368    /// A frame exceeded the hard control-plane bound.
369    #[error("control frame is too large: {0}")]
370    FrameTooLarge(usize),
371    /// A frame exceeded the per-message immutable-reference bound.
372    #[error("control frame contains too many references: {0}")]
373    TooManyReferences(usize),
374    /// Immutable references were not sorted and unique.
375    #[error("control-frame references are not canonical")]
376    NonCanonicalReferences,
377    /// The encoded version is not supported.
378    #[error("unsupported sync protocol version: {0}")]
379    UnsupportedVersion(u16),
380    /// A hello advertised its version interval backwards.
381    #[error("invalid protocol range {min}..={max}")]
382    InvalidProtocolRange { min: u16, max: u16 },
383    /// The peers have no mutually understood object protocol.
384    #[error(
385        "no common protocol: local {local_min}..={local_max}, remote {remote_min}..={remote_max}"
386    )]
387    NoCommonProtocol {
388        local_min: u16,
389        local_max: u16,
390        remote_min: u16,
391        remote_max: u16,
392    },
393    /// The frame kind and its populated fields disagree.
394    #[error("control frame shape is invalid")]
395    InvalidShape,
396    /// CBOR was malformed or outside protocol types.
397    #[error("control frame codec failed: {0}")]
398    Codec(String),
399    /// Valid CBOR did not use the unique canonical representation.
400    #[error("control frame is not canonical")]
401    NonCanonical,
402    /// A gossip announcement exceeded its much smaller hard bound.
403    #[error("gossip hint is too large: {0}")]
404    GossipHintTooLarge(usize),
405    /// An invitation payload exceeded its 64 KiB bound.
406    #[error("invitation payload is too large: {0}")]
407    InvitationTooLarge(usize),
408    /// An authority message exceeded the maximum capability-chain depth.
409    #[error("authority message contains too many capability objects: {0}")]
410    TooManyAuthorityObjects(usize),
411    /// One canonical capability object exceeded its hard bound.
412    #[error("authority object is too large: {0}")]
413    AuthorityObjectTooLarge(usize),
414}
415
416#[derive(minicbor::Encode, minicbor::Decode)]
417#[cbor(array)]
418struct GossipContent {
419    #[n(0)]
420    version: u16,
421    #[n(1)]
422    author: AuthorId,
423    #[n(2)]
424    commit: CommitId,
425    #[n(3)]
426    frontier_digest: minicbor::bytes::ByteArray<32>,
427    #[n(4)]
428    counter: u64,
429}
430
431#[derive(minicbor::Encode, minicbor::Decode)]
432#[cbor(array)]
433struct GossipWire {
434    #[n(0)]
435    content: Vec<u8>,
436    #[n(1)]
437    signature: [u8; 64],
438}
439
440#[derive(minicbor::Encode, minicbor::Decode)]
441#[cbor(array)]
442struct WireFrame {
443    #[n(0)]
444    version: u16,
445    #[n(1)]
446    kind: u8,
447    #[n(2)]
448    domain_id: Option<DomainId>,
449    #[n(3)]
450    nonce_or_proof: Option<minicbor::bytes::ByteArray<32>>,
451    #[n(4)]
452    commits: Vec<CommitId>,
453    #[n(5)]
454    max_frame: Option<u32>,
455    #[n(6)]
456    commit: Option<CommitId>,
457    #[n(7)]
458    error_code: Option<u16>,
459    #[n(8)]
460    payload: Option<Vec<u8>>,
461    #[n(9)]
462    epoch: Option<u64>,
463    #[n(10)]
464    sequence: Option<u64>,
465    #[n(11)]
466    head: Option<minicbor::bytes::ByteArray<32>>,
467    #[n(12)]
468    payloads: Vec<Vec<u8>>,
469    #[n(13)]
470    epoch_key: Option<minicbor::bytes::ByteArray<32>>,
471    #[n(14)]
472    min_protocol: Option<u16>,
473    #[n(15)]
474    max_protocol: Option<u16>,
475}
476
477impl From<&SyncFrame> for WireFrame {
478    fn from(frame: &SyncFrame) -> Self {
479        let mut wire = Self {
480            version: WIRE_FORMAT_VERSION,
481            kind: 0,
482            domain_id: None,
483            nonce_or_proof: None,
484            commits: Vec::new(),
485            max_frame: None,
486            commit: None,
487            error_code: None,
488            payload: None,
489            epoch: None,
490            sequence: None,
491            head: None,
492            payloads: Vec::new(),
493            epoch_key: None,
494            min_protocol: None,
495            max_protocol: None,
496        };
497        match frame {
498            SyncFrame::Hello {
499                nonce,
500                max_frame,
501                min_protocol,
502                max_protocol,
503            } => {
504                wire.kind = 0;
505                wire.nonce_or_proof = Some((*nonce).into());
506                wire.max_frame = Some(*max_frame);
507                wire.min_protocol = Some(*min_protocol);
508                wire.max_protocol = Some(*max_protocol);
509            }
510            SyncFrame::DomainRequest {
511                domain_id,
512                epoch,
513                proof,
514                capability_chain,
515            } => {
516                wire.kind = 1;
517                wire.domain_id = Some(*domain_id);
518                wire.nonce_or_proof = Some((*proof).into());
519                wire.epoch = Some(*epoch);
520                wire.payloads.clone_from(capability_chain);
521            }
522            SyncFrame::Frontier { commits } => {
523                wire.kind = 2;
524                wire.commits.clone_from(commits);
525            }
526            SyncFrame::Need { commits } => {
527                wire.kind = 3;
528                wire.commits.clone_from(commits);
529            }
530            SyncFrame::Refs { commits } => {
531                wire.kind = 4;
532                wire.commits.clone_from(commits);
533            }
534            SyncFrame::Ack { commit } => {
535                wire.kind = 5;
536                wire.commit = Some(*commit);
537            }
538            SyncFrame::Done => wire.kind = 6,
539            SyncFrame::Error { code } => {
540                wire.kind = 7;
541                wire.error_code = Some(*code);
542            }
543            SyncFrame::InvitationRequest => wire.kind = 8,
544            SyncFrame::Invitation { payload } => {
545                wire.kind = 9;
546                wire.payload = Some(payload.clone());
547            }
548            SyncFrame::AuthorityState {
549                epoch,
550                sequence,
551                head,
552            } => {
553                wire.kind = 10;
554                wire.epoch = Some(*epoch);
555                wire.sequence = Some(*sequence);
556                wire.head = head.map(Into::into);
557            }
558            SyncFrame::ControlGrant {
559                transition,
560                epoch_key,
561                capability_chain,
562            } => {
563                wire.kind = 11;
564                wire.payload = Some(transition.clone());
565                wire.epoch_key = Some((*epoch_key).into());
566                wire.payloads.clone_from(capability_chain);
567            }
568            SyncFrame::ControlEvidence { transition } => {
569                wire.kind = 12;
570                wire.payload = Some(transition.clone());
571            }
572        }
573        wire
574    }
575}
576
577impl TryFrom<WireFrame> for SyncFrame {
578    type Error = SyncFrameError;
579
580    fn try_from(wire: WireFrame) -> Result<Self, SyncFrameError> {
581        if wire.version != WIRE_FORMAT_VERSION {
582            return Err(SyncFrameError::UnsupportedVersion(wire.version));
583        }
584        if wire.kind != 0 && (wire.min_protocol.is_some() || wire.max_protocol.is_some()) {
585            return Err(SyncFrameError::InvalidShape);
586        }
587        if wire.kind >= 10 {
588            let frame = decode_authority_frame(wire)?;
589            validate(&frame)?;
590            return Ok(frame);
591        }
592        let empty_refs = wire.commits.is_empty();
593        let frame = match wire.kind {
594            0 if wire.domain_id.is_none()
595                && empty_refs
596                && wire.commit.is_none()
597                && wire.error_code.is_none()
598                && wire.payload.is_none() =>
599            {
600                Self::Hello {
601                    nonce: fixed_proof(wire.nonce_or_proof)?,
602                    max_frame: wire.max_frame.ok_or(SyncFrameError::InvalidShape)?,
603                    min_protocol: wire.min_protocol.ok_or(SyncFrameError::InvalidShape)?,
604                    max_protocol: wire.max_protocol.ok_or(SyncFrameError::InvalidShape)?,
605                }
606            }
607            1 if wire.max_frame.is_none()
608                && empty_refs
609                && wire.commit.is_none()
610                && wire.error_code.is_none()
611                && wire.payload.is_none()
612                && wire.sequence.is_none()
613                && wire.head.is_none()
614                && wire.epoch_key.is_none() =>
615            {
616                Self::DomainRequest {
617                    domain_id: wire.domain_id.ok_or(SyncFrameError::InvalidShape)?,
618                    epoch: wire.epoch.ok_or(SyncFrameError::InvalidShape)?,
619                    proof: fixed_proof(wire.nonce_or_proof)?,
620                    capability_chain: wire.payloads,
621                }
622            }
623            2 if unused_scalar_fields(&wire) => Self::Frontier {
624                commits: wire.commits,
625            },
626            3 if unused_scalar_fields(&wire) => Self::Need {
627                commits: wire.commits,
628            },
629            4 if unused_scalar_fields(&wire) => Self::Refs {
630                commits: wire.commits,
631            },
632            5 if wire.domain_id.is_none()
633                && wire.nonce_or_proof.is_none()
634                && empty_refs
635                && wire.max_frame.is_none()
636                && wire.error_code.is_none()
637                && wire.payload.is_none() =>
638            {
639                Self::Ack {
640                    commit: wire.commit.ok_or(SyncFrameError::InvalidShape)?,
641                }
642            }
643            6 if unused_scalar_fields(&wire) && empty_refs => Self::Done,
644            7 if wire.domain_id.is_none()
645                && wire.nonce_or_proof.is_none()
646                && empty_refs
647                && wire.max_frame.is_none()
648                && wire.commit.is_none()
649                && wire.payload.is_none() =>
650            {
651                Self::Error {
652                    code: wire.error_code.ok_or(SyncFrameError::InvalidShape)?,
653                }
654            }
655            8 if unused_scalar_fields(&wire) && empty_refs => Self::InvitationRequest,
656            9 if unused_scalar_fields_except_payload(&wire) && empty_refs => Self::Invitation {
657                payload: wire.payload.ok_or(SyncFrameError::InvalidShape)?,
658            },
659            _ => return Err(SyncFrameError::InvalidShape),
660        };
661        validate(&frame)?;
662        Ok(frame)
663    }
664}
665
666fn decode_authority_frame(wire: WireFrame) -> Result<SyncFrame, SyncFrameError> {
667    let common_shape = wire.domain_id.is_none()
668        && wire.nonce_or_proof.is_none()
669        && wire.commits.is_empty()
670        && wire.max_frame.is_none()
671        && wire.commit.is_none()
672        && wire.error_code.is_none();
673    match wire.kind {
674        10 if common_shape
675            && wire.payload.is_none()
676            && wire.payloads.is_empty()
677            && wire.epoch_key.is_none() =>
678        {
679            Ok(SyncFrame::AuthorityState {
680                epoch: wire.epoch.ok_or(SyncFrameError::InvalidShape)?,
681                sequence: wire.sequence.ok_or(SyncFrameError::InvalidShape)?,
682                head: wire.head.map(Into::into),
683            })
684        }
685        11 if common_shape
686            && wire.epoch.is_none()
687            && wire.sequence.is_none()
688            && wire.head.is_none() =>
689        {
690            Ok(SyncFrame::ControlGrant {
691                transition: wire.payload.ok_or(SyncFrameError::InvalidShape)?,
692                epoch_key: wire
693                    .epoch_key
694                    .map(Into::into)
695                    .ok_or(SyncFrameError::InvalidShape)?,
696                capability_chain: wire.payloads,
697            })
698        }
699        12 if common_shape
700            && wire.epoch.is_none()
701            && wire.sequence.is_none()
702            && wire.head.is_none()
703            && wire.epoch_key.is_none()
704            && wire.payloads.is_empty() =>
705        {
706            Ok(SyncFrame::ControlEvidence {
707                transition: wire.payload.ok_or(SyncFrameError::InvalidShape)?,
708            })
709        }
710        _ => Err(SyncFrameError::InvalidShape),
711    }
712}
713
714fn unused_scalar_fields(wire: &WireFrame) -> bool {
715    wire.domain_id.is_none()
716        && wire.nonce_or_proof.is_none()
717        && wire.max_frame.is_none()
718        && wire.commit.is_none()
719        && wire.error_code.is_none()
720        && wire.payload.is_none()
721        && wire.epoch.is_none()
722        && wire.sequence.is_none()
723        && wire.head.is_none()
724        && wire.payloads.is_empty()
725        && wire.epoch_key.is_none()
726        && wire.min_protocol.is_none()
727        && wire.max_protocol.is_none()
728}
729
730fn unused_scalar_fields_except_payload(wire: &WireFrame) -> bool {
731    wire.domain_id.is_none()
732        && wire.nonce_or_proof.is_none()
733        && wire.max_frame.is_none()
734        && wire.commit.is_none()
735        && wire.error_code.is_none()
736        && wire.epoch.is_none()
737        && wire.sequence.is_none()
738        && wire.head.is_none()
739        && wire.payloads.is_empty()
740        && wire.epoch_key.is_none()
741        && wire.min_protocol.is_none()
742        && wire.max_protocol.is_none()
743}
744
745fn fixed_proof(value: Option<minicbor::bytes::ByteArray<32>>) -> Result<[u8; 32], SyncFrameError> {
746    value.map(Into::into).ok_or(SyncFrameError::InvalidShape)
747}
748
749fn validate(frame: &SyncFrame) -> Result<(), SyncFrameError> {
750    match frame {
751        SyncFrame::Hello { max_frame, .. }
752            if *max_frame == 0 || *max_frame as usize > MAX_CONTROL_FRAME =>
753        {
754            Err(SyncFrameError::FrameTooLarge(*max_frame as usize))
755        }
756        SyncFrame::Hello {
757            min_protocol,
758            max_protocol,
759            ..
760        } if min_protocol > max_protocol => Err(SyncFrameError::InvalidProtocolRange {
761            min: *min_protocol,
762            max: *max_protocol,
763        }),
764        SyncFrame::AuthorityState {
765            epoch, sequence, ..
766        } if epoch != sequence => Err(SyncFrameError::InvalidShape),
767        SyncFrame::Frontier { commits }
768        | SyncFrame::Need { commits }
769        | SyncFrame::Refs { commits } => validate_references(commits),
770        SyncFrame::Invitation { payload } if payload.len() > 64 * 1024 => {
771            Err(SyncFrameError::InvitationTooLarge(payload.len()))
772        }
773        SyncFrame::DomainRequest {
774            capability_chain, ..
775        } => validate_capability_chain(capability_chain, MAX_CAPABILITY_CHAIN),
776        SyncFrame::ControlGrant {
777            capability_chain, ..
778        } => validate_capability_chain(capability_chain, MAX_CONTROL_CAPABILITIES),
779        _ => Ok(()),
780    }
781}
782
783fn validate_capability_chain(chain: &[Vec<u8>], maximum: usize) -> Result<(), SyncFrameError> {
784    if chain.len() > maximum {
785        return Err(SyncFrameError::TooManyAuthorityObjects(chain.len()));
786    }
787    if let Some(object) = chain
788        .iter()
789        .find(|object| object.len() > MAX_CAPABILITY_BYTES)
790    {
791        return Err(SyncFrameError::AuthorityObjectTooLarge(object.len()));
792    }
793    Ok(())
794}
795
796fn validate_references(commits: &[CommitId]) -> Result<(), SyncFrameError> {
797    if commits.len() > MAX_REFERENCES {
798        return Err(SyncFrameError::TooManyReferences(commits.len()));
799    }
800    if commits.windows(2).any(|pair| pair[0] >= pair[1]) {
801        return Err(SyncFrameError::NonCanonicalReferences);
802    }
803    Ok(())
804}
805
806fn encode_hint_content(
807    author: AuthorId,
808    commit: CommitId,
809    frontier_digest: [u8; 32],
810    counter: u64,
811) -> Result<Vec<u8>, SyncFrameError> {
812    minicbor::to_vec(GossipContent {
813        version: GOSSIP_VERSION,
814        author,
815        commit,
816        frontier_digest: frontier_digest.into(),
817        counter,
818    })
819    .map_err(codec_error)
820}
821
822fn decode_exact(bytes: &[u8]) -> Result<GossipContent, SyncFrameError> {
823    let mut decoder = minicbor::Decoder::new(bytes);
824    let content = decoder.decode().map_err(codec_error)?;
825    if decoder.position() != bytes.len() {
826        return Err(SyncFrameError::NonCanonical);
827    }
828    Ok(content)
829}
830
831#[allow(clippy::needless_pass_by_value)]
832fn codec_error(error: impl std::fmt::Display) -> SyncFrameError {
833    SyncFrameError::Codec(error.to_string())
834}
iroh_db_sync/lib.rs

iroh_db_sync/
lib.rs
