heddle_thread_api/publication/
acceptance.rs

1//! Explicit publication proposals. Signature/manifest validation is deliberately
2//! separate from the host's current accepting-authority and ownership checks.
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    sync::Arc,
6};
7
8use crypto::{Signer, original_boundary_acceptance::SignedBoundaryAcceptance};
9use heddle_object_model::object::{
10    ContentHash, StateId,
11    original_boundary_acceptance::{
12        AdmissionBasis, BoundaryOriginalKind, FORMAT, ManifestSubject, OriginalBoundaryAcceptance,
13        OriginalManifestEntry, OriginalPublicationManifest, PublicationIntent,
14    },
15    thread_replication::{SourceAuthor, ThreadGenesis},
16};
17use prost::Message;
18use uuid::Uuid;
19
20use super::PublicationOriginals;
21use crate::{contract::*, transport::Error};
22
23/// An immutable, signature-checked selection, NOT current permission. A host
24/// must inspect original provenance and verify the accepting capability for
25/// every selected subject under its current resource/ownership fence.
26pub struct ProposedAcceptance {
27    signed: Arc<SignedBoundaryAcceptance>,
28    value: OriginalBoundaryAcceptance,
29    subjects: BTreeSet<ManifestSubject>,
30    manifest: Arc<OriginalPublicationManifest>,
31}
32impl ProposedAcceptance {
33    /// Return only an exact selected descriptor from the full signed manifest.
34    /// The host compares rebuilt original identity/envelope to this value before
35    /// current acceptance verification; an original ID alone is insufficient for
36    /// the unsigned creator-envelope sidecar of an Account genesis.
37    pub fn entry(&self, subject: &ManifestSubject) -> Option<&OriginalManifestEntry> {
38        if !self.subjects.contains(subject) {
39            return None;
40        }
41        self.manifest
42            .entries
43            .binary_search_by(|entry| entry.subject.cmp(subject))
44            .ok()
45            .map(|index| &self.manifest.entries[index])
46    }
47    pub fn signed(&self) -> &Arc<SignedBoundaryAcceptance> {
48        &self.signed
49    }
50    pub fn value(&self) -> &OriginalBoundaryAcceptance {
51        &self.value
52    }
53    pub fn subjects(&self) -> &BTreeSet<ManifestSubject> {
54        &self.subjects
55    }
56}
57
58/// Full immutable publication selection shared by the client and host. Retained
59/// receipts are verified separately and cannot silently become fresh proposals.
60pub struct PublicationAcceptancePlan {
61    intent: PublicationIntent,
62    manifest: Arc<OriginalPublicationManifest>,
63    proposed: BTreeMap<ContentHash, ProposedAcceptance>,
64    subjects: BTreeMap<ManifestSubject, ContentHash>,
65}
66impl PublicationAcceptancePlan {
67    pub fn intent(&self) -> &PublicationIntent {
68        &self.intent
69    }
70    pub fn manifest(&self) -> &OriginalPublicationManifest {
71        &self.manifest
72    }
73    pub fn proposed(&self) -> &BTreeMap<ContentHash, ProposedAcceptance> {
74        &self.proposed
75    }
76    pub fn acceptance_for(&self, subject: &ManifestSubject) -> Option<&ProposedAcceptance> {
77        self.subjects
78            .get(subject)
79            .and_then(|id| self.proposed.get(id))
80    }
81    fn insert(&mut self, signed: SignedBoundaryAcceptance) -> Result<ContentHash, Error> {
82        if self.proposed.len() >= crate::boundary_acceptance::MAX_ACCEPTANCES {
83            return Err(Error::Protocol("fresh acceptance count exceeded"));
84        }
85        let value = signed.verify_signature().map_err(invalid)?;
86        let id = value.id().map_err(invalid)?;
87        if self.proposed.contains_key(&id) {
88            return Err(Error::Protocol("duplicate fresh acceptance"));
89        }
90        let selected = value
91            .selected(&self.intent, &self.manifest)
92            .map_err(invalid)?;
93        let subjects: BTreeSet<_> = selected.iter().map(|entry| entry.subject.clone()).collect();
94        if subjects
95            .iter()
96            .any(|subject| self.subjects.contains_key(subject))
97        {
98            return Err(Error::Protocol("overlapping fresh acceptance selection"));
99        }
100        for subject in &subjects {
101            self.subjects.insert(subject.clone(), id);
102        }
103        self.proposed.insert(
104            id,
105            ProposedAcceptance {
106                signed: Arc::new(signed),
107                value,
108                subjects,
109                manifest: self.manifest.clone(),
110            },
111        );
112        Ok(id)
113    }
114}
115
116/// Owned preparation for an explicit local prepare/sign/send workflow. Preparing
117/// or signing never refreshes identity, contacts Weft, or modifies an original.
118pub struct PreparedPublication {
119    opening: PublishContentClientFrame,
120    originals: PublicationOriginals,
121    plan: PublicationAcceptancePlan,
122}
123impl PreparedPublication {
124    pub fn new(
125        opening: PublishContentClientFrame,
126        originals: PublicationOriginals,
127        spool_genesis: ContentHash,
128    ) -> Result<Self, Error> {
129        // Ordinary inputs only. Callers deliberately attach a new signature with
130        // sign_acceptance/accept; unreferenced incoming evidence is not trusted.
131        let plan = prepare_plan(&opening, &originals, spool_genesis)?;
132        Ok(Self {
133            opening,
134            originals,
135            plan,
136        })
137    }
138    pub fn opening(&self) -> &PublishContentClientFrame {
139        &self.opening
140    }
141    pub fn originals(&self) -> &PublicationOriginals {
142        &self.originals
143    }
144    pub fn plan(&self) -> &PublicationAcceptancePlan {
145        &self.plan
146    }
147    /// Build an unsigned exact intent for an external signer. The caller chooses
148    /// the accepting authority explicitly; this is no claim of current validity.
149    pub fn acceptance(
150        &self,
151        author: SourceAuthor,
152        publisher: [u8; 32],
153        kinds: BTreeSet<BoundaryOriginalKind>,
154    ) -> Result<OriginalBoundaryAcceptance, Error> {
155        let SourceAuthor::Account { actor, .. } = &author else {
156            return Err(Error::Protocol(
157                "explicit account accepting authority required",
158            ));
159        };
160        let value = OriginalBoundaryAcceptance {
161            version: 1,
162            publication_intent: self.plan.intent.id().map_err(invalid)?,
163            originals_manifest: self.plan.manifest.id().map_err(invalid)?,
164            original_account: actor.principal_id,
165            kinds,
166            accepting_publisher: publisher,
167            accepting_author: author,
168        };
169        value
170            .selected(&self.plan.intent, &self.plan.manifest)
171            .map_err(invalid)?;
172        Ok(value)
173    }
174    /// Explicitly sign a separate acceptance. Original signatures and cached
175    /// author envelopes remain byte-for-byte unchanged, including revoked work.
176    pub fn sign_acceptance(
177        &mut self,
178        author: SourceAuthor,
179        kinds: BTreeSet<BoundaryOriginalKind>,
180        signer: &impl Signer,
181    ) -> Result<ContentHash, Error> {
182        let publisher = signer
183            .public_key()
184            .try_into()
185            .map_err(|_| Error::Protocol("invalid accepting key"))?;
186        let value = self.acceptance(author, publisher, kinds)?;
187        self.accept(SignedBoundaryAcceptance::sign(&value, signer).map_err(invalid)?)
188    }
189    /// Attach a signature produced by an external signer. Reject a changed
190    /// intent, extra originals, ambiguous selection, or exceeding carrier bounds.
191    pub fn accept(&mut self, signed: SignedBoundaryAcceptance) -> Result<ContentHash, Error> {
192        let wire = crate::boundary_acceptance::encode(&signed)?;
193        let wire_len = wire.encoded_len() + 10; // field key + maximum length varint
194        let total: usize = self
195            .originals
196            .geneses
197            .iter()
198            .map(Message::encoded_len)
199            .chain(self.originals.operations.iter().map(Message::encoded_len))
200            .sum();
201        let existing: BTreeSet<_> = self
202            .originals
203            .geneses
204            .iter()
205            .flat_map(|g| &g.boundary_acceptances)
206            .chain(
207                self.originals
208                    .operations
209                    .iter()
210                    .flat_map(|b| &b.boundary_acceptances),
211            )
212            .map(|record| record.canonical_record.as_slice())
213            .collect();
214        if existing.contains(wire.canonical_record.as_slice()) {
215            return Err(Error::Protocol(
216                "duplicate fresh acceptance or retained evidence",
217            ));
218        }
219        if total.saturating_add(wire_len) > 16 * 1024 * 1024 || existing.len() >= 128 {
220            return Err(Error::Protocol(
221                "publication acceptance metadata budget exceeded",
222            ));
223        }
224        // Choose an existing original carrier. Never add an empty operation
225        // batch or copy the 16 MiB original collection just to attach a proof.
226        let wrapper = self
227            .originals
228            .geneses
229            .iter()
230            .position(|g| g.encoded_len().saturating_add(wire_len) <= 256 * 1024);
231        let batch = self
232            .originals
233            .operations
234            .iter()
235            .position(|b| b.encoded_len().saturating_add(wire_len) <= 256 * 1024);
236        if wrapper.is_none() && batch.is_none() {
237            return Err(Error::Protocol("no bounded acceptance carrier available"));
238        }
239        let id = self.plan.insert(signed)?;
240        if let Some(index) = wrapper {
241            self.originals.geneses[index]
242                .boundary_acceptances
243                .push(wire);
244        } else if let Some(index) = batch {
245            self.originals.operations[index]
246                .boundary_acceptances
247                .push(wire);
248        }
249        Ok(id)
250    }
251    pub fn into_parts(
252        self,
253    ) -> (
254        PublishContentClientFrame,
255        PublicationOriginals,
256        PublicationAcceptancePlan,
257    ) {
258        (self.opening, self.originals, self.plan)
259    }
260}
261
262/// Publication-only candidate extraction. Removes only unreferenced acceptance
263/// carriers from the owned input, leaving generic receipt matching strict.
264/// Full signature, manifest, and intent checks finish before a proposal is exposed.
265pub fn proposed_publication(
266    opening: &PublishContentClientFrame,
267    mut originals: PublicationOriginals,
268    spool_genesis: ContentHash,
269) -> Result<(PublicationOriginals, PublicationAcceptancePlan), Error> {
270    originals.validate_bounds().map_err(invalid)?;
271    let mut candidates = BTreeMap::new();
272    let mut retained = BTreeSet::new();
273    for wrapper in &mut originals.geneses {
274        let mut references = BTreeSet::new();
275        if let Some(receipt) = &wrapper.admission {
276            let value = heddle_object_model::object::thread_genesis_admission::ThreadGenesisAdmission::decode(&receipt.canonical_record).map_err(invalid)?;
277            reference(&value.basis, &mut references);
278        }
279        for receipt in &wrapper.ownership_claim_admissions {
280            reference(
281                &crate::authority_admission::verify_signature(receipt)?.basis,
282                &mut references,
283            );
284        }
285        for receipt in &wrapper.ownership_resolution_admissions {
286            reference(
287                &crate::authority_admission::verify_signature(receipt)?.basis,
288                &mut references,
289            );
290        }
291        separate(
292            &mut wrapper.boundary_acceptances,
293            &references,
294            &mut retained,
295            &mut candidates,
296        )?;
297    }
298    for batch in &mut originals.operations {
299        let mut references = BTreeSet::new();
300        for receipt in &batch.authority_admissions {
301            reference(
302                &crate::authority_admission::verify_signature(receipt)?.basis,
303                &mut references,
304            );
305        }
306        separate(
307            &mut batch.boundary_acceptances,
308            &references,
309            &mut retained,
310            &mut candidates,
311        )?;
312    }
313    if candidates.keys().any(|id| retained.contains(id)) {
314        return Err(Error::Protocol(
315            "acceptance cannot be both fresh and retained evidence",
316        ));
317    }
318    let mut plan = prepare_plan(opening, &originals, spool_genesis)?;
319    for (_, signed) in candidates {
320        plan.insert(signed)?;
321    }
322    Ok((originals, plan))
323}
324fn reference(basis: &AdmissionBasis, ids: &mut BTreeSet<ContentHash>) {
325    if let AdmissionBasis::BoundaryAcceptance { acceptance } = basis {
326        ids.insert(*acceptance);
327    }
328}
329fn separate(
330    records: &mut Vec<SignedRecord>,
331    references: &BTreeSet<ContentHash>,
332    retained: &mut BTreeSet<ContentHash>,
333    candidates: &mut BTreeMap<ContentHash, SignedBoundaryAcceptance>,
334) -> Result<(), Error> {
335    let mut local = BTreeSet::new();
336    let mut keep = Vec::new();
337    for wire in std::mem::take(records) {
338        let signed = crate::boundary_acceptance::decode(&wire)?;
339        let id = ContentHash::compute_typed(FORMAT, &signed.canonical);
340        if !local.insert(id) {
341            return Err(Error::Protocol(
342                "duplicate acceptance in publication carrier",
343            ));
344        }
345        if references.contains(&id) {
346            retained.insert(id);
347            keep.push(wire);
348        } else if candidates.insert(id, signed).is_some() {
349            return Err(Error::Protocol("duplicate fresh acceptance"));
350        }
351    }
352    *records = keep;
353    Ok(())
354}
355fn prepare_plan(
356    opening: &PublishContentClientFrame,
357    originals: &PublicationOriginals,
358    spool_genesis: ContentHash,
359) -> Result<PublicationAcceptancePlan, Error> {
360    originals.validate_bounds().map_err(invalid)?;
361    let intent = publication_intent(opening, spool_genesis)?;
362    let mut entries = Vec::new();
363    let mut geneses = BTreeSet::new();
364    for wrapper in &originals.geneses {
365        let wire = wrapper
366            .genesis
367            .as_ref()
368            .ok_or(Error::Protocol("missing original genesis"))?;
369        let decoded = ThreadGenesis::decode(&wire.canonical_record).map_err(invalid)?;
370        let id = decoded.id().map_err(invalid)?;
371        let reference = ThreadRef {
372            spool: Some(SpoolRef {
373                id: intent.spool.to_string(),
374            }),
375            id: Some(ThreadId {
376                value: id.as_bytes().to_vec(),
377            }),
378        };
379        let genesis = crate::fetch::verify_origin(wrapper, &reference).map_err(invalid)?;
380        if !geneses.insert(id) {
381            return Err(Error::Protocol("duplicate original genesis"));
382        }
383        entries.push(
384            OriginalManifestEntry::from_genesis(&genesis, &wrapper.creator_authority)
385                .map_err(invalid)?,
386        );
387        for claim in crate::replication::ownership::verify_claims(wrapper, &genesis)? {
388            entries.push(
389                OriginalManifestEntry::from_claim(&claim.original.verify().map_err(invalid)?)
390                    .map_err(invalid)?,
391            );
392        }
393        for resolution in crate::replication::ownership::verify_resolutions(wrapper, &genesis)? {
394            let value = heddle_object_model::object::thread_replication::ownership_resolution::ThreadOwnershipResolution::decode(&resolution.original.canonical)
395                .map_err(invalid)?;
396            entries.push(OriginalManifestEntry::from_resolution(&value).map_err(invalid)?);
397        }
398    }
399    if !geneses.contains(&intent.thread) {
400        return Err(Error::Protocol("selected original genesis missing"));
401    }
402    for batch in &originals.operations {
403        for received in crate::authority_admission::match_batch(batch)? {
404            let operation = received.original.verify().map_err(invalid)?;
405            if !geneses.contains(&operation.thread) {
406                return Err(Error::Protocol("operation original genesis missing"));
407            }
408            let entry = OriginalManifestEntry::from_operation(&operation).map_err(invalid)?;
409            if entry
410                .authority
411                .as_ref()
412                .is_some_and(|authority| authority.spool != intent.spool)
413            {
414                return Err(Error::Protocol(
415                    "original authority differs from publication Spool",
416                ));
417            }
418            entries.push(entry);
419        }
420    }
421    let manifest = OriginalPublicationManifest::new(entries).map_err(invalid)?;
422    Ok(PublicationAcceptancePlan {
423        intent,
424        manifest: Arc::new(manifest),
425        proposed: BTreeMap::new(),
426        subjects: BTreeMap::new(),
427    })
428}
429/// Shared exact immutable intent. Checkpoint is a transport cursor, excluded so
430/// a resume does not require re-signing; the operation ID remains committed.
431pub fn publication_intent(
432    opening: &PublishContentClientFrame,
433    spool_genesis: ContentHash,
434) -> Result<PublicationIntent, Error> {
435    let Some(publish_content_client_frame::Body::Open(open)) = &opening.body else {
436        return Err(Error::Protocol("publication Open required"));
437    };
438    let thread = open
439        .thread
440        .as_ref()
441        .ok_or(Error::Protocol("publication Thread required"))?;
442    let spool = thread
443        .spool
444        .as_ref()
445        .ok_or(Error::Protocol("publication Spool required"))?;
446    let revision = open
447        .revision
448        .as_ref()
449        .ok_or(Error::Protocol("publication revision required"))?;
450    if revision.spool.as_ref() != Some(spool) {
451        return Err(Error::Protocol("publication revision Spool differs"));
452    }
453    let Some(revision_ref::Revision::State(state)) = &revision.revision else {
454        return Err(Error::Protocol("exact publication State required"));
455    };
456    let hash = |bytes: &[u8]| -> Result<ContentHash, Error> {
457        let value: [u8; 32] = bytes
458            .try_into()
459            .map_err(|_| Error::Protocol("32-byte publication identity required"))?;
460        if value == [0; 32] {
461            return Err(Error::Protocol("zero publication identity"));
462        }
463        Ok(ContentHash::from_bytes(value))
464    };
465    let endpoint = |endpoint: Option<&EndpointRef>| -> Result<[u8; 32], Error> {
466        endpoint
467            .ok_or(Error::Protocol("publication endpoint required"))?
468            .public_key
469            .as_slice()
470            .try_into()
471            .map_err(|_| Error::Protocol("32-byte endpoint required"))
472    };
473    if spool_genesis.as_bytes() == &[0; 32] {
474        return Err(Error::Protocol("verified Spool genesis required"));
475    }
476    let value = PublicationIntent {
477        spool: Uuid::parse_str(&spool.id).map_err(invalid)?,
478        spool_genesis,
479        thread: hash(
480            &thread
481                .id
482                .as_ref()
483                .ok_or(Error::Protocol("Thread ID required"))?
484                .value,
485        )?,
486        revision: StateId::from_bytes(*hash(&state.value)?.as_bytes()),
487        inventory: ContentHash::from_bytes(super::inventory_digest(&open.packs).map_err(invalid)?),
488        sharing_policy: if open.sharing_policy_version.is_empty() {
489            None
490        } else {
491            Some(hash(&open.sharing_policy_version)?)
492        },
493        source: endpoint(open.source.as_ref())?,
494        destination: endpoint(open.destination.as_ref())?,
495        client_operation_id: Uuid::parse_str(&opening.client_operation_id).map_err(invalid)?,
496    };
497    value.id().map_err(invalid)?;
498    Ok(value)
499}
500fn invalid(error: impl std::fmt::Display) -> Error {
501    Error::Io(error.to_string())
502}
503
504#[cfg(test)]
505#[path = "acceptance_tests.rs"]
506mod tests;
heddle_thread_api/publication/acceptance.rs

heddle_thread_api/publication/
acceptance.rs
