structured_cms/
payload.rs

1//! One complete source pass for resolved digest and raw-signature states.
2
3use crate::Error;
4use crate::Octets;
5use crate::digest::{DigestBatch, DigestError, DigestMemo, DigestOutput};
6use crate::provider::{SignatureAnswer, SignatureProvider};
7use crate::stream::{self, Sink, Source, StreamError};
8use structured_asn1::memory::{OwnedVec, Work};
9use structured_pkix::path::WorkingKey;
10
11/// A raw challenge already resolved by this operation's selected provider.
12///
13/// The original signature catalogue must remain stable for the duration of
14/// the pass. The effective key is a value, not a borrow of a temporary key
15/// descriptor. A raw challenge is never satisfied by feeding digest bytes.
16pub struct RawRequest<'a, P: SignatureProvider<'a>> {
17    /// Fully resolved signature identity from the selected provider.
18    pub identity: P::SignatureIdentity,
19    /// Original effective key, including inherited parameters.
20    pub key: WorkingKey<'a>,
21    /// Original logical signature octets, without concatenation or re-encoding.
22    pub signature: &'a Octets<'a>,
23}
24
25/// Why a prepared payload operation could not execute.
26#[derive(Debug)]
27pub enum PayloadError<E> {
28    /// Common storage or fan-out admission failed before execution.
29    Processing(Error),
30    /// Original raw-verification callback failure.
31    Callback(E),
32    /// Digest processing or callback failure, with its distinct cause.
33    Digest(DigestError<E>),
34}
35
36/// Source acquisition, processing and selected-provider causes for one pass.
37pub type PassError<SE, PE> = StreamError<SE, PayloadError<PE>>;
38
39/// A prepared fan-out bound to the digest batch's selected provider instance.
40///
41/// This is a processing mechanism, not CMS schema, signer selection or trust
42/// validation. The caller must resolve and validate every request first.
43pub struct PayloadBatch<'a, P: SignatureProvider<'a>> {
44    digests: DigestBatch<'a, P>,
45    raw: OwnedVec<P::RawState>,
46}
47
48/// Outputs available only after complete source EOF and successful callbacks.
49pub struct Completed<'a, P: SignatureProvider<'a>> {
50    digests: DigestMemo<'a, P>,
51    raw: OwnedVec<SignatureAnswer>,
52}
53
54impl<'a, P: SignatureProvider<'a>> Completed<'a, P> {
55    /// Complete typed digest outputs with their original request mapping.
56    #[must_use]
57    pub fn digests(&self) -> &DigestMemo<'a, P> {
58        &self.digests
59    }
60
61    /// Raw answers in request order. These do not assert certificate trust.
62    #[must_use]
63    pub fn raw(&self) -> &[SignatureAnswer] {
64        &self.raw
65    }
66}
67
68impl<'a, P: SignatureProvider<'a>> PayloadBatch<'a, P> {
69    /// Initializes raw challenges before source I/O, without retaining Work.
70    ///
71    /// No identity/state cloning, provider switching or prehash substitution
72    /// occurs. Every request's metadata storage is admitted before its start
73    /// callback. The callback accounts for its own complete transitive work.
74    pub fn prepare<W: Work>(
75        digests: DigestBatch<'a, P>,
76        requests: OwnedVec<RawRequest<'a, P>>,
77        execution: &W,
78    ) -> Result<Self, PayloadError<P::Error>> {
79        let (field, at) = digests.location();
80        let refusal = |reason| PayloadError::Processing(Error::resource(reason, field, at));
81        let mut raw = execution.vector();
82        raw.try_reserve_exact(requests.len()).map_err(refusal)?;
83        execution
84            .charge_work(core::mem::size_of_val(&*requests) as u64)
85            .map_err(refusal)?;
86        for request in requests {
87            execution.charge_work(1).map_err(refusal)?;
88            let state = digests
89                .provider()
90                .start_raw(&request.identity, request.key, request.signature, execution)
91                .map_err(PayloadError::Callback)?;
92            raw.try_push(state).map_err(refusal)?;
93        }
94        Ok(Self {
95            digests,
96            raw: raw.into_owned(),
97        })
98    }
99
100    /// Streams once to every digest and raw state, then consumes each once.
101    ///
102    /// Output catalogues are admitted before calling the source. Incomplete
103    /// source acquisition or any sticky callback refusal returns no completed
104    /// digests or raw answers, even if other states processed an input prefix.
105    pub fn run<S: Source, W: Work>(
106        mut self,
107        source: &mut S,
108        execution: &W,
109    ) -> Result<Completed<'a, P>, PassError<S::Error, P::Error>> {
110        let (field, at) = self.digests.location();
111        let refusal = |reason| StreamError::Processing(Error::resource(reason, field, at));
112        let mut outputs: OwnedVec<DigestOutput<'a, P>, W> = execution.vector();
113        outputs
114            .try_reserve_exact(self.digests.len())
115            .map_err(refusal)?;
116        let mut answers = execution.vector();
117        answers.try_reserve_exact(self.raw.len()).map_err(refusal)?;
118        stream::run(source, &mut PayloadSink(&mut self), execution, field, at)?;
119        execution
120            .charge_work(core::mem::size_of_val(&*self.raw) as u64)
121            .map_err(refusal)?;
122        let provider = self.digests.provider();
123        let digests = self
124            .digests
125            .finish(outputs, execution)
126            .map_err(|error| StreamError::Callback(PayloadError::Digest(error)))?;
127        for state in self.raw {
128            execution.charge_work(1).map_err(refusal)?;
129            let answer = provider
130                .finish_raw(state, execution)
131                .map_err(|error| StreamError::Callback(PayloadError::Callback(error)))?;
132            answers.try_push(answer).map_err(refusal)?;
133        }
134        Ok(Completed {
135            digests,
136            raw: answers.into_owned(),
137        })
138    }
139}
140
141struct PayloadSink<'s, 'a, P: SignatureProvider<'a>>(&'s mut PayloadBatch<'a, P>);
142
143impl<'a, P: SignatureProvider<'a>> Sink for PayloadSink<'_, 'a, P> {
144    type Error = PayloadError<P::Error>;
145
146    fn write<W: Work>(&mut self, bytes: &[u8], execution: &W) -> Result<(), Self::Error> {
147        self.0
148            .digests
149            .feed(bytes, execution)
150            .map_err(PayloadError::Digest)?;
151        let (field, at) = self.0.digests.location();
152        for state in &mut self.0.raw {
153            execution
154                .charge_work(1)
155                .map_err(|reason| PayloadError::Processing(Error::resource(reason, field, at)))?;
156            self.0
157                .digests
158                .provider()
159                .update_raw(state, bytes, execution)
160                .map_err(PayloadError::Callback)?;
161        }
162        Ok(())
163    }
164}
structured_cms/payload.rs

structured_cms/
payload.rs
