heddle_thread_api/replication/
opening.rs

1//! Shared negotiation for already-authorized, already-resolved Threads.
2//! This validates transport bindings and limits; it does not grant authority.
3use std::collections::BTreeSet;
4
5use crypto::{Signer, thread_operation::SignedGenesis};
6use heddle_object_model::object::thread_replication::{
7    GENESIS_FORMAT, OPERATION_FORMAT, ThreadFacet, ThreadGenesis,
8};
9
10use crate::{contract::*, transport::Error};
11
12pub const FRAME_LIMIT: usize = 512 * 1024;
13pub const MAX_ITEMS: u32 = 64;
14
15pub fn validate_endpoint(endpoint: &EndpointRef) -> Result<(), Error> {
16    if endpoint.public_key.len() != 32
17        || !matches!(
18            EndpointKind::try_from(endpoint.kind),
19            Ok(EndpointKind::Device | EndpointKind::Weft)
20        )
21    {
22        return Err(Error::Protocol("invalid replication endpoint"));
23    }
24    Ok(())
25}
26
27pub fn parse_facets(values: &[i32]) -> Result<BTreeSet<ThreadFacet>, Error> {
28    if values.is_empty() || values.len() > ThreadFacet::ALL.len() {
29        return Err(Error::Protocol(
30            "replication requires a bounded set of distinct facets",
31        ));
32    }
33    let facets = values
34        .iter()
35        .map(|value| {
36            super::native_facet(*value)
37                .map_err(|_| Error::Protocol("unsupported replication facet"))
38        })
39        .collect::<Result<BTreeSet<_>, _>>()?;
40    if facets.len() != values.len() {
41        return Err(Error::Protocol("duplicate replication facet"));
42    }
43    Ok(facets)
44}
45
46/// The remote key must come from the authenticated transport, never the frame.
47/// An attached genesis retains the original creator's signature and identity.
48/// The host must authorize publication and commit that genesis before Ready;
49/// parsing it grants no authority. PoP covers the entire exact opening.
50pub fn accept(
51    open: &ReplicationOpen,
52    thread: &ThreadRef,
53    local: &EndpointRef,
54    remote_key: [u8; 32],
55    admission: &BTreeSet<ThreadFacet>,
56    sharing_policy_version: Vec<u8>,
57) -> Result<AcceptedOpening, Error> {
58    validate_endpoint(local)?;
59    if open.thread.as_ref() != Some(thread) {
60        return Err(Error::Protocol(
61            "replication requires an already resolved Thread",
62        ));
63    }
64    let source = open
65        .source
66        .as_ref()
67        .ok_or(Error::Protocol("opening requires source endpoint"))?;
68    validate_endpoint(source)?;
69    if source.public_key != remote_key || open.destination.as_ref() != Some(local) {
70        return Err(Error::Protocol(
71            "opening endpoints differ from Iroh connection",
72        ));
73    }
74    if open.session_nonce.len() != 16
75        || !open
76            .record_formats
77            .iter()
78            .any(|format| format == OPERATION_FORMAT)
79    {
80        return Err(Error::Protocol("unsupported replication session or format"));
81    }
82    // Native records are indivisible. Negotiate the supported frame size
83    // explicitly instead of promising a smaller budget and exceeding it.
84    if open.budget.as_ref().is_some_and(|budget| {
85        budget.max_frame_bytes != 0 && budget.max_frame_bytes != FRAME_LIMIT as u32
86    }) {
87        return Err(Error::Protocol("unsupported replication frame budget"));
88    }
89    let facets: BTreeSet<_> = parse_facets(&open.facets)?
90        .intersection(admission)
91        .copied()
92        .collect();
93    if facets.is_empty() {
94        return Err(Error::Protocol("no authorized replication facets"));
95    }
96    let requested = open.budget.as_ref().map_or(0, |budget| budget.max_items);
97    let genesis = open
98        .thread_genesis
99        .as_ref()
100        .map(|record| verify_genesis_record(record, thread))
101        .transpose()?;
102    Ok(AcceptedOpening {
103        genesis,
104        genesis_record: open.thread_genesis.clone(),
105        ready: ReplicationReady {
106            thread: Some(thread.clone()),
107            endpoint: Some(local.clone()),
108            facets: facets.into_iter().map(super::wire_facet).collect(),
109            sharing_policy_version,
110            budget: Some(ReadBudget {
111                max_items: if requested == 0 {
112                    MAX_ITEMS
113                } else {
114                    requested.min(MAX_ITEMS)
115                },
116                max_frame_bytes: FRAME_LIMIT as u32,
117                max_snapshot_bytes: 0,
118            }),
119            record_formats: vec![OPERATION_FORMAT.into()],
120        },
121    })
122}
123
124/// Parsed proposal; authorization and durable installation belong to the host.
125pub struct AcceptedOpening {
126    pub ready: ReplicationReady,
127    pub genesis: Option<ThreadGenesis>,
128    pub genesis_record: Option<ThreadGenesisRecord>,
129}
130
131/// Structural and original-signature validation only. Account authority and
132/// hosted admission receipts still require independently retained trust.
133pub fn verify_genesis_record(
134    record: &ThreadGenesisRecord,
135    thread: &ThreadRef,
136) -> Result<ThreadGenesis, Error> {
137    use prost::Message;
138    if record.boundary_acceptances.len() > crate::boundary_acceptance::MAX_ACCEPTANCES
139        || record.encoded_len() > 256 * 1024
140    {
141        return Err(Error::Protocol("genesis wrapper evidence exceeds bounds"));
142    }
143    let signed = record
144        .genesis
145        .as_ref()
146        .ok_or(Error::Protocol("original signed genesis missing"))?;
147    let genesis = verify_genesis(signed, thread)?;
148    if record.creator_authority.len() > 64 * 1024 {
149        return Err(Error::Protocol("creator authority exceeds bound"));
150    }
151    use heddle_object_model::object::thread_replication::GenesisOwner;
152    match genesis.owner {
153        GenesisOwner::LocalKey(_)
154            if !record.creator_authority.is_empty() || record.admission.is_some() =>
155        {
156            return Err(Error::Protocol(
157                "local-key ownership requires an explicit claim, not an account envelope",
158            ));
159        }
160        GenesisOwner::Account(_) if record.creator_authority.is_empty() => {
161            return Err(Error::Protocol(
162                "account-owned genesis requires original creator authority",
163            ));
164        }
165        _ => {}
166    }
167    super::ownership::verify_claims(record, &genesis)?;
168    Ok(genesis)
169}
170
171pub fn sign_genesis(genesis: &ThreadGenesis, signer: &impl Signer) -> Result<SignedRecord, Error> {
172    let signed =
173        SignedGenesis::sign(genesis, signer).map_err(|error| Error::Io(error.to_string()))?;
174    Ok(SignedRecord {
175        format: GENESIS_FORMAT.into(),
176        canonical_record: signed.canonical,
177        signatures: vec![RecordSignature {
178            public_key: genesis.creator.to_vec(),
179            signature: signed.signature,
180        }],
181    })
182}
183
184pub fn verify_genesis(record: &SignedRecord, thread: &ThreadRef) -> Result<ThreadGenesis, Error> {
185    if record.format != GENESIS_FORMAT || record.signatures.len() != 1 {
186        return Err(Error::Protocol("unsupported Thread genesis record"));
187    }
188    let genesis = SignedGenesis {
189        canonical: record.canonical_record.clone(),
190        signature: record.signatures[0].signature.clone(),
191    }
192    .verify()
193    .map_err(|_| Error::Protocol("invalid Thread genesis signature"))?;
194    let id = genesis
195        .id()
196        .map_err(|_| Error::Protocol("invalid Thread genesis"))?;
197    if record.signatures[0].public_key != genesis.creator
198        || thread
199            .spool
200            .as_ref()
201            .is_none_or(|spool| spool.id != genesis.spool)
202        || thread
203            .id
204            .as_ref()
205            .is_none_or(|thread| thread.value != id.as_bytes())
206    {
207        return Err(Error::Protocol(
208            "Thread genesis identity does not match opening",
209        ));
210    }
211    Ok(genesis)
212}
213
214pub fn validate_ready(
215    ready: &ReplicationReady,
216    thread: &ThreadRef,
217    destination: &EndpointRef,
218    requested_facets: &BTreeSet<ThreadFacet>,
219    requested_max_items: u32,
220) -> Result<(BTreeSet<ThreadFacet>, usize), Error> {
221    validate_endpoint(destination)?;
222    if ready.endpoint.as_ref() != Some(destination)
223        || ready.thread.as_ref() != Some(thread)
224        || ready.record_formats != [OPERATION_FORMAT]
225    {
226        return Err(Error::Protocol(
227            "replication Ready binding differs from opening",
228        ));
229    }
230    let facets = parse_facets(&ready.facets)?;
231    if !facets.is_subset(requested_facets) {
232        return Err(Error::Protocol("replication Ready widened admission scope"));
233    }
234    let budget = ready
235        .budget
236        .as_ref()
237        .ok_or(Error::Protocol("replication Ready requires budget"))?;
238    let ceiling = if requested_max_items == 0 {
239        MAX_ITEMS
240    } else {
241        requested_max_items.min(MAX_ITEMS)
242    };
243    if budget.max_items == 0
244        || budget.max_items > ceiling
245        || budget.max_frame_bytes != FRAME_LIMIT as u32
246    {
247        return Err(Error::Protocol("unsupported replication Ready budget"));
248    }
249    Ok((facets, budget.max_items as usize))
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255
256    #[test]
257    fn first_publication_retains_signed_genesis_and_rejects_changed_identity() {
258        use crypto::{Ed25519Signer, Signer};
259        use heddle_object_model::object::{
260            StateId,
261            thread_replication::{GENESIS_FORMAT, ThreadGenesis},
262        };
263        let signer = Ed25519Signer::from_seed(&[23; 32]).expect("origin device");
264        let genesis = ThreadGenesis {
265            version: 1,
266            spool: "01980000-0000-7000-8000-000000000001".into(),
267            parent: None,
268            base: StateId::from_bytes([1; 32]),
269            name: "original".into(),
270            intent: "publish once".into(),
271            creator: signer.public_key().try_into().expect("public key"),
272            owner: heddle_object_model::object::thread_replication::GenesisOwner::LocalKey(
273                signer.public_key().try_into().expect("public key"),
274            ),
275            nonce: vec![2; 16],
276        };
277        let canonical = genesis.encode().expect("canonical genesis");
278        let mut signing = GENESIS_FORMAT.as_bytes().to_vec();
279        signing.push(0);
280        signing.extend(&canonical);
281        let signed = SignedRecord {
282            format: GENESIS_FORMAT.into(),
283            canonical_record: canonical,
284            signatures: vec![RecordSignature {
285                public_key: signer.public_key().to_vec(),
286                signature: signer.sign(&signing).expect("origin signature"),
287            }],
288        };
289        let thread = ThreadRef {
290            spool: Some(SpoolRef {
291                id: genesis.spool.clone(),
292            }),
293            id: Some(ThreadId {
294                value: genesis.id().expect("Thread ID").as_bytes().to_vec(),
295            }),
296        };
297        let local = EndpointRef {
298            public_key: vec![7; 32],
299            kind: EndpointKind::Weft as i32,
300        };
301        let open = ReplicationOpen {
302            thread: Some(thread.clone()),
303            thread_genesis: Some(ThreadGenesisRecord {
304                boundary_acceptances: Vec::new(),
305                ownership_claims: vec![],
306                ownership_claim_admissions: vec![],
307                ownership_resolutions: vec![],
308                ownership_resolution_admissions: vec![],
309                genesis: Some(signed),
310                creator_authority: vec![],
311                admission: None,
312            }),
313            source: Some(EndpointRef {
314                public_key: vec![8; 32],
315                kind: EndpointKind::Device as i32,
316            }),
317            destination: Some(local.clone()),
318            facets: vec![SharedFacet::Source as i32],
319            session_nonce: vec![9; 16],
320            record_formats: vec![OPERATION_FORMAT.into()],
321            ..Default::default()
322        };
323        let facets = BTreeSet::from([ThreadFacet::Source]);
324        assert!(
325            accept(&open, &thread, &local, [8; 32], &facets, vec![]).is_ok(),
326            "first publication must accept the original signed genesis"
327        );
328        let mut changed = open.clone();
329        changed
330            .thread_genesis
331            .as_mut()
332            .expect("genesis")
333            .genesis
334            .as_mut()
335            .expect("signed genesis")
336            .signatures[0]
337            .signature[0] ^= 1;
338        assert!(accept(&changed, &thread, &local, [8; 32], &facets, vec![]).is_err());
339        changed = open.clone();
340        changed
341            .thread
342            .as_mut()
343            .expect("Thread")
344            .id
345            .as_mut()
346            .expect("ID")
347            .value[0] ^= 1;
348        let claimed = changed.thread.clone().expect("claimed Thread");
349        assert!(accept(&changed, &claimed, &local, [8; 32], &facets, vec![]).is_err());
350        changed = open;
351        changed
352            .thread
353            .as_mut()
354            .expect("Thread")
355            .spool
356            .as_mut()
357            .expect("spool")
358            .id = "another".into();
359        let claimed = changed.thread.clone().expect("claimed Thread");
360        assert!(accept(&changed, &claimed, &local, [8; 32], &facets, vec![]).is_err());
361    }
362
363    #[test]
364    fn opening_binds_both_endpoints_thread_formats_and_negotiated_limits() {
365        let thread = ThreadRef {
366            spool: Some(SpoolRef { id: "spool".into() }),
367            id: Some(ThreadId { value: vec![3; 32] }),
368        };
369        let local = EndpointRef {
370            kind: EndpointKind::Weft as i32,
371            public_key: vec![1; 32],
372        };
373        let source = EndpointRef {
374            kind: EndpointKind::Device as i32,
375            public_key: vec![2; 32],
376        };
377        let facets = BTreeSet::from([ThreadFacet::Source, ThreadFacet::Discussion]);
378        let open = ReplicationOpen {
379            thread: Some(thread.clone()),
380            source: Some(source),
381            destination: Some(local.clone()),
382            facets: facets
383                .iter()
384                .copied()
385                .map(super::super::wire_facet)
386                .collect(),
387            session_nonce: vec![4; 16],
388            record_formats: vec![OPERATION_FORMAT.into()],
389            budget: Some(ReadBudget {
390                max_items: 1,
391                max_frame_bytes: FRAME_LIMIT as u32,
392                max_snapshot_bytes: 0,
393            }),
394            ..Default::default()
395        };
396        let allowed = BTreeSet::from([ThreadFacet::Source]);
397        let ready = accept(&open, &thread, &local, [2; 32], &allowed, vec![5; 32])
398            .expect("authorized opening")
399            .ready;
400        assert_eq!(
401            validate_ready(&ready, &thread, &local, &facets, 1).expect("bound ready"),
402            (allowed.clone(), 1)
403        );
404        assert_eq!(ready.sharing_policy_version, vec![5; 32]);
405        assert!(accept(&open, &thread, &local, [7; 32], &allowed, vec![]).is_err());
406        let mut changed = open.clone();
407        changed
408            .destination
409            .as_mut()
410            .expect("destination")
411            .public_key = vec![8; 32];
412        assert!(accept(&changed, &thread, &local, [2; 32], &allowed, vec![]).is_err());
413        changed = open.clone();
414        changed
415            .thread
416            .as_mut()
417            .expect("thread")
418            .id
419            .as_mut()
420            .expect("id")
421            .value = vec![8; 32];
422        assert!(accept(&changed, &thread, &local, [2; 32], &allowed, vec![]).is_err());
423        changed = open.clone();
424        changed.facets = vec![SharedFacet::Source as i32; 2];
425        assert!(accept(&changed, &thread, &local, [2; 32], &allowed, vec![]).is_err());
426        changed = open.clone();
427        changed.record_formats.clear();
428        assert!(accept(&changed, &thread, &local, [2; 32], &allowed, vec![]).is_err());
429        changed = open.clone();
430        changed.budget.as_mut().expect("budget").max_frame_bytes = 128;
431        assert!(accept(&changed, &thread, &local, [2; 32], &allowed, vec![]).is_err());
432        let mut widened = ready.clone();
433        widened.budget.as_mut().expect("budget").max_items = 2;
434        assert!(validate_ready(&widened, &thread, &local, &facets, 1).is_err());
435        widened = ready;
436        widened.facets.push(SharedFacet::Collaboration as i32);
437        assert!(validate_ready(&widened, &thread, &local, &allowed, 1).is_err());
438    }
439}
heddle_thread_api/replication/opening.rs

heddle_thread_api/replication/
opening.rs
