commonware_consensus/aggregation/
types.rs

1//! Types used in [aggregation](super).
2
3use crate::{
4    aggregation::scheme,
5    types::{Epoch, Height},
6    Heightable,
7};
8use bytes::{Buf, BufMut, Bytes};
9use commonware_codec::{Encode, EncodeSize, Error as CodecError, Read, ReadExt, Write};
10use commonware_cryptography::{
11    certificate::{Attestation, Namespace as CertificateNamespace, Scheme, Subject},
12    Digest,
13};
14use commonware_parallel::Strategy;
15use commonware_utils::{channel::oneshot, union, N3f1};
16use rand_core::CryptoRng;
17use std::hash::Hash;
18
19/// Error that may be encountered when interacting with `aggregation`.
20#[derive(Debug, thiserror::Error)]
21pub enum Error {
22    // Proposal Errors
23    /// The proposal was canceled by the application
24    #[error("Application verify error: {0}")]
25    AppProposeCanceled(oneshot::error::RecvError),
26
27    // Epoch Errors
28    /// The specified validator is not a participant in the epoch
29    #[error("Epoch {0} has no validator {1}")]
30    UnknownValidator(Epoch, String),
31    /// The local node is not a signer in the scheme for the specified epoch.
32    #[error("Not a signer at epoch {0}")]
33    NotSigner(Epoch),
34
35    // Peer Errors
36    /// The sender's public key doesn't match the expected key
37    #[error("Peer mismatch")]
38    PeerMismatch,
39
40    // Signature Errors
41    /// The acknowledgment signature is invalid
42    #[error("Invalid ack signature")]
43    InvalidAckSignature,
44
45    // Ignorable Message Errors
46    /// The acknowledgment's epoch is outside the accepted bounds
47    #[error("Invalid ack epoch {0} outside bounds {1} - {2}")]
48    AckEpochOutsideBounds(Epoch, Epoch, Epoch),
49    /// The acknowledgment's height is outside the accepted bounds
50    #[error("Non-useful ack height {0}")]
51    AckHeight(Height),
52    /// The acknowledgment's digest is incorrect
53    #[error("Invalid ack digest {0}")]
54    AckDigest(Height),
55    /// Duplicate acknowledgment for the same height
56    #[error("Duplicate ack from sender {0} for height {1}")]
57    AckDuplicate(String, Height),
58    /// The acknowledgement is for a height that already has a certificate
59    #[error("Ack for height {0} already has been certified")]
60    AckCertified(Height),
61    /// The epoch is unknown
62    #[error("Unknown epoch {0}")]
63    UnknownEpoch(Epoch),
64}
65
66impl Error {
67    /// Returns true if the error represents a blockable offense by a peer.
68    pub const fn blockable(&self) -> bool {
69        matches!(self, Self::PeerMismatch | Self::InvalidAckSignature)
70    }
71}
72
73/// Suffix used to identify an acknowledgment (ack) namespace for domain separation.
74/// Used when signing and verifying acks to prevent signature reuse across different message types.
75const ACK_SUFFIX: &[u8] = b"_AGG_ACK";
76
77/// Returns a suffixed namespace for signing an ack.
78///
79/// This provides domain separation for signatures, preventing cross-protocol attacks
80/// by ensuring signatures for acks cannot be reused for other message types.
81#[inline]
82fn ack_namespace(namespace: &[u8]) -> Vec<u8> {
83    union(namespace, ACK_SUFFIX)
84}
85
86/// Namespace type for aggregation acknowledgments.
87///
88/// This type encapsulates the pre-computed namespace bytes used for signing and
89/// verifying acks.
90#[derive(Clone, Debug)]
91pub struct Namespace(Vec<u8>);
92
93impl CertificateNamespace for Namespace {
94    fn derive(namespace: &[u8]) -> Self {
95        Self(ack_namespace(namespace))
96    }
97}
98
99/// Item represents a single element being aggregated in the protocol.
100/// Each item has a unique height and contains a digest that validators sign.
101#[derive(Clone, Debug, PartialEq, Eq, Hash)]
102pub struct Item<D: Digest> {
103    /// Sequential position of this item within the current epoch
104    pub height: Height,
105    /// Cryptographic digest of the data being aggregated
106    pub digest: D,
107}
108
109impl<D: Digest> Heightable for Item<D> {
110    fn height(&self) -> Height {
111        self.height
112    }
113}
114
115impl<D: Digest> Write for Item<D> {
116    fn write(&self, writer: &mut impl BufMut) {
117        self.height.write(writer);
118        self.digest.write(writer);
119    }
120}
121
122impl<D: Digest> Read for Item<D> {
123    type Cfg = ();
124
125    fn read_cfg(reader: &mut impl Buf, _: &()) -> Result<Self, CodecError> {
126        let height = Height::read(reader)?;
127        let digest = D::read(reader)?;
128        Ok(Self { height, digest })
129    }
130}
131
132impl<D: Digest> EncodeSize for Item<D> {
133    fn encode_size(&self) -> usize {
134        self.height.encode_size() + self.digest.encode_size()
135    }
136}
137
138/// The signed message covers only the height and digest, intentionally excluding the epoch.
139/// See the [module docs](super#epoch-independent-signatures).
140impl<D: Digest> Subject for &Item<D> {
141    type Namespace = Namespace;
142
143    fn namespace<'a>(&self, derived: &'a Self::Namespace) -> &'a [u8] {
144        &derived.0
145    }
146
147    fn message(&self) -> Bytes {
148        self.encode()
149    }
150}
151
152#[cfg(feature = "arbitrary")]
153impl<D: Digest> arbitrary::Arbitrary<'_> for Item<D>
154where
155    D: for<'a> arbitrary::Arbitrary<'a>,
156{
157    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
158        let height = u.arbitrary::<Height>()?;
159        let digest = u.arbitrary::<D>()?;
160        Ok(Self { height, digest })
161    }
162}
163
164/// Acknowledgment (ack) represents a validator's vote on an item.
165/// Multiple acks can be recovered into a certificate for consensus.
166#[derive(Clone, Debug, PartialEq, Eq, Hash)]
167pub struct Ack<S: Scheme, D: Digest> {
168    /// The item being acknowledged
169    pub item: Item<D>,
170    /// The epoch in which this acknowledgment was created.
171    ///
172    /// Not part of the signed message: it selects the scheme used to verify the attestation and
173    /// assemble a certificate. See the [module docs](super#epoch-independent-signatures).
174    pub epoch: Epoch,
175    /// Scheme-specific attestation material
176    pub attestation: Attestation<S>,
177}
178
179impl<S: Scheme, D: Digest> Ack<S, D> {
180    /// Verifies the attestation on this acknowledgment.
181    ///
182    /// Returns `true` if the attestation is valid for the given namespace and public key.
183    /// Domain separation is automatically applied to prevent signature reuse.
184    pub fn verify<R>(&self, rng: &mut R, scheme: &S, strategy: &impl Strategy) -> bool
185    where
186        R: CryptoRng,
187        S: scheme::Scheme<D>,
188    {
189        scheme.verify_attestation::<_, D>(rng, &self.item, &self.attestation, strategy)
190    }
191
192    /// Creates a new acknowledgment by signing an item with a validator's key.
193    ///
194    /// The signature uses domain separation to prevent cross-protocol attacks. The epoch is
195    /// carried in the ack but is not part of the signed message. See the
196    /// [module docs](super#epoch-independent-signatures).
197    ///
198    /// # Determinism
199    ///
200    /// Signatures produced by this function are deterministic and safe for consensus.
201    pub fn sign(scheme: &S, epoch: Epoch, item: Item<D>) -> Option<Self>
202    where
203        S: scheme::Scheme<D>,
204    {
205        let attestation = scheme.sign::<D>(&item)?;
206        Some(Self {
207            item,
208            epoch,
209            attestation,
210        })
211    }
212}
213
214impl<S: Scheme, D: Digest> Write for Ack<S, D> {
215    fn write(&self, writer: &mut impl BufMut) {
216        self.item.write(writer);
217        self.epoch.write(writer);
218        self.attestation.write(writer);
219    }
220}
221
222impl<S: Scheme, D: Digest> Read for Ack<S, D> {
223    type Cfg = ();
224
225    fn read_cfg(reader: &mut impl Buf, _: &()) -> Result<Self, CodecError> {
226        let item = Item::read(reader)?;
227        let epoch = Epoch::read(reader)?;
228        let attestation = Attestation::read(reader)?;
229        Ok(Self {
230            item,
231            epoch,
232            attestation,
233        })
234    }
235}
236
237impl<S: Scheme, D: Digest> EncodeSize for Ack<S, D> {
238    fn encode_size(&self) -> usize {
239        self.item.encode_size() + self.epoch.encode_size() + self.attestation.encode_size()
240    }
241}
242
243#[cfg(feature = "arbitrary")]
244impl<S: Scheme, D: Digest> arbitrary::Arbitrary<'_> for Ack<S, D>
245where
246    S::Signature: for<'a> arbitrary::Arbitrary<'a>,
247    D: for<'a> arbitrary::Arbitrary<'a>,
248{
249    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
250        let item = u.arbitrary::<Item<D>>()?;
251        let epoch = u.arbitrary::<Epoch>()?;
252        let attestation = Attestation::arbitrary(u)?;
253        Ok(Self {
254            item,
255            epoch,
256            attestation,
257        })
258    }
259}
260
261/// Message exchanged between peers containing an acknowledgment and tip information.
262/// This combines a validator's vote with their view of consensus progress.
263#[derive(Clone, Debug, PartialEq, Eq, Hash)]
264pub struct TipAck<S: Scheme, D: Digest> {
265    /// The peer's local view of the tip (the lowest height that is not yet confirmed).
266    pub tip: Height,
267
268    /// The peer's acknowledgement (vote) for an item.
269    pub ack: Ack<S, D>,
270}
271
272impl<S: Scheme, D: Digest> Write for TipAck<S, D> {
273    fn write(&self, writer: &mut impl BufMut) {
274        self.tip.write(writer);
275        self.ack.write(writer);
276    }
277}
278
279impl<S: Scheme, D: Digest> Read for TipAck<S, D> {
280    type Cfg = ();
281
282    fn read_cfg(reader: &mut impl Buf, _: &()) -> Result<Self, CodecError> {
283        let tip = Height::read(reader)?;
284        let ack = Ack::read(reader)?;
285        Ok(Self { tip, ack })
286    }
287}
288
289impl<S: Scheme, D: Digest> EncodeSize for TipAck<S, D> {
290    fn encode_size(&self) -> usize {
291        self.tip.encode_size() + self.ack.encode_size()
292    }
293}
294
295#[cfg(feature = "arbitrary")]
296impl<S: Scheme, D: Digest> arbitrary::Arbitrary<'_> for TipAck<S, D>
297where
298    D: for<'a> arbitrary::Arbitrary<'a>,
299    Ack<S, D>: for<'a> arbitrary::Arbitrary<'a>,
300{
301    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
302        let tip = u.arbitrary::<Height>()?;
303        let ack = u.arbitrary::<Ack<S, D>>()?;
304        Ok(Self { tip, ack })
305    }
306}
307
308/// A recovered certificate for some [Item].
309#[derive(Clone, Debug, PartialEq, Eq, Hash)]
310pub struct Certificate<S: Scheme, D: Digest> {
311    /// The item that was recovered.
312    pub item: Item<D>,
313    /// The recovered certificate.
314    pub certificate: S::Certificate,
315}
316
317impl<S: Scheme, D: Digest> Certificate<S, D> {
318    pub fn from_acks<'a, I>(scheme: &S, acks: I, strategy: &impl Strategy) -> Option<Self>
319    where
320        S: scheme::Scheme<D>,
321        I: IntoIterator<Item = &'a Ack<S, D>>,
322        I::IntoIter: Send,
323    {
324        let mut iter = acks.into_iter().peekable();
325        let item = iter.peek()?.item.clone();
326        let attestations = iter
327            .filter(|ack| ack.item == item)
328            .map(|ack| ack.attestation.clone());
329        let certificate = scheme.assemble::<_, N3f1>(attestations, strategy)?;
330
331        Some(Self { item, certificate })
332    }
333
334    /// Verifies the recovered certificate for the item.
335    pub fn verify<R>(&self, rng: &mut R, scheme: &S, strategy: &impl Strategy) -> bool
336    where
337        R: CryptoRng,
338        S: scheme::Scheme<D>,
339    {
340        scheme.verify_certificate::<_, D, N3f1>(rng, &self.item, &self.certificate, strategy)
341    }
342}
343
344impl<S: Scheme, D: Digest> Write for Certificate<S, D> {
345    fn write(&self, writer: &mut impl BufMut) {
346        self.item.write(writer);
347        self.certificate.write(writer);
348    }
349}
350
351impl<S: Scheme, D: Digest> Read for Certificate<S, D> {
352    type Cfg = <S::Certificate as Read>::Cfg;
353
354    fn read_cfg(reader: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, CodecError> {
355        let item = Item::read(reader)?;
356        let certificate = S::Certificate::read_cfg(reader, cfg)?;
357        Ok(Self { item, certificate })
358    }
359}
360
361impl<S: Scheme, D: Digest> EncodeSize for Certificate<S, D> {
362    fn encode_size(&self) -> usize {
363        self.item.encode_size() + self.certificate.encode_size()
364    }
365}
366
367#[cfg(feature = "arbitrary")]
368impl<S: Scheme, D: Digest> arbitrary::Arbitrary<'_> for Certificate<S, D>
369where
370    D: for<'a> arbitrary::Arbitrary<'a>,
371    S::Certificate: for<'a> arbitrary::Arbitrary<'a>,
372{
373    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
374        let item = u.arbitrary::<Item<D>>()?;
375        let certificate = u.arbitrary::<S::Certificate>()?;
376        Ok(Self { item, certificate })
377    }
378}
379
380/// Used as [Reporter::Activity](crate::Reporter::Activity) to report activities that occur during
381/// aggregation. Also used to journal events that are needed to initialize the aggregation engine
382/// when the node restarts.
383#[derive(Clone, Debug, PartialEq)]
384pub enum Activity<S: Scheme, D: Digest> {
385    /// Received an ack from a participant.
386    Ack(Ack<S, D>),
387
388    /// Certified an [Item].
389    Certified(Certificate<S, D>),
390
391    /// Moved the tip to a new height.
392    Tip(Height),
393}
394
395impl<S: Scheme, D: Digest> Write for Activity<S, D> {
396    fn write(&self, writer: &mut impl BufMut) {
397        match self {
398            Self::Ack(ack) => {
399                0u8.write(writer);
400                ack.write(writer);
401            }
402            Self::Certified(certificate) => {
403                1u8.write(writer);
404                certificate.write(writer);
405            }
406            Self::Tip(height) => {
407                2u8.write(writer);
408                height.write(writer);
409            }
410        }
411    }
412}
413
414impl<S: Scheme, D: Digest> Read for Activity<S, D> {
415    type Cfg = <S::Certificate as Read>::Cfg;
416
417    fn read_cfg(reader: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, CodecError> {
418        match u8::read(reader)? {
419            0 => Ok(Self::Ack(Ack::read(reader)?)),
420            1 => Ok(Self::Certified(Certificate::read_cfg(reader, cfg)?)),
421            2 => Ok(Self::Tip(Height::read(reader)?)),
422            _ => Err(CodecError::Invalid(
423                "consensus::aggregation::Activity",
424                "Invalid type",
425            )),
426        }
427    }
428}
429
430impl<S: Scheme, D: Digest> EncodeSize for Activity<S, D> {
431    fn encode_size(&self) -> usize {
432        1 + match self {
433            Self::Ack(ack) => ack.encode_size(),
434            Self::Certified(certificate) => certificate.encode_size(),
435            Self::Tip(height) => height.encode_size(),
436        }
437    }
438}
439
440#[cfg(feature = "arbitrary")]
441impl<S: Scheme, D: Digest> arbitrary::Arbitrary<'_> for Activity<S, D>
442where
443    D: for<'a> arbitrary::Arbitrary<'a>,
444    Ack<S, D>: for<'a> arbitrary::Arbitrary<'a>,
445    Certificate<S, D>: for<'a> arbitrary::Arbitrary<'a>,
446{
447    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
448        let choice = u.int_in_range(0..=2)?;
449        match choice {
450            0 => Ok(Self::Ack(u.arbitrary::<Ack<S, D>>()?)),
451            1 => Ok(Self::Certified(u.arbitrary::<Certificate<S, D>>()?)),
452            2 => Ok(Self::Tip(u.arbitrary::<Height>()?)),
453            _ => unreachable!(),
454        }
455    }
456}
457
458#[cfg(test)]
459mod tests {
460    use super::*;
461    use crate::aggregation::scheme::{
462        bls12381_multisig, bls12381_threshold, ed25519, secp256r1, Scheme,
463    };
464    use bytes::BytesMut;
465    use commonware_codec::{Decode, DecodeExt, Encode};
466    use commonware_cryptography::{
467        bls12381::primitives::variant::{MinPk, MinSig},
468        certificate::mocks::Fixture,
469        Hasher, Sha256,
470    };
471    use commonware_parallel::Sequential;
472    use commonware_utils::{ordered::Quorum, test_rng, N3f1, TestRng};
473
474    const NAMESPACE: &[u8] = b"test";
475
476    type Sha256Digest = <Sha256 as Hasher>::Digest;
477
478    #[test]
479    fn test_ack_namespace() {
480        let namespace = b"test_namespace";
481        let expected = [namespace, ACK_SUFFIX].concat();
482        assert_eq!(ack_namespace(namespace), expected);
483    }
484
485    fn codec<S, F>(fixture: F)
486    where
487        S: Scheme<Sha256Digest>,
488        F: FnOnce(&mut TestRng, &[u8], u32) -> Fixture<S>,
489    {
490        let mut rng = test_rng();
491        let fixture = fixture(&mut rng, NAMESPACE, 4);
492        let schemes = &fixture.schemes;
493        let item = Item {
494            height: Height::new(100),
495            digest: Sha256::hash(b"test_item"),
496        };
497
498        // Test Item codec
499        let restored_item = Item::decode(item.encode()).unwrap();
500        assert_eq!(item, restored_item);
501
502        // Test Ack creation and codec
503        let ack = Ack::sign(&schemes[0], Epoch::new(1), item.clone()).unwrap();
504        let cfg = schemes[0].certificate_codec_config();
505        let encoded_ack = ack.encode();
506        let restored_ack: Ack<S, Sha256Digest> = Ack::decode(encoded_ack).unwrap();
507
508        // Verify the restored ack
509        assert_eq!(restored_ack.item, item);
510        assert_eq!(restored_ack.epoch, Epoch::new(1));
511        assert!(restored_ack.verify(&mut rng, &schemes[0], &Sequential));
512
513        // Test TipAck codec
514        let tip_ack = TipAck {
515            ack: ack.clone(),
516            tip: Height::new(42),
517        };
518        let encoded_tip_ack = tip_ack.encode();
519        let restored_tip_ack: TipAck<S, Sha256Digest> = TipAck::decode(encoded_tip_ack).unwrap();
520        assert_eq!(restored_tip_ack.tip, Height::new(42));
521        assert_eq!(restored_tip_ack.ack.item, item);
522        assert_eq!(restored_tip_ack.ack.epoch, Epoch::new(1));
523
524        // Test Activity codec - Ack variant
525        let activity_ack = Activity::Ack(ack);
526        let encoded_activity = activity_ack.encode();
527        let restored_activity_ack: Activity<S, Sha256Digest> =
528            Activity::decode_cfg(encoded_activity, &cfg).unwrap();
529        if let Activity::Ack(restored) = restored_activity_ack {
530            assert_eq!(restored.item, item);
531            assert_eq!(restored.epoch, Epoch::new(1));
532        } else {
533            panic!("Expected Activity::Ack");
534        }
535
536        // Test Activity codec - Certified variant
537        // Collect enough acks for a certificate
538        let acks: Vec<_> = schemes
539            .iter()
540            .take(schemes[0].participants().quorum::<N3f1>() as usize)
541            .filter_map(|scheme| Ack::sign(scheme, Epoch::new(1), item.clone()))
542            .collect();
543
544        let certificate = Certificate::from_acks(&schemes[0], &acks, &Sequential).unwrap();
545        assert!(certificate.verify(&mut rng, &schemes[0], &Sequential));
546
547        let activity_certified = Activity::Certified(certificate.clone());
548        let encoded_certified = activity_certified.encode();
549        let restored_activity_certified: Activity<S, Sha256Digest> =
550            Activity::decode_cfg(encoded_certified, &cfg).unwrap();
551        if let Activity::Certified(restored) = restored_activity_certified {
552            assert_eq!(restored.item, item);
553            assert!(restored.verify(&mut rng, &schemes[0], &Sequential));
554        } else {
555            panic!("Expected Activity::Certified");
556        }
557
558        // Test Activity codec - Tip variant
559        let activity_tip: Activity<S, Sha256Digest> = Activity::Tip(Height::new(123));
560        let encoded_tip = activity_tip.encode();
561        let restored_activity_tip: Activity<S, Sha256Digest> =
562            Activity::decode_cfg(encoded_tip, &cfg).unwrap();
563        if let Activity::Tip(height) = restored_activity_tip {
564            assert_eq!(height, Height::new(123));
565        } else {
566            panic!("Expected Activity::Tip");
567        }
568    }
569
570    #[test]
571    fn test_codec() {
572        codec(ed25519::fixture);
573        codec(secp256r1::fixture);
574        codec(bls12381_multisig::fixture::<MinPk, _>);
575        codec(bls12381_multisig::fixture::<MinSig, _>);
576        codec(bls12381_threshold::fixture::<MinPk, _>);
577        codec(bls12381_threshold::fixture::<MinSig, _>);
578    }
579
580    fn activity_invalid_enum<S, F>(fixture: F)
581    where
582        S: Scheme<Sha256Digest>,
583        F: FnOnce(&mut TestRng, &[u8], u32) -> Fixture<S>,
584    {
585        let fixture = fixture(&mut test_rng(), NAMESPACE, 4);
586        let mut buf = BytesMut::new();
587        3u8.write(&mut buf); // Invalid discriminant
588
589        let cfg = fixture.schemes[0].certificate_codec_config();
590        let result = Activity::<S, Sha256Digest>::read_cfg(&mut &buf[..], &cfg);
591        assert!(matches!(
592            result,
593            Err(CodecError::Invalid(
594                "consensus::aggregation::Activity",
595                "Invalid type"
596            ))
597        ));
598    }
599
600    #[test]
601    fn test_activity_invalid_enum() {
602        activity_invalid_enum(ed25519::fixture);
603        activity_invalid_enum(secp256r1::fixture);
604        activity_invalid_enum(bls12381_multisig::fixture::<MinPk, _>);
605        activity_invalid_enum(bls12381_multisig::fixture::<MinSig, _>);
606        activity_invalid_enum(bls12381_threshold::fixture::<MinPk, _>);
607        activity_invalid_enum(bls12381_threshold::fixture::<MinSig, _>);
608    }
609
610    #[cfg(feature = "arbitrary")]
611    mod conformance {
612        use super::*;
613        use crate::aggregation::scheme::bls12381_threshold;
614        use commonware_codec::conformance::CodecConformance;
615        use commonware_cryptography::{ed25519::PublicKey, sha256::Digest as Sha256Digest};
616
617        type Scheme = bls12381_threshold::Scheme<PublicKey, MinSig>;
618
619        commonware_conformance::conformance_tests! {
620            CodecConformance<Item<Sha256Digest>>,
621            CodecConformance<Ack<Scheme, Sha256Digest>>,
622            CodecConformance<TipAck<Scheme, Sha256Digest>>,
623            CodecConformance<Certificate<Scheme, Sha256Digest>>,
624            CodecConformance<Activity<Scheme, Sha256Digest>>,
625        }
626    }
627}
commonware_consensus/aggregation/types.rs

commonware_consensus/aggregation/
types.rs
