heddle_thread_api/
authority_admission.rs

1#[path = "authority_batches.rs"]
2mod batching;
3pub use batching::{AuthorityBatches, batches};
4// Hosted original-author receipts require receiver-owned executor trust.
5pub use crypto::thread_authority_admission::SignedAuthorityAdmission;
6use crypto::{Signer, thread_operation::SignedOperation};
7use heddle_object_model::object::thread_authority_admission::FORMAT;
8pub use heddle_object_model::object::{
9    thread_authority_admission::ThreadAuthorityAdmission,
10    thread_replication::integration::TrustedHostedExecutor,
11};
12
13use crate::{contract as wire, transport::Error};
14
15/// Issuers persist this exact statement with original bytes at first admission.
16pub fn sign(
17    value: &ThreadAuthorityAdmission,
18    signer: &impl Signer,
19) -> Result<wire::SignedRecord, Error> {
20    encode(
21        &SignedAuthorityAdmission::sign(value, signer)
22            .map_err(|_| Error::Protocol("authority admission signing failed"))?,
23    )
24}
25/// Verify original and executor signatures against independently admitted trust.
26pub fn verify(
27    receipt: &wire::SignedRecord,
28    original: &SignedOperation,
29    trust: &TrustedHostedExecutor,
30) -> Result<ThreadAuthorityAdmission, Error> {
31    decode(receipt)?.verify(original, trust).map_err(|_| {
32        Error::Protocol("authority receipt differs from original operation or pinned executor")
33    })
34}
35/// Signature-only decoding for sidecar routing; this does not admit anything.
36pub fn verify_signature(receipt: &wire::SignedRecord) -> Result<ThreadAuthorityAdmission, Error> {
37    decode(receipt)?
38        .verify_signature()
39        .map_err(|_| Error::Protocol("invalid authority admission signature"))
40}
41pub fn decode(receipt: &wire::SignedRecord) -> Result<SignedAuthorityAdmission, Error> {
42    if receipt.format != FORMAT {
43        return Err(Error::Protocol("invalid authority admission format"));
44    }
45    let [signature] = receipt.signatures.as_slice() else {
46        return Err(Error::Protocol(
47            "authority admission requires one executor signature",
48        ));
49    };
50    let value = ThreadAuthorityAdmission::decode(&receipt.canonical_record)
51        .map_err(|_| Error::Protocol("invalid canonical authority admission"))?;
52    if signature.public_key != value.executor {
53        return Err(Error::Protocol(
54            "authority admission signature key differs from executor",
55        ));
56    }
57    let signed = SignedAuthorityAdmission {
58        boundary_acceptance: None,
59        canonical: receipt.canonical_record.clone(),
60        signature: signature.signature.clone(),
61    };
62    signed
63        .verify_signature()
64        .map_err(|_| Error::Protocol("invalid authority admission signature"))?;
65    Ok(signed)
66}
67pub fn encode(receipt: &SignedAuthorityAdmission) -> Result<wire::SignedRecord, Error> {
68    let value = receipt
69        .verify_signature()
70        .map_err(|_| Error::Protocol("invalid authority admission signature"))?;
71    Ok(wire::SignedRecord {
72        format: FORMAT.into(),
73        canonical_record: receipt.canonical.clone(),
74        signatures: vec![wire::RecordSignature {
75            public_key: value.executor.to_vec(),
76            signature: receipt.signature.clone(),
77        }],
78    })
79}
80
81/// Decode a bounded batch and match every sidecar to an exact original in that
82/// batch. This verifies signatures and immutable bindings, not issuer trust.
83/// Callers still authorize current disclosure and independently pin receipts.
84pub fn match_batch(
85    batch: &wire::ReplicationOperations,
86) -> Result<Vec<crate::replication::store::ReceivedOperation>, Error> {
87    use std::collections::{BTreeMap, BTreeSet};
88
89    use prost::Message;
90    crate::hybrid::operations(batch).map_err(Error::Protocol)?;
91    let import_authority = batch.import_authority.clone().map(std::sync::Arc::new);
92    if batch.operations.is_empty()
93        || batch.operations.len() > 128
94        || batch.authority_admissions.len() > 128
95        || batch.encoded_len() > 1024 * 1024
96    {
97        return Err(Error::Protocol("original authority batch exceeds bounds"));
98    }
99    let mut evidence = crate::boundary_acceptance::Evidence::default();
100    evidence.add(&batch.boundary_acceptances)?;
101    let mut receipts = BTreeMap::new();
102    for record in &batch.authority_admissions {
103        let mut signed = decode(record)?;
104        let statement = signed
105            .verify_signature()
106            .map_err(|_| Error::Protocol("invalid authority admission signature"))?;
107        signed.boundary_acceptance = evidence.matched(&statement.basis)?;
108        let operation_id = statement.subject.operation_id().ok_or(Error::Protocol(
109            "operation batch cannot carry ownership claim admission",
110        ))?;
111        if receipts.insert(operation_id, (signed, statement)).is_some() {
112            return Err(Error::Protocol("duplicate authority admission sidecar"));
113        }
114    }
115    let mut ids = BTreeSet::new();
116    let mut output = Vec::new();
117    for record in &batch.operations {
118        let original = crate::replication::decode_record(record.clone())
119            .map_err(|_| Error::Protocol("invalid original operation signature"))?;
120        let operation = original
121            .verify()
122            .map_err(|_| Error::Protocol("invalid original operation signature"))?;
123        let id = operation
124            .id()
125            .map_err(|_| Error::Protocol("invalid original operation identity"))?;
126        if !ids.insert(id) {
127            return Err(Error::Protocol("duplicate original operation"));
128        }
129        let receipt = if let Some((receipt, statement)) = receipts.remove(&id) {
130            receipt
131                .verify(
132                    &original,
133                    &TrustedHostedExecutor {
134                        spool: statement.spool,
135                        spool_genesis: statement.spool_genesis,
136                        executor: statement.executor,
137                    },
138                )
139                .map_err(|_| {
140                    Error::Protocol("authority receipt differs from original operation")
141                })?;
142            Some(receipt)
143        } else {
144            None
145        };
146        output.push(crate::replication::store::ReceivedOperation {
147            native_authority: batch.native_authority.clone().map(std::sync::Arc::new),
148            original,
149            authority_admission: receipt,
150            import_authority: import_authority.clone(),
151        });
152    }
153    if !receipts.is_empty() {
154        return Err(Error::Protocol("unmatched authority admission sidecar"));
155    }
156    evidence.finish()?;
157    Ok(output)
158}
159
160#[cfg(test)]
161mod tests {
162    use crypto::Ed25519Signer;
163    use heddle_object_model::object::{
164        CollaborationActor, ContentHash,
165        thread_authority_admission::MAX_BYTES,
166        thread_replication::{
167            ThreadOperation, ThreadOperationBody,
168            metadata::{AUTHORITY_FORMAT, Control, ThreadControl},
169        },
170    };
171    use uuid::Uuid;
172
173    use super::*;
174
175    fn fixture() -> (
176        ThreadAuthorityAdmission,
177        SignedOperation,
178        TrustedHostedExecutor,
179        Ed25519Signer,
180    ) {
181        let author = Ed25519Signer::from_seed(&[41; 32]).expect("author");
182        let executor = Ed25519Signer::from_seed(&[42; 32]).expect("executor");
183        let envelope = b"original authority independently checked at first admission".to_vec();
184        let control = ThreadControl {
185            version: 1,
186            spool: Uuid::from_u128(100),
187            actor: CollaborationActor {
188                principal_id: Uuid::from_u128(101),
189                agent_id: Some("original-agent".into()),
190            },
191            authority_digest: ContentHash::compute_typed(AUTHORITY_FORMAT, &envelope),
192            authority_envelope: envelope,
193            client_operation_id: Uuid::from_u128(102),
194            occurred_at_ms: 1,
195            control: Control::Name("original agent work".into()),
196        };
197        let operation = ThreadOperation {
198            version: 1,
199            thread: ContentHash::from_bytes([43; 32]),
200            parents: Default::default(),
201            publisher: author.public_key().try_into().expect("key"),
202            body: ThreadOperationBody::Metadata(control.encode().expect("control")),
203        };
204        let trust = TrustedHostedExecutor {
205            spool: control.spool,
206            spool_genesis: ContentHash::from_bytes([44; 32]),
207            executor: executor.public_key().try_into().expect("key"),
208        };
209        let receipt = ThreadAuthorityAdmission {
210            version: 3,
211            basis: heddle_object_model::object::original_boundary_acceptance::AdmissionBasis::OriginalAuthority,
212            spool: trust.spool,
213            spool_genesis: trust.spool_genesis,
214            thread: operation.thread,
215            subject: heddle_object_model::object::thread_authority_admission::OriginalAuthoritySubject::Operation(operation.id().expect("ID")),
216            actor: control.actor,
217            publisher: operation.publisher,
218            authority_digest: control.authority_digest,
219            executor: trust.executor,
220            admitted_at_ms: 2000,
221        };
222        (
223            receipt,
224            SignedOperation::sign(&operation, &author).expect("original signature"),
225            trust,
226            executor,
227        )
228    }
229    #[test]
230    fn admission_requires_independent_executor_and_exact_original_identity() {
231        let (value, original, trust, executor) = fixture();
232        let signed = sign(&value, &executor).expect("receipt");
233        assert_eq!(
234            verify(&signed, &original, &trust).expect("independent original admission"),
235            value
236        );
237        for wrong in [
238            TrustedHostedExecutor {
239                executor: [45; 32],
240                ..trust.clone()
241            },
242            TrustedHostedExecutor {
243                spool: Uuid::from_u128(999),
244                ..trust.clone()
245            },
246            TrustedHostedExecutor {
247                spool_genesis: ContentHash::from_bytes([46; 32]),
248                ..trust.clone()
249            },
250        ] {
251            assert!(
252                verify(&signed, &original, &wrong).is_err(),
253                "incoming proof never enrolls its own trust"
254            );
255        }
256        let mut mutations = Vec::new();
257        let mut changed = value.clone();
258        changed.subject =
259            heddle_object_model::object::thread_authority_admission::OriginalAuthoritySubject::Operation(
260                ContentHash::from_bytes([47; 32]),
261            );
262        mutations.push(changed);
263        let mut changed = value.clone();
264        changed.thread = ContentHash::from_bytes([48; 32]);
265        mutations.push(changed);
266        let mut changed = value.clone();
267        changed.publisher = [49; 32];
268        mutations.push(changed);
269        let mut changed = value.clone();
270        changed.actor.agent_id = None;
271        mutations.push(changed);
272        let mut changed = value.clone();
273        changed.actor.principal_id = Uuid::from_u128(888);
274        mutations.push(changed);
275        let mut changed = value;
276        changed.authority_digest = ContentHash::from_bytes([50; 32]);
277        mutations.push(changed);
278        for changed in mutations {
279            assert!(
280                verify(
281                    &sign(&changed, &executor).expect("valid executor signature"),
282                    &original,
283                    &trust
284                )
285                .is_err(),
286                "receipt cannot substitute original scope or authorship"
287            );
288        }
289    }
290    #[test]
291    fn admission_verifies_both_signatures_and_canonical_bounds() {
292        let (value, original, trust, executor) = fixture();
293        let signed = sign(&value, &executor).expect("receipt");
294        let mut changed = signed.clone();
295        let mut new_value = value;
296        new_value.admitted_at_ms += 1;
297        changed.canonical_record = new_value.encode().expect("different timestamp");
298        assert!(
299            verify(&changed, &original, &trust).is_err(),
300            "executor signature binds first admission time"
301        );
302        let mut changed = original.clone();
303        changed.signature[0] ^= 1;
304        assert!(
305            verify(&signed, &changed, &trust).is_err(),
306            "receipt never replaces original signature"
307        );
308        let mut changed = signed.clone();
309        changed.signatures.push(changed.signatures[0].clone());
310        assert!(
311            verify(&changed, &original, &trust).is_err(),
312            "exact one executor signature"
313        );
314        let mut changed = signed;
315        changed.canonical_record.resize(MAX_BYTES + 1, 0);
316        assert!(
317            verify(&changed, &original, &trust).is_err(),
318            "bounded before decoding"
319        );
320        assert!(
321            ThreadAuthorityAdmission::decode(&changed.canonical_record)
322                .expect_err("oversized canonical bytes rejected before parsing")
323                .to_string()
324                .contains("byte bound")
325        );
326    }
327}
heddle_thread_api/authority_admission.rs

heddle_thread_api/
authority_admission.rs
