asupersync/net/atp/
channel_bonding.rs

1//! Channel-bonding invariants for ATP RaptorQ transfers.
2//!
3//! Channel bonding lets multiple donors seed the same byte-identical transfer:
4//! each donor emits a disjoint residue class of RaptorQ ESIs, and the receiver
5//! treats every authenticated symbol as fungible. Phase A is deliberately pure
6//! data and math. Transport wiring consumes these types later; this module does
7//! not open sockets, read files, or mutate transfer state.
8//!
9//! The security model is layered:
10//!
11//! - Descriptor identity is content-addressed: donors recompute per-entry
12//!   SHA-256, content ids, the flat object-graph merkle root, and the RQ
13//!   transfer id before donating. A mismatch refuses donation.
14//! - Symbol auth uses the existing symbol HMAC key shared by all approved
15//!   donors. The key reference here is an out-of-band handle; raw key bytes must
16//!   travel only through the encrypted control plane, not argv or logs.
17//! - An authenticated but malicious donor can waste bandwidth by sending symbols
18//!   that fail auth or duplicate already-seen ESIs. It cannot make the receiver
19//!   commit corrupt output because the receiver still verifies per-entry SHA-256
20//!   and merkle root before commit.
21
22use std::collections::{BTreeMap, BTreeSet};
23use std::fmt;
24use std::net::SocketAddr;
25
26use serde::{Deserialize, Serialize};
27use sha2::{Digest, Sha256};
28
29use crate::atp::object::{ContentId, ObjectId as AtpObjectId};
30use crate::net::atp::transport_common::{EntryDigest, flat_merkle_root_from_digests, hex_encode};
31use crate::types::symbol::ObjectId as RaptorqObjectId;
32
33/// Current channel-bonding protocol version.
34pub const BONDING_PROTOCOL_VERSION: u16 = 1;
35
36/// Maximum donor count accepted by the Phase A static-residue scheme.
37pub const MAX_STATIC_RESIDUE_DONORS: u32 = 1024;
38
39/// Transfer descriptor shared by receiver and donors.
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41pub struct BondTransferDescriptor {
42    /// RQ transfer id derived from the merkle root, total bytes, and file count.
43    pub transfer_id: String,
44    /// Display/root name from the original transfer manifest.
45    pub root_name: String,
46    /// Whether the transfer represents a directory tree.
47    pub is_directory: bool,
48    /// Total bytes across all entries.
49    pub total_bytes: u64,
50    /// Canonical flat object-graph merkle root.
51    pub merkle_root_hex: String,
52    /// Per-entry byte identity.
53    pub entries: Vec<BondDescriptorEntry>,
54    /// RaptorQ symbol size donors must use.
55    pub symbol_size: u16,
56    /// Maximum RaptorQ source block size donors must use.
57    pub max_block_size: usize,
58    /// Out-of-band key handle for shared symbol authentication.
59    pub auth_key_ref: Option<String>,
60}
61
62impl BondTransferDescriptor {
63    /// Build a descriptor from a donor-local byte proof.
64    pub fn from_donor_proof(
65        root_name: impl Into<String>,
66        is_directory: bool,
67        symbol_size: u16,
68        max_block_size: usize,
69        auth_key_ref: Option<String>,
70        proof_entries: Vec<BondDonorProofEntry>,
71    ) -> Result<Self, ChannelBondingError> {
72        validate_proof_entries(&proof_entries)?;
73        let total_bytes = total_bytes_for_proof(&proof_entries)?;
74        let merkle_root_hex = merkle_root_for_proof(&proof_entries)?;
75        let transfer_id = transfer_id_hex(&merkle_root_hex, total_bytes, proof_entries.len());
76        let entries = proof_entries
77            .iter()
78            .map(BondDescriptorEntry::from)
79            .collect();
80
81        let descriptor = Self {
82            transfer_id,
83            root_name: root_name.into(),
84            is_directory,
85            total_bytes,
86            merkle_root_hex,
87            entries,
88            symbol_size,
89            max_block_size,
90            auth_key_ref,
91        };
92        descriptor.validate()?;
93        Ok(descriptor)
94    }
95
96    /// Validate descriptor self-consistency.
97    pub fn validate(&self) -> Result<(), ChannelBondingError> {
98        validate_hex_32("merkle_root_hex", None, &self.merkle_root_hex)?;
99        if self.transfer_id
100            != transfer_id_hex(&self.merkle_root_hex, self.total_bytes, self.entries.len())
101        {
102            return Err(ChannelBondingError::TransferIdMismatch {
103                expected: transfer_id_hex(
104                    &self.merkle_root_hex,
105                    self.total_bytes,
106                    self.entries.len(),
107                ),
108                actual: self.transfer_id.clone(),
109            });
110        }
111        validate_descriptor_entries(&self.entries)?;
112        let computed_total = total_bytes_for_descriptor(&self.entries)?;
113        if computed_total != self.total_bytes {
114            return Err(ChannelBondingError::TotalBytesMismatch {
115                expected: self.total_bytes,
116                actual: computed_total,
117            });
118        }
119        Ok(())
120    }
121
122    /// Verify that a donor holds bytes identical to this descriptor.
123    ///
124    /// Donors build [`BondDonorByteProof`] from their local content stream. The
125    /// verifier checks descriptor fields first, then recomputes the canonical
126    /// flat merkle root from the donor's content ids and SHA-256 digests. Any
127    /// mismatch is fail-closed.
128    pub fn verify_donor_byte_match(
129        &self,
130        proof: &BondDonorByteProof,
131    ) -> Result<(), ChannelBondingError> {
132        self.validate()?;
133        proof.validate()?;
134
135        if proof.transfer_id != self.transfer_id {
136            return Err(ChannelBondingError::TransferIdMismatch {
137                expected: self.transfer_id.clone(),
138                actual: proof.transfer_id.clone(),
139            });
140        }
141        if proof.total_bytes != self.total_bytes {
142            return Err(ChannelBondingError::TotalBytesMismatch {
143                expected: self.total_bytes,
144                actual: proof.total_bytes,
145            });
146        }
147        if proof.merkle_root_hex != self.merkle_root_hex {
148            return Err(ChannelBondingError::MerkleRootMismatch {
149                expected: self.merkle_root_hex.clone(),
150                actual: proof.merkle_root_hex.clone(),
151            });
152        }
153
154        let expected = descriptor_entries_by_index(&self.entries)?;
155        let actual = proof_entries_by_index(&proof.entries)?;
156        if expected.len() != actual.len() {
157            return Err(ChannelBondingError::EntryCountMismatch {
158                expected: expected.len(),
159                actual: actual.len(),
160            });
161        }
162
163        for (index, descriptor_entry) in expected {
164            let proof_entry = actual
165                .get(&index)
166                .ok_or(ChannelBondingError::MissingProofEntry { index })?;
167            compare_entry_identity(descriptor_entry, proof_entry)?;
168        }
169
170        let computed_merkle = merkle_root_for_proof(&proof.entries)?;
171        if computed_merkle != self.merkle_root_hex {
172            return Err(ChannelBondingError::MerkleRootMismatch {
173                expected: self.merkle_root_hex.clone(),
174                actual: computed_merkle,
175            });
176        }
177
178        Ok(())
179    }
180
181    /// Derive the RaptorQ object id for an entry in this transfer.
182    #[must_use]
183    pub fn entry_object_id(&self, entry_index: u32) -> RaptorqObjectId {
184        entry_object_id(&self.transfer_id, entry_index)
185    }
186}
187
188/// Descriptor row for one transfer entry.
189#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
190pub struct BondDescriptorEntry {
191    /// Stable entry index from the transfer manifest.
192    pub index: u32,
193    /// Transfer-relative path.
194    pub rel_path: String,
195    /// Entry size in bytes.
196    pub size: u64,
197    /// Plain content SHA-256, lowercase hex in canonical producers.
198    pub sha256_hex: String,
199}
200
201impl From<&BondDonorProofEntry> for BondDescriptorEntry {
202    fn from(entry: &BondDonorProofEntry) -> Self {
203        Self {
204            index: entry.index,
205            rel_path: entry.rel_path.clone(),
206            size: entry.size,
207            sha256_hex: entry.sha256_hex.clone(),
208        }
209    }
210}
211
212/// Donor-local proof that its bytes match a descriptor.
213#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214pub struct BondDonorByteProof {
215    /// Transfer id recomputed from the donor's local entries.
216    pub transfer_id: String,
217    /// Total bytes recomputed from the donor's local entries.
218    pub total_bytes: u64,
219    /// Merkle root recomputed from donor-local content ids and SHA-256 digests.
220    pub merkle_root_hex: String,
221    /// Per-entry donor-local byte proof.
222    pub entries: Vec<BondDonorProofEntry>,
223}
224
225impl BondDonorByteProof {
226    /// Build a donor proof from precomputed streaming entry digests.
227    pub fn from_entries(entries: Vec<BondDonorProofEntry>) -> Result<Self, ChannelBondingError> {
228        validate_proof_entries(&entries)?;
229        let total_bytes = total_bytes_for_proof(&entries)?;
230        let merkle_root_hex = merkle_root_for_proof(&entries)?;
231        let transfer_id = transfer_id_hex(&merkle_root_hex, total_bytes, entries.len());
232        Ok(Self {
233            transfer_id,
234            total_bytes,
235            merkle_root_hex,
236            entries,
237        })
238    }
239
240    /// Validate proof self-consistency.
241    pub fn validate(&self) -> Result<(), ChannelBondingError> {
242        validate_hex_32("merkle_root_hex", None, &self.merkle_root_hex)?;
243        validate_proof_entries(&self.entries)?;
244        let computed_total = total_bytes_for_proof(&self.entries)?;
245        if computed_total != self.total_bytes {
246            return Err(ChannelBondingError::TotalBytesMismatch {
247                expected: self.total_bytes,
248                actual: computed_total,
249            });
250        }
251        let computed_merkle = merkle_root_for_proof(&self.entries)?;
252        if computed_merkle != self.merkle_root_hex {
253            return Err(ChannelBondingError::MerkleRootMismatch {
254                expected: self.merkle_root_hex.clone(),
255                actual: computed_merkle,
256            });
257        }
258        let computed_transfer_id =
259            transfer_id_hex(&self.merkle_root_hex, self.total_bytes, self.entries.len());
260        if computed_transfer_id != self.transfer_id {
261            return Err(ChannelBondingError::TransferIdMismatch {
262                expected: computed_transfer_id,
263                actual: self.transfer_id.clone(),
264            });
265        }
266        Ok(())
267    }
268}
269
270/// Proof row for one donor-local entry.
271#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
272pub struct BondDonorProofEntry {
273    /// Stable entry index from the transfer manifest.
274    pub index: u32,
275    /// Transfer-relative path.
276    pub rel_path: String,
277    /// Entry size in bytes.
278    pub size: u64,
279    /// Domain-separated content id (`ContentId::from_bytes`) as hex.
280    pub content_id_hex: String,
281    /// Plain content SHA-256 as hex.
282    pub sha256_hex: String,
283}
284
285impl BondDonorProofEntry {
286    /// Build an entry proof from bytes already held by a caller.
287    ///
288    /// Streaming transport code should prefer `ContentIdHasher` plus
289    /// `Sha256` over this convenience constructor; tests and small fixtures can
290    /// use it directly.
291    #[must_use]
292    pub fn from_bytes(index: u32, rel_path: impl Into<String>, bytes: &[u8]) -> Self {
293        let content_id = ContentId::from_bytes(bytes);
294        let sha256 = Sha256::digest(bytes);
295        Self {
296            index,
297            rel_path: rel_path.into(),
298            size: bytes.len() as u64,
299            content_id_hex: content_id.to_hex(),
300            sha256_hex: hex_encode(&sha256),
301        }
302    }
303}
304
305/// Per-donor ESI assignment.
306#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
307pub struct DonorAssignment {
308    /// Zero-based donor index.
309    pub donor_index: u32,
310    /// Total donors participating in this transfer.
311    pub donor_count: u32,
312    /// Optional explicit ESI windows reserved for Phase E dynamic allocation.
313    pub esi_windows: Option<Vec<EsiWindow>>,
314    /// Receiver UDP endpoints reachable by this donor.
315    pub receiver_udp_endpoints: Vec<SocketAddr>,
316    /// Out-of-band key handle for the shared symbol auth key.
317    pub auth_key_ref: Option<String>,
318}
319
320impl DonorAssignment {
321    /// Validate index bounds and optional ESI windows.
322    pub fn validate(&self) -> Result<(), ChannelBondingError> {
323        validate_donor_index(self.donor_index, self.donor_count)?;
324        if self.donor_count > MAX_STATIC_RESIDUE_DONORS {
325            return Err(ChannelBondingError::TooManyDonors {
326                donor_count: self.donor_count,
327                max: MAX_STATIC_RESIDUE_DONORS,
328            });
329        }
330        if let Some(windows) = &self.esi_windows {
331            for window in windows {
332                window.validate()?;
333            }
334        }
335        Ok(())
336    }
337
338    /// Whether this assignment owns an ESI.
339    #[must_use]
340    pub fn owns_esi(&self, esi: u32) -> bool {
341        if !owns_esi(self.donor_index, self.donor_count, esi) {
342            return false;
343        }
344        self.esi_windows
345            .as_ref()
346            .is_none_or(|windows| windows.iter().any(|window| window.contains(esi)))
347    }
348}
349
350/// Encrypted control plane used to distribute the shared donor auth key.
351#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
352pub enum BondAuthControlPlane {
353    /// SSH control connection.
354    Ssh,
355    /// Tailscale/WireGuard control connection.
356    Tailscale,
357}
358
359/// Donor-local location for shared symbol-auth key material.
360#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
361pub enum BondAuthKeyLocation {
362    /// Environment variable containing key material.
363    EnvVar(String),
364    /// File containing key material.
365    KeyFile(String),
366    /// Explicitly rejected: argv is visible through process listings on shared hosts.
367    Argv(String),
368}
369
370/// Reference to the shared symbol-auth key for a bonded transfer.
371#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
372pub struct BondAuthKeyRef {
373    /// Stable key identifier; never raw key material.
374    pub key_id: String,
375    /// Encrypted control plane that delivered the key material.
376    pub control_plane: BondAuthControlPlane,
377    /// Donor-local key material location.
378    pub location: BondAuthKeyLocation,
379}
380
381impl BondAuthKeyRef {
382    /// Validate the key reference and fail closed on unsafe delivery.
383    pub fn validate(&self) -> Result<(), ChannelBondingError> {
384        if self.key_id.trim().is_empty() {
385            return Err(ChannelBondingError::MissingAuthKeyId);
386        }
387        match &self.location {
388            BondAuthKeyLocation::EnvVar(name) => {
389                if name.trim().is_empty() {
390                    return Err(ChannelBondingError::MissingAuthKeyLocation);
391                }
392            }
393            BondAuthKeyLocation::KeyFile(path) => {
394                if path.trim().is_empty() {
395                    return Err(ChannelBondingError::MissingAuthKeyLocation);
396                }
397            }
398            BondAuthKeyLocation::Argv(_) => {
399                return Err(ChannelBondingError::InsecureAuthKeyDelivery);
400            }
401        }
402        Ok(())
403    }
404}
405
406/// Fail-closed security model for bonded donor symbols.
407#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
408pub struct BondingSecurityModel {
409    /// Shared symbol-auth key reference.
410    pub auth_key: BondAuthKeyRef,
411    /// Receiver verifies each symbol tag before decode.
412    pub auth_before_decode: bool,
413    /// Receiver commits only after final SHA-256 and Merkle verification.
414    pub fail_closed_on_merkle_mismatch: bool,
415}
416
417impl BondingSecurityModel {
418    /// Validate that both auth and final content integrity remain enabled.
419    pub fn validate(&self) -> Result<(), ChannelBondingError> {
420        self.auth_key.validate()?;
421        if !self.auth_before_decode {
422            return Err(ChannelBondingError::AuthBeforeDecodeDisabled);
423        }
424        if !self.fail_closed_on_merkle_mismatch {
425            return Err(ChannelBondingError::MerkleFailClosedDisabled);
426        }
427        Ok(())
428    }
429}
430
431/// Half-open ESI window `[start, end)`.
432#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
433pub struct EsiWindow {
434    /// First included ESI.
435    pub start_inclusive: u32,
436    /// First excluded ESI.
437    pub end_exclusive: u32,
438}
439
440impl EsiWindow {
441    /// Construct a half-open ESI window.
442    #[must_use]
443    pub const fn new(start_inclusive: u32, end_exclusive: u32) -> Self {
444        Self {
445            start_inclusive,
446            end_exclusive,
447        }
448    }
449
450    /// Check that the window is non-empty.
451    pub fn validate(&self) -> Result<(), ChannelBondingError> {
452        if self.start_inclusive >= self.end_exclusive {
453            return Err(ChannelBondingError::InvalidEsiWindow {
454                start_inclusive: self.start_inclusive,
455                end_exclusive: self.end_exclusive,
456            });
457        }
458        Ok(())
459    }
460
461    /// Whether the half-open window contains `esi`.
462    #[must_use]
463    pub const fn contains(&self, esi: u32) -> bool {
464        self.start_inclusive <= esi && esi < self.end_exclusive
465    }
466}
467
468/// Transport family advertised during bonding negotiation.
469#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
470pub enum BondTransport {
471    /// Direct IP path.
472    DirectIp,
473    /// SSH control/data path.
474    Ssh,
475    /// Tailscale/WireGuard path.
476    Tailscale,
477}
478
479/// Negotiation offer from a receiver or donor.
480#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
481pub struct BondingHandshake {
482    /// Lowest compatible bonding protocol version.
483    pub min_protocol_version: u16,
484    /// Highest compatible bonding protocol version.
485    pub max_protocol_version: u16,
486    /// Supported path transports.
487    pub supported_transports: BTreeSet<BondTransport>,
488    /// Whether explicit dynamic ESI windows are supported.
489    pub supports_dynamic_windows: bool,
490    /// Whether resume metadata is supported.
491    pub supports_resume: bool,
492    /// Whether symbol auth is required by this endpoint.
493    pub auth_required: bool,
494    /// Maximum donor count accepted by this endpoint.
495    pub max_donor_count: u32,
496    /// Forward-compatible extension tokens. Unknown tokens are ignored.
497    pub extension_capabilities: BTreeSet<String>,
498}
499
500impl BondingHandshake {
501    /// Build a version-1 static-residue offer.
502    #[must_use]
503    pub fn v1_static(
504        transports: impl IntoIterator<Item = BondTransport>,
505        max_donor_count: u32,
506        auth_required: bool,
507    ) -> Self {
508        Self {
509            min_protocol_version: BONDING_PROTOCOL_VERSION,
510            max_protocol_version: BONDING_PROTOCOL_VERSION,
511            supported_transports: transports.into_iter().collect(),
512            supports_dynamic_windows: false,
513            supports_resume: false,
514            auth_required,
515            max_donor_count,
516            extension_capabilities: BTreeSet::new(),
517        }
518    }
519
520    /// Set an inclusive bonding protocol version range.
521    #[must_use]
522    pub const fn with_protocol_range(
523        mut self,
524        min_protocol_version: u16,
525        max_protocol_version: u16,
526    ) -> Self {
527        self.min_protocol_version = min_protocol_version;
528        self.max_protocol_version = max_protocol_version;
529        self
530    }
531
532    /// Advertise receiver-allocated dynamic ESI windows.
533    #[must_use]
534    pub const fn with_dynamic_windows(mut self, supported: bool) -> Self {
535        self.supports_dynamic_windows = supported;
536        self
537    }
538
539    /// Advertise partial-transfer resume metadata.
540    #[must_use]
541    pub const fn with_resume(mut self, supported: bool) -> Self {
542        self.supports_resume = supported;
543        self
544    }
545
546    /// Advertise a forward-compatible extension token.
547    #[must_use]
548    pub fn with_extension_capability(mut self, capability: impl Into<String>) -> Self {
549        self.extension_capabilities.insert(capability.into());
550        self
551    }
552
553    /// Negotiate a compatible agreement with a peer.
554    pub fn negotiate(&self, peer: &Self) -> Result<BondingAgreement, ChannelBondingError> {
555        validate_handshake_offer(self)?;
556        validate_handshake_offer(peer)?;
557        if self.supported_transports.is_empty() || peer.supported_transports.is_empty() {
558            return Err(ChannelBondingError::NoCommonTransport);
559        }
560
561        let selected_version = self.max_protocol_version.min(peer.max_protocol_version);
562        let required_min = self.min_protocol_version.max(peer.min_protocol_version);
563        if selected_version < required_min {
564            return Err(ChannelBondingError::IncompatibleProtocolVersion {
565                local_min: self.min_protocol_version,
566                local_max: self.max_protocol_version,
567                peer_min: peer.min_protocol_version,
568                peer_max: peer.max_protocol_version,
569            });
570        }
571
572        let supported_transports = self
573            .supported_transports
574            .intersection(&peer.supported_transports)
575            .copied()
576            .collect::<BTreeSet<_>>();
577        if supported_transports.is_empty() {
578            return Err(ChannelBondingError::NoCommonTransport);
579        }
580
581        let max_donor_count = self.max_donor_count.min(peer.max_donor_count);
582        if max_donor_count == 0 {
583            return Err(ChannelBondingError::InvalidDonorCount { donor_count: 0 });
584        }
585
586        Ok(BondingAgreement {
587            protocol_version: selected_version,
588            supported_transports,
589            assignment_mode: if self.supports_dynamic_windows && peer.supports_dynamic_windows {
590                BondingAssignmentMode::DynamicWindows
591            } else {
592                BondingAssignmentMode::StaticResidue
593            },
594            resume_supported: self.supports_resume && peer.supports_resume,
595            auth_required: self.auth_required || peer.auth_required,
596            max_donor_count,
597        })
598    }
599}
600
601/// Result of a compatible bonding handshake.
602#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
603pub struct BondingAgreement {
604    /// Selected bonding protocol version.
605    pub protocol_version: u16,
606    /// Common supported transports.
607    pub supported_transports: BTreeSet<BondTransport>,
608    /// Assignment mode for this donor/receiver pair.
609    pub assignment_mode: BondingAssignmentMode,
610    /// Whether resume may be used.
611    pub resume_supported: bool,
612    /// Whether symbols must carry auth tags.
613    pub auth_required: bool,
614    /// Negotiated donor-count ceiling.
615    pub max_donor_count: u32,
616}
617
618impl BondingAgreement {
619    /// Whether this agreement permits a transport family.
620    #[must_use]
621    pub fn supports_transport(&self, transport: BondTransport) -> bool {
622        self.supported_transports.contains(&transport)
623    }
624}
625
626/// ESI allocation mode selected by negotiation.
627#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
628pub enum BondingAssignmentMode {
629    /// Static donor residue classes: donor `i` owns `esi % N == i`.
630    StaticResidue,
631    /// Receiver-allocated explicit ESI windows.
632    DynamicWindows,
633}
634
635/// Iterator over the ESI residue class for one donor.
636#[derive(Debug, Clone)]
637pub struct DonorEsiStream {
638    donor_index: u32,
639    donor_count: u32,
640    next_seq: Option<u32>,
641}
642
643impl DonorEsiStream {
644    /// Create an ESI stream for `donor_index` of `donor_count`.
645    pub fn new(donor_index: u32, donor_count: u32) -> Result<Self, ChannelBondingError> {
646        validate_donor_index(donor_index, donor_count)?;
647        Ok(Self {
648            donor_index,
649            donor_count,
650            next_seq: Some(0),
651        })
652    }
653}
654
655impl Iterator for DonorEsiStream {
656    type Item = u32;
657
658    fn next(&mut self) -> Option<Self::Item> {
659        let seq = self.next_seq?;
660        match esi_for_donor(self.donor_index, self.donor_count, seq) {
661            Ok(esi) => {
662                self.next_seq = seq.checked_add(1);
663                Some(esi)
664            }
665            Err(_) => {
666                self.next_seq = None;
667                None
668            }
669        }
670    }
671}
672
673/// Return the `seq`th ESI owned by donor `i` of `N`.
674pub fn esi_for_donor(
675    donor_index: u32,
676    donor_count: u32,
677    seq: u32,
678) -> Result<u32, ChannelBondingError> {
679    validate_donor_index(donor_index, donor_count)?;
680    seq.checked_mul(donor_count)
681        .and_then(|base| base.checked_add(donor_index))
682        .ok_or(ChannelBondingError::EsiOverflow)
683}
684
685/// Whether donor `i` of `N` owns `esi` in the static-residue scheme.
686#[must_use]
687pub const fn owns_esi(donor_index: u32, donor_count: u32, esi: u32) -> bool {
688    donor_count != 0 && donor_index < donor_count && esi % donor_count == donor_index
689}
690
691/// Derive the per-entry RaptorQ object id from a transfer id and entry index.
692#[must_use]
693pub fn entry_object_id(transfer_id: &str, index: u32) -> RaptorqObjectId {
694    let mut hasher = Sha256::new();
695    hasher.update(b"asupersync.atp.rq.entry-object-id.v1\0");
696    hasher.update(transfer_id.as_bytes());
697    hasher.update(index.to_be_bytes());
698    let digest = hasher.finalize();
699    let high = u64::from_be_bytes([
700        digest[0], digest[1], digest[2], digest[3], digest[4], digest[5], digest[6], digest[7],
701    ]);
702    let low = u64::from_be_bytes([
703        digest[8], digest[9], digest[10], digest[11], digest[12], digest[13], digest[14],
704        digest[15],
705    ]);
706    RaptorqObjectId::new(high, low)
707}
708
709/// Derive the RQ transfer id for a bonded descriptor.
710#[must_use]
711pub fn transfer_id_hex(merkle_root_hex: &str, total_bytes: u64, file_count: usize) -> String {
712    let mut hasher = Sha256::new();
713    hasher.update(b"asupersync.atp.rq.transfer-id.v1\0");
714    hasher.update(merkle_root_hex.as_bytes());
715    hasher.update(total_bytes.to_be_bytes());
716    hasher.update(u64::try_from(file_count).unwrap_or(u64::MAX).to_be_bytes());
717    hex_encode(&hasher.finalize()[..16])
718}
719
720/// Channel-bonding Phase A validation failure.
721#[derive(Debug, Clone, PartialEq, Eq)]
722pub enum ChannelBondingError {
723    /// Donor count was zero.
724    InvalidDonorCount {
725        /// Invalid donor count.
726        donor_count: u32,
727    },
728    /// Donor index is outside `0..donor_count`.
729    InvalidDonorIndex {
730        /// Donor index supplied.
731        donor_index: u32,
732        /// Donor count supplied.
733        donor_count: u32,
734    },
735    /// Donor count exceeds the static-residue cap.
736    TooManyDonors {
737        /// Donor count supplied.
738        donor_count: u32,
739        /// Maximum accepted donor count.
740        max: u32,
741    },
742    /// ESI arithmetic overflowed `u32`.
743    EsiOverflow,
744    /// ESI window is empty or inverted.
745    InvalidEsiWindow {
746        /// First included ESI.
747        start_inclusive: u32,
748        /// First excluded ESI.
749        end_exclusive: u32,
750    },
751    /// Protocol range was internally invalid.
752    InvalidProtocolRange {
753        /// Minimum accepted version.
754        min: u16,
755        /// Maximum accepted version.
756        max: u16,
757    },
758    /// Local and peer protocol ranges do not overlap.
759    IncompatibleProtocolVersion {
760        /// Local minimum accepted version.
761        local_min: u16,
762        /// Local maximum accepted version.
763        local_max: u16,
764        /// Peer minimum accepted version.
765        peer_min: u16,
766        /// Peer maximum accepted version.
767        peer_max: u16,
768    },
769    /// No transport family is common to both peers.
770    NoCommonTransport,
771    /// Hex field is not a 32-byte digest.
772    InvalidHexDigest {
773        /// Field name.
774        field: &'static str,
775        /// Optional entry index.
776        index: Option<u32>,
777        /// Invalid value.
778        value: String,
779    },
780    /// Duplicate descriptor or proof entry index.
781    DuplicateEntryIndex {
782        /// Duplicate index.
783        index: u32,
784    },
785    /// Duplicate descriptor or proof relative path.
786    DuplicateEntryPath {
787        /// Duplicate path.
788        rel_path: String,
789    },
790    /// Entry count differs between descriptor and proof.
791    EntryCountMismatch {
792        /// Expected count.
793        expected: usize,
794        /// Actual count.
795        actual: usize,
796    },
797    /// Proof omitted a descriptor entry.
798    MissingProofEntry {
799        /// Missing entry index.
800        index: u32,
801    },
802    /// Entry field differs between descriptor and proof.
803    EntryMismatch {
804        /// Entry index.
805        index: u32,
806        /// Field name.
807        field: &'static str,
808        /// Expected value.
809        expected: String,
810        /// Actual value.
811        actual: String,
812    },
813    /// Total bytes differs from the entries.
814    TotalBytesMismatch {
815        /// Expected total.
816        expected: u64,
817        /// Actual total.
818        actual: u64,
819    },
820    /// Entry sizes overflowed `u64`.
821    TotalBytesOverflow,
822    /// Transfer id differs from the canonical RQ descriptor id.
823    TransferIdMismatch {
824        /// Expected transfer id.
825        expected: String,
826        /// Actual transfer id.
827        actual: String,
828    },
829    /// Merkle root differs from the canonical donor proof root.
830    MerkleRootMismatch {
831        /// Expected merkle root.
832        expected: String,
833        /// Actual merkle root.
834        actual: String,
835    },
836    /// Shared symbol-auth key id is empty.
837    MissingAuthKeyId,
838    /// Shared symbol-auth key location is empty.
839    MissingAuthKeyLocation,
840    /// Shared symbol-auth key was configured for argv delivery.
841    InsecureAuthKeyDelivery,
842    /// Receiver auth-before-decode was disabled.
843    AuthBeforeDecodeDisabled,
844    /// Final Merkle/SHA fail-closed verification was disabled.
845    MerkleFailClosedDisabled,
846}
847
848impl fmt::Display for ChannelBondingError {
849    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
850        match self {
851            Self::InvalidDonorCount { donor_count } => {
852                write!(f, "invalid donor_count {donor_count}; must be nonzero")
853            }
854            Self::InvalidDonorIndex {
855                donor_index,
856                donor_count,
857            } => write!(
858                f,
859                "invalid donor_index {donor_index}; donor_count is {donor_count}"
860            ),
861            Self::TooManyDonors { donor_count, max } => {
862                write!(f, "donor_count {donor_count} exceeds max {max}")
863            }
864            Self::EsiOverflow => f.write_str("ESI arithmetic overflow"),
865            Self::InvalidEsiWindow {
866                start_inclusive,
867                end_exclusive,
868            } => write!(f, "invalid ESI window [{start_inclusive}, {end_exclusive})"),
869            Self::InvalidProtocolRange { min, max } => {
870                write!(f, "invalid bonding protocol range {min}..={max}")
871            }
872            Self::IncompatibleProtocolVersion {
873                local_min,
874                local_max,
875                peer_min,
876                peer_max,
877            } => write!(
878                f,
879                "incompatible bonding protocol versions: local {local_min}..={local_max}, peer {peer_min}..={peer_max}"
880            ),
881            Self::NoCommonTransport => f.write_str("no common bonding transport"),
882            Self::InvalidHexDigest {
883                field,
884                index,
885                value,
886            } => write!(f, "invalid {field} digest for entry {index:?}: {value:?}"),
887            Self::DuplicateEntryIndex { index } => {
888                write!(f, "duplicate bonded entry index {index}")
889            }
890            Self::DuplicateEntryPath { rel_path } => {
891                write!(f, "duplicate bonded entry path {rel_path:?}")
892            }
893            Self::EntryCountMismatch { expected, actual } => {
894                write!(
895                    f,
896                    "entry count mismatch: expected {expected}, actual {actual}"
897                )
898            }
899            Self::MissingProofEntry { index } => {
900                write!(f, "donor proof missing entry index {index}")
901            }
902            Self::EntryMismatch {
903                index,
904                field,
905                expected,
906                actual,
907            } => write!(
908                f,
909                "entry {index} {field} mismatch: expected {expected:?}, actual {actual:?}"
910            ),
911            Self::TotalBytesMismatch { expected, actual } => {
912                write!(
913                    f,
914                    "total bytes mismatch: expected {expected}, actual {actual}"
915                )
916            }
917            Self::TotalBytesOverflow => f.write_str("total bytes overflow"),
918            Self::TransferIdMismatch { expected, actual } => write!(
919                f,
920                "transfer id mismatch: expected {expected:?}, actual {actual:?}"
921            ),
922            Self::MerkleRootMismatch { expected, actual } => write!(
923                f,
924                "merkle root mismatch: expected {expected:?}, actual {actual:?}"
925            ),
926            Self::MissingAuthKeyId => f.write_str("bonding auth key id must not be empty"),
927            Self::MissingAuthKeyLocation => {
928                f.write_str("bonding auth key location must not be empty")
929            }
930            Self::InsecureAuthKeyDelivery => {
931                f.write_str("bonding auth key must not be delivered through argv")
932            }
933            Self::AuthBeforeDecodeDisabled => {
934                f.write_str("channel bonding requires auth-before-decode")
935            }
936            Self::MerkleFailClosedDisabled => {
937                f.write_str("channel bonding requires fail-closed Merkle/SHA verification")
938            }
939        }
940    }
941}
942
943impl std::error::Error for ChannelBondingError {}
944
945fn validate_donor_index(donor_index: u32, donor_count: u32) -> Result<(), ChannelBondingError> {
946    validate_donor_count(donor_count)?;
947    if donor_index >= donor_count {
948        return Err(ChannelBondingError::InvalidDonorIndex {
949            donor_index,
950            donor_count,
951        });
952    }
953    Ok(())
954}
955
956fn validate_donor_count(donor_count: u32) -> Result<(), ChannelBondingError> {
957    if donor_count == 0 {
958        return Err(ChannelBondingError::InvalidDonorCount { donor_count });
959    }
960    Ok(())
961}
962
963fn validate_protocol_range(min: u16, max: u16) -> Result<(), ChannelBondingError> {
964    if min == 0 || min > max {
965        return Err(ChannelBondingError::InvalidProtocolRange { min, max });
966    }
967    Ok(())
968}
969
970fn validate_handshake_offer(offer: &BondingHandshake) -> Result<(), ChannelBondingError> {
971    validate_protocol_range(offer.min_protocol_version, offer.max_protocol_version)?;
972    validate_donor_count(offer.max_donor_count)?;
973    if offer.max_donor_count > MAX_STATIC_RESIDUE_DONORS {
974        return Err(ChannelBondingError::TooManyDonors {
975            donor_count: offer.max_donor_count,
976            max: MAX_STATIC_RESIDUE_DONORS,
977        });
978    }
979    Ok(())
980}
981
982fn validate_descriptor_entries(entries: &[BondDescriptorEntry]) -> Result<(), ChannelBondingError> {
983    let mut indices = BTreeSet::new();
984    let mut paths = BTreeSet::new();
985    for entry in entries {
986        if !indices.insert(entry.index) {
987            return Err(ChannelBondingError::DuplicateEntryIndex { index: entry.index });
988        }
989        if !paths.insert(entry.rel_path.clone()) {
990            return Err(ChannelBondingError::DuplicateEntryPath {
991                rel_path: entry.rel_path.clone(),
992            });
993        }
994        validate_hex_32("sha256_hex", Some(entry.index), &entry.sha256_hex)?;
995    }
996    Ok(())
997}
998
999fn validate_proof_entries(entries: &[BondDonorProofEntry]) -> Result<(), ChannelBondingError> {
1000    let mut indices = BTreeSet::new();
1001    let mut paths = BTreeSet::new();
1002    for entry in entries {
1003        if !indices.insert(entry.index) {
1004            return Err(ChannelBondingError::DuplicateEntryIndex { index: entry.index });
1005        }
1006        if !paths.insert(entry.rel_path.clone()) {
1007            return Err(ChannelBondingError::DuplicateEntryPath {
1008                rel_path: entry.rel_path.clone(),
1009            });
1010        }
1011        validate_hex_32("content_id_hex", Some(entry.index), &entry.content_id_hex)?;
1012        validate_hex_32("sha256_hex", Some(entry.index), &entry.sha256_hex)?;
1013    }
1014    Ok(())
1015}
1016
1017fn validate_hex_32(
1018    field: &'static str,
1019    index: Option<u32>,
1020    value: &str,
1021) -> Result<(), ChannelBondingError> {
1022    if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
1023        return Err(ChannelBondingError::InvalidHexDigest {
1024            field,
1025            index,
1026            value: value.to_string(),
1027        });
1028    }
1029    Ok(())
1030}
1031
1032fn descriptor_entries_by_index(
1033    entries: &[BondDescriptorEntry],
1034) -> Result<BTreeMap<u32, &BondDescriptorEntry>, ChannelBondingError> {
1035    let mut by_index = BTreeMap::new();
1036    for entry in entries {
1037        if by_index.insert(entry.index, entry).is_some() {
1038            return Err(ChannelBondingError::DuplicateEntryIndex { index: entry.index });
1039        }
1040    }
1041    Ok(by_index)
1042}
1043
1044fn proof_entries_by_index(
1045    entries: &[BondDonorProofEntry],
1046) -> Result<BTreeMap<u32, &BondDonorProofEntry>, ChannelBondingError> {
1047    let mut by_index = BTreeMap::new();
1048    for entry in entries {
1049        if by_index.insert(entry.index, entry).is_some() {
1050            return Err(ChannelBondingError::DuplicateEntryIndex { index: entry.index });
1051        }
1052    }
1053    Ok(by_index)
1054}
1055
1056fn compare_entry_identity(
1057    descriptor: &BondDescriptorEntry,
1058    proof: &BondDonorProofEntry,
1059) -> Result<(), ChannelBondingError> {
1060    if descriptor.rel_path != proof.rel_path {
1061        return Err(ChannelBondingError::EntryMismatch {
1062            index: descriptor.index,
1063            field: "rel_path",
1064            expected: descriptor.rel_path.clone(),
1065            actual: proof.rel_path.clone(),
1066        });
1067    }
1068    if descriptor.size != proof.size {
1069        return Err(ChannelBondingError::EntryMismatch {
1070            index: descriptor.index,
1071            field: "size",
1072            expected: descriptor.size.to_string(),
1073            actual: proof.size.to_string(),
1074        });
1075    }
1076    if descriptor.sha256_hex != proof.sha256_hex {
1077        return Err(ChannelBondingError::EntryMismatch {
1078            index: descriptor.index,
1079            field: "sha256_hex",
1080            expected: descriptor.sha256_hex.clone(),
1081            actual: proof.sha256_hex.clone(),
1082        });
1083    }
1084    Ok(())
1085}
1086
1087fn total_bytes_for_descriptor(entries: &[BondDescriptorEntry]) -> Result<u64, ChannelBondingError> {
1088    entries.iter().try_fold(0u64, |acc, entry| {
1089        acc.checked_add(entry.size)
1090            .ok_or(ChannelBondingError::TotalBytesOverflow)
1091    })
1092}
1093
1094fn total_bytes_for_proof(entries: &[BondDonorProofEntry]) -> Result<u64, ChannelBondingError> {
1095    entries.iter().try_fold(0u64, |acc, entry| {
1096        acc.checked_add(entry.size)
1097            .ok_or(ChannelBondingError::TotalBytesOverflow)
1098    })
1099}
1100
1101fn merkle_root_for_proof(entries: &[BondDonorProofEntry]) -> Result<String, ChannelBondingError> {
1102    let digests = entries
1103        .iter()
1104        .map(|entry| {
1105            let content_id = ContentId::new(parse_hex_32(
1106                "content_id_hex",
1107                Some(entry.index),
1108                &entry.content_id_hex,
1109            )?);
1110            Ok(EntryDigest {
1111                rel_path: entry.rel_path.clone(),
1112                size: entry.size,
1113                content_id: AtpObjectId::content(content_id),
1114                content_sha256: parse_hex_32("sha256_hex", Some(entry.index), &entry.sha256_hex)?,
1115            })
1116        })
1117        .collect::<Result<Vec<_>, ChannelBondingError>>()?;
1118    Ok(flat_merkle_root_from_digests(&digests))
1119}
1120
1121fn parse_hex_32(
1122    field: &'static str,
1123    index: Option<u32>,
1124    value: &str,
1125) -> Result<[u8; 32], ChannelBondingError> {
1126    validate_hex_32(field, index, value)?;
1127    let mut out = [0u8; 32];
1128    hex::decode_to_slice(value, &mut out).map_err(|_| ChannelBondingError::InvalidHexDigest {
1129        field,
1130        index,
1131        value: value.to_string(),
1132    })?;
1133    Ok(out)
1134}
1135
1136#[cfg(test)]
1137mod tests {
1138    use super::*;
1139    use crate::security::{AuthenticatedSymbol, SecurityContext};
1140    use crate::types::{Symbol, SymbolId, SymbolKind};
1141
1142    fn sample_proof() -> BondDonorByteProof {
1143        BondDonorByteProof::from_entries(vec![
1144            BondDonorProofEntry::from_bytes(0, "alpha.txt", b"alpha"),
1145            BondDonorProofEntry::from_bytes(1, "dir/beta.txt", b"beta"),
1146        ])
1147        .expect("valid proof")
1148    }
1149
1150    #[test]
1151    fn donor_esi_partition_covers_without_overlap() {
1152        for donor_count in [1, 2, 3, 5, 8] {
1153            let limit = 257u32;
1154            let mut owners = vec![0u32; limit as usize];
1155            for donor_index in 0..donor_count {
1156                for esi in 0..limit {
1157                    if owns_esi(donor_index, donor_count, esi) {
1158                        owners[esi as usize] += 1;
1159                    }
1160                }
1161                let stream = DonorEsiStream::new(donor_index, donor_count).expect("valid stream");
1162                for (seq, esi) in stream.take(32).enumerate() {
1163                    assert_eq!(
1164                        esi,
1165                        esi_for_donor(donor_index, donor_count, seq as u32).expect("esi")
1166                    );
1167                    assert!(owns_esi(donor_index, donor_count, esi));
1168                }
1169            }
1170            assert!(owners.iter().all(|owner_count| *owner_count == 1));
1171        }
1172    }
1173
1174    #[test]
1175    fn n_one_owns_every_esi() {
1176        for esi in 0..512 {
1177            assert!(owns_esi(0, 1, esi));
1178            assert_eq!(esi_for_donor(0, 1, esi).expect("esi"), esi);
1179        }
1180    }
1181
1182    #[test]
1183    fn donor_esi_partition_rejects_invalid_or_overflowing_inputs() {
1184        assert_eq!(
1185            DonorEsiStream::new(0, 0).unwrap_err(),
1186            ChannelBondingError::InvalidDonorCount { donor_count: 0 }
1187        );
1188        assert_eq!(
1189            esi_for_donor(2, 2, 0).unwrap_err(),
1190            ChannelBondingError::InvalidDonorIndex {
1191                donor_index: 2,
1192                donor_count: 2,
1193            }
1194        );
1195        assert_eq!(
1196            esi_for_donor(u32::MAX - 1, u32::MAX, 1).unwrap_err(),
1197            ChannelBondingError::EsiOverflow
1198        );
1199    }
1200
1201    #[test]
1202    fn donor_esi_stream_yields_final_u32_esi_once() {
1203        let mut stream = DonorEsiStream::new(0, 1).expect("single donor stream");
1204        stream.next_seq = Some(u32::MAX);
1205
1206        assert_eq!(stream.next(), Some(u32::MAX));
1207        assert_eq!(stream.next(), None);
1208        assert_eq!(stream.next(), None);
1209    }
1210
1211    #[test]
1212    fn descriptor_accepts_matching_donor_proof() {
1213        let proof = sample_proof();
1214        let descriptor = BondTransferDescriptor::from_donor_proof(
1215            "sample",
1216            true,
1217            1200,
1218            256 * 1024,
1219            Some("env:ATP_BOND_AUTH_KEY".to_string()),
1220            proof.entries.clone(),
1221        )
1222        .expect("descriptor");
1223
1224        descriptor
1225            .verify_donor_byte_match(&proof)
1226            .expect("matching donor proof accepted");
1227    }
1228
1229    #[test]
1230    fn descriptor_rejects_content_mismatch_fail_closed() {
1231        let proof = sample_proof();
1232        let descriptor = BondTransferDescriptor::from_donor_proof(
1233            "sample",
1234            true,
1235            1200,
1236            256 * 1024,
1237            None,
1238            proof.entries.clone(),
1239        )
1240        .expect("descriptor");
1241        let tampered = BondDonorByteProof::from_entries(vec![
1242            BondDonorProofEntry::from_bytes(0, "alpha.txt", b"alpha"),
1243            BondDonorProofEntry::from_bytes(1, "dir/beta.txt", b"tampered"),
1244        ])
1245        .expect("tampered proof is internally consistent");
1246
1247        let err = descriptor
1248            .verify_donor_byte_match(&tampered)
1249            .expect_err("mismatch rejected");
1250        assert!(matches!(
1251            err,
1252            ChannelBondingError::TransferIdMismatch { .. }
1253                | ChannelBondingError::MerkleRootMismatch { .. }
1254                | ChannelBondingError::EntryMismatch { .. }
1255        ));
1256    }
1257
1258    #[test]
1259    fn descriptor_rejects_same_sha_with_wrong_content_id_merkle() {
1260        let proof = sample_proof();
1261        let descriptor = BondTransferDescriptor::from_donor_proof(
1262            "sample",
1263            true,
1264            1200,
1265            256 * 1024,
1266            None,
1267            proof.entries.clone(),
1268        )
1269        .expect("descriptor");
1270        let mut forged_entries = proof.entries.clone();
1271        forged_entries[0].content_id_hex = "11".repeat(32);
1272        let forged = BondDonorByteProof {
1273            transfer_id: proof.transfer_id,
1274            total_bytes: proof.total_bytes,
1275            merkle_root_hex: proof.merkle_root_hex,
1276            entries: forged_entries,
1277        };
1278
1279        let err = descriptor
1280            .verify_donor_byte_match(&forged)
1281            .expect_err("wrong content id rejected");
1282        assert!(matches!(
1283            err,
1284            ChannelBondingError::MerkleRootMismatch { .. }
1285        ));
1286    }
1287
1288    #[test]
1289    fn entry_object_id_is_deterministic_and_entry_specific() {
1290        let transfer = "0123456789abcdef0123456789abcdef";
1291        let first = entry_object_id(transfer, 0);
1292        assert_eq!(first, entry_object_id(transfer, 0));
1293        assert_ne!(first, entry_object_id(transfer, 1));
1294        assert_ne!(first, entry_object_id("different", 0));
1295    }
1296
1297    #[test]
1298    fn assignment_windows_restrict_static_residue() {
1299        let assignment = DonorAssignment {
1300            donor_index: 1,
1301            donor_count: 3,
1302            esi_windows: Some(vec![EsiWindow::new(0, 10)]),
1303            receiver_udp_endpoints: Vec::new(),
1304            auth_key_ref: Some("keyfile:/run/asupersync/bond.key".to_string()),
1305        };
1306        assignment.validate().expect("valid assignment");
1307        assert!(assignment.owns_esi(1));
1308        assert!(assignment.owns_esi(4));
1309        assert!(!assignment.owns_esi(10));
1310        assert!(!assignment.owns_esi(2));
1311    }
1312
1313    #[test]
1314    fn assignment_windows_are_exact_static_residue_intersections() {
1315        let windows = vec![EsiWindow::new(2, 11), EsiWindow::new(15, 22)];
1316        for donor_count in [2, 3, 5, 8] {
1317            for donor_index in 0..donor_count {
1318                let assignment = DonorAssignment {
1319                    donor_index,
1320                    donor_count,
1321                    esi_windows: Some(windows.clone()),
1322                    receiver_udp_endpoints: Vec::new(),
1323                    auth_key_ref: None,
1324                };
1325                assignment.validate().expect("valid assignment");
1326
1327                for esi in 0..25 {
1328                    let expected = owns_esi(donor_index, donor_count, esi)
1329                        && windows.iter().any(|window| window.contains(esi));
1330                    assert_eq!(
1331                        assignment.owns_esi(esi),
1332                        expected,
1333                        "donor {donor_index}/{donor_count} ownership mismatch for esi {esi}"
1334                    );
1335                }
1336            }
1337        }
1338    }
1339
1340    #[test]
1341    fn assignment_validation_rejects_invalid_window_and_donor_ceiling() {
1342        let invalid_window = DonorAssignment {
1343            donor_index: 0,
1344            donor_count: 1,
1345            esi_windows: Some(vec![EsiWindow::new(5, 5)]),
1346            receiver_udp_endpoints: Vec::new(),
1347            auth_key_ref: None,
1348        };
1349        assert_eq!(
1350            invalid_window.validate().unwrap_err(),
1351            ChannelBondingError::InvalidEsiWindow {
1352                start_inclusive: 5,
1353                end_exclusive: 5,
1354            }
1355        );
1356
1357        let too_many_donors = DonorAssignment {
1358            donor_index: 0,
1359            donor_count: MAX_STATIC_RESIDUE_DONORS + 1,
1360            esi_windows: None,
1361            receiver_udp_endpoints: Vec::new(),
1362            auth_key_ref: None,
1363        };
1364        assert_eq!(
1365            too_many_donors.validate().unwrap_err(),
1366            ChannelBondingError::TooManyDonors {
1367                donor_count: MAX_STATIC_RESIDUE_DONORS + 1,
1368                max: MAX_STATIC_RESIDUE_DONORS,
1369            }
1370        );
1371    }
1372
1373    #[test]
1374    fn handshake_degrades_to_static_residue_when_dynamic_missing() {
1375        let receiver = BondingHandshake::v1_static(
1376            [BondTransport::DirectIp, BondTransport::Tailscale],
1377            16,
1378            true,
1379        )
1380        .with_dynamic_windows(true)
1381        .with_resume(true)
1382        .with_extension_capability("phase-e.dynamic-window-v1");
1383        let donor = BondingHandshake::v1_static([BondTransport::Tailscale], 8, false);
1384
1385        let agreement = receiver.negotiate(&donor).expect("compatible");
1386        assert_eq!(agreement.protocol_version, BONDING_PROTOCOL_VERSION);
1387        assert_eq!(
1388            agreement.assignment_mode,
1389            BondingAssignmentMode::StaticResidue
1390        );
1391        assert!(!agreement.resume_supported);
1392        assert!(agreement.auth_required);
1393        assert_eq!(agreement.max_donor_count, 8);
1394        assert_eq!(
1395            agreement.supported_transports,
1396            BTreeSet::from([BondTransport::Tailscale])
1397        );
1398        assert!(agreement.supports_transport(BondTransport::Tailscale));
1399        assert!(!agreement.supports_transport(BondTransport::DirectIp));
1400    }
1401
1402    #[test]
1403    fn handshake_selects_dynamic_windows_when_both_peers_support_them() {
1404        let receiver = BondingHandshake::v1_static([BondTransport::DirectIp], 16, true)
1405            .with_dynamic_windows(true)
1406            .with_resume(true)
1407            .with_extension_capability("receiver-private-future");
1408        let donor = BondingHandshake::v1_static([BondTransport::DirectIp], 12, true)
1409            .with_dynamic_windows(true)
1410            .with_resume(true)
1411            .with_extension_capability("donor-private-future");
1412
1413        let agreement = receiver.negotiate(&donor).expect("compatible");
1414
1415        assert_eq!(
1416            agreement.assignment_mode,
1417            BondingAssignmentMode::DynamicWindows
1418        );
1419        assert!(agreement.resume_supported);
1420        assert_eq!(agreement.max_donor_count, 12);
1421        assert!(agreement.supports_transport(BondTransport::DirectIp));
1422    }
1423
1424    #[test]
1425    fn handshake_ignores_unknown_extensions_but_refuses_no_common_transport() {
1426        let receiver = BondingHandshake::v1_static([BondTransport::DirectIp], 16, true)
1427            .with_extension_capability("unknown.receiver.future");
1428        let donor = BondingHandshake::v1_static([BondTransport::Tailscale], 16, true)
1429            .with_extension_capability("unknown.donor.future");
1430
1431        let err = receiver.negotiate(&donor).expect_err("no common transport");
1432        assert_eq!(err, ChannelBondingError::NoCommonTransport);
1433    }
1434
1435    #[test]
1436    fn handshake_refuses_incompatible_versions() {
1437        let receiver = BondingHandshake::v1_static([BondTransport::DirectIp], 16, true)
1438            .with_protocol_range(2, 2);
1439        let donor = BondingHandshake::v1_static([BondTransport::DirectIp], 16, true);
1440
1441        let err = receiver.negotiate(&donor).expect_err("version mismatch");
1442        assert!(matches!(
1443            err,
1444            ChannelBondingError::IncompatibleProtocolVersion { .. }
1445        ));
1446    }
1447
1448    #[test]
1449    fn handshake_refuses_invalid_version_zero_and_donor_ceiling() {
1450        let donor = BondingHandshake::v1_static([BondTransport::DirectIp], 16, true);
1451        let version_zero = BondingHandshake::v1_static([BondTransport::DirectIp], 16, true)
1452            .with_protocol_range(0, BONDING_PROTOCOL_VERSION);
1453        let err = version_zero.negotiate(&donor).expect_err("version zero");
1454        assert_eq!(
1455            err,
1456            ChannelBondingError::InvalidProtocolRange {
1457                min: 0,
1458                max: BONDING_PROTOCOL_VERSION,
1459            }
1460        );
1461
1462        let zero_donors = BondingHandshake::v1_static([BondTransport::DirectIp], 0, true);
1463        let err = zero_donors.negotiate(&donor).expect_err("zero donors");
1464        assert_eq!(
1465            err,
1466            ChannelBondingError::InvalidDonorCount { donor_count: 0 }
1467        );
1468
1469        let too_many_donors = BondingHandshake::v1_static(
1470            [BondTransport::DirectIp],
1471            MAX_STATIC_RESIDUE_DONORS + 1,
1472            true,
1473        );
1474        let err = too_many_donors
1475            .negotiate(&donor)
1476            .expect_err("donor ceiling exceeds phase-a cap");
1477        assert_eq!(
1478            err,
1479            ChannelBondingError::TooManyDonors {
1480                donor_count: MAX_STATIC_RESIDUE_DONORS + 1,
1481                max: MAX_STATIC_RESIDUE_DONORS,
1482            }
1483        );
1484    }
1485
1486    #[test]
1487    fn security_model_rejects_argv_and_disabled_fail_closed_layers() {
1488        let mut model = BondingSecurityModel {
1489            auth_key: BondAuthKeyRef {
1490                key_id: "bond-key-1".to_string(),
1491                control_plane: BondAuthControlPlane::Ssh,
1492                location: BondAuthKeyLocation::EnvVar("ATP_BOND_AUTH_KEY".to_string()),
1493            },
1494            auth_before_decode: true,
1495            fail_closed_on_merkle_mismatch: true,
1496        };
1497        model.validate().expect("valid security model");
1498
1499        model.auth_key.location = BondAuthKeyLocation::Argv("--bond-key=secret".to_string());
1500        assert_eq!(
1501            model.validate().unwrap_err(),
1502            ChannelBondingError::InsecureAuthKeyDelivery
1503        );
1504
1505        model.auth_key.location = BondAuthKeyLocation::EnvVar("ATP_BOND_AUTH_KEY".to_string());
1506        model.auth_before_decode = false;
1507        assert_eq!(
1508            model.validate().unwrap_err(),
1509            ChannelBondingError::AuthBeforeDecodeDisabled
1510        );
1511
1512        model.auth_before_decode = true;
1513        model.fail_closed_on_merkle_mismatch = false;
1514        assert_eq!(
1515            model.validate().unwrap_err(),
1516            ChannelBondingError::MerkleFailClosedDisabled
1517        );
1518    }
1519
1520    #[test]
1521    fn wrong_donor_auth_key_rejects_symbol_before_decode() {
1522        let signer = SecurityContext::for_testing(0xA11CE);
1523        let verifier = SecurityContext::for_testing(0xB0B);
1524        let symbol = Symbol::new(
1525            SymbolId::new_for_test(7, 0, 3),
1526            b"bonded repair".to_vec(),
1527            SymbolKind::Repair,
1528        );
1529        let signed = signer.sign_symbol(&symbol);
1530        let tag = *signed.tag();
1531        let mut received = AuthenticatedSymbol::from_parts(signed.into_symbol(), tag);
1532
1533        let err = verifier
1534            .verify_authenticated_symbol(&mut received)
1535            .expect_err("wrong shared donor key must reject symbol");
1536        assert!(err.is_invalid_tag());
1537        assert!(!received.is_verified());
1538    }
1539}
asupersync/net/atp/channel_bonding.rs

asupersync/net/atp/
channel_bonding.rs
