feat: add parsigex

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "crates/app",
+    "crates/parsigex",
     "crates/build-proto",
     "crates/cli",
     "crates/cluster",
@@ -99,6 +100,7 @@ wiremock = "0.6"
 
 # Crates in the workspace
 pluto-app = { path = "crates/app" }
+pluto-parsigex = { path = "crates/parsigex" }
 pluto-build-proto = { path = "crates/build-proto" }
 pluto-cli = { path = "crates/cli" }
 pluto-cluster = { path = "crates/cluster" }

crates/core/Cargo.toml

@@ -12,6 +12,7 @@ cancellation.workspace = true
 chrono.workspace = true
 crossbeam.workspace = true
 futures.workspace = true
+futures-timer.workspace = true
 dyn-clone.workspace = true
 dyn-eq.workspace = true
 hex.workspace = true
@@ -31,18 +32,25 @@ tokio-util.workspace = true
 tracing.workspace = true
 pluto-eth2util.workspace = true
 tree_hash.workspace = true
+unsigned-varint.workspace = true
 
 [dev-dependencies]
+anyhow.workspace = true
 alloy.workspace = true
+clap.workspace = true
 rand.workspace = true
 libp2p.workspace = true
+k256.workspace = true
 prost.workspace = true
 prost-types.workspace = true
 hex.workspace = true
 chrono.workspace = true
 test-case.workspace = true
 pluto-eth2util.workspace = true
+pluto-cluster.workspace = true
+pluto-p2p.workspace = true
 pluto-testutil.workspace = true
+pluto-tracing.workspace = true
 tokio = { workspace = true, features = ["test-util"] }
 wiremock.workspace = true
 pluto-ssz.workspace = true

crates/core/src/lib.rs

@@ -26,6 +26,10 @@ pub mod deadline;
 /// parsigdb
 pub mod parsigdb;
 
+mod parsigex_codec;
+
+pub use parsigex_codec::ParSigExCodecError;
+
 /// Test utilities.
 #[cfg(test)]
 pub mod testutils;

crates/core/src/parsigex_codec.rs

@@ -0,0 +1,118 @@
+//! Partial signature exchange codec helpers used by core types.
+
+use std::any::Any;
+
+use crate::{
+    signeddata::{
+        Attestation, BeaconCommitteeSelection, SignedAggregateAndProof, SignedRandao,
+        SignedSyncContributionAndProof, SignedSyncMessage, SignedVoluntaryExit,
+        SyncCommitteeSelection, VersionedAttestation, VersionedSignedAggregateAndProof,
+        VersionedSignedProposal, VersionedSignedValidatorRegistration,
+    },
+    types::{DutyType, Signature, SignedData},
+};
+
+/// Error type for partial signature exchange codec operations.
+#[derive(Debug, thiserror::Error)]
+pub enum ParSigExCodecError {
+    /// Missing duty or data set fields.
+    #[error("invalid parsigex msg fields")]
+    InvalidMessageFields,
+
+    /// Invalid partial signed data set proto.
+    #[error("invalid partial signed data set proto fields")]
+    InvalidParSignedDataSetFields,
+
+    /// Invalid partial signed proto.
+    #[error("invalid partial signed proto")]
+    InvalidParSignedProto,
+
+    /// Invalid duty type.
+    #[error("invalid duty")]
+    InvalidDuty,
+
+    /// Unsupported duty type.
+    #[error("unsupported duty type")]
+    UnsupportedDutyType,
+
+    /// Deprecated builder proposer duty.
+    #[error("deprecated duty builder proposer")]
+    DeprecatedBuilderProposer,
+
+    /// Failed to parse a public key.
+    #[error("invalid public key: {0}")]
+    InvalidPubKey(String),
+
+    /// Invalid share index.
+    #[error("invalid share index")]
+    InvalidShareIndex,
+
+    /// Serialization failed.
+    #[error("marshal signed data: {0}")]
+    Serialize(#[from] serde_json::Error),
+}
+
+pub(crate) fn serialize_signed_data(data: &dyn SignedData) -> Result<Vec<u8>, ParSigExCodecError> {
+    let any = data as &dyn Any;
+
+    macro_rules! serialize_as {
+        ($ty:ty) => {
+            if let Some(value) = any.downcast_ref::<$ty>() {
+                return Ok(serde_json::to_vec(value)?);
+            }
+        };
+    }
+
+    serialize_as!(Attestation);
+    serialize_as!(VersionedAttestation);
+    serialize_as!(VersionedSignedProposal);
+    serialize_as!(VersionedSignedValidatorRegistration);
+    serialize_as!(SignedVoluntaryExit);
+    serialize_as!(SignedRandao);
+    serialize_as!(Signature);
+    serialize_as!(BeaconCommitteeSelection);
+    serialize_as!(SignedAggregateAndProof);
+    serialize_as!(VersionedSignedAggregateAndProof);
+    serialize_as!(SignedSyncMessage);
+    serialize_as!(SyncCommitteeSelection);
+    serialize_as!(SignedSyncContributionAndProof);
+
+    Err(ParSigExCodecError::UnsupportedDutyType)
+}
+
+pub(crate) fn deserialize_signed_data(
+    duty_type: &DutyType,
+    bytes: &[u8],
+) -> Result<Box<dyn SignedData>, ParSigExCodecError> {
+    macro_rules! deserialize_json {
+        ($ty:ty) => {
+            serde_json::from_slice::<$ty>(bytes)
+                .map(|value| Box::new(value) as Box<dyn SignedData>)
+                .map_err(ParSigExCodecError::from)
+        };
+    }
+
+    match duty_type {
+        // Match Go order: old Attestation format first, then VersionedAttestation.
+        DutyType::Attester => deserialize_json!(Attestation)
+            .or_else(|_| deserialize_json!(VersionedAttestation))
+            .map_err(|_| ParSigExCodecError::UnsupportedDutyType),
+        DutyType::Proposer => deserialize_json!(VersionedSignedProposal),
+        DutyType::BuilderProposer => Err(ParSigExCodecError::DeprecatedBuilderProposer),
+        DutyType::BuilderRegistration => deserialize_json!(VersionedSignedValidatorRegistration),
+        DutyType::Exit => deserialize_json!(SignedVoluntaryExit),
+        DutyType::Randao => deserialize_json!(SignedRandao),
+        DutyType::Signature => deserialize_json!(Signature),
+        DutyType::PrepareAggregator => deserialize_json!(BeaconCommitteeSelection),
+        // Match Go order: old SignedAggregateAndProof format first, then versioned.
+        DutyType::Aggregator => deserialize_json!(SignedAggregateAndProof)
+            .or_else(|_| deserialize_json!(VersionedSignedAggregateAndProof))
+            .map_err(|_| ParSigExCodecError::UnsupportedDutyType),
+        DutyType::SyncMessage => deserialize_json!(SignedSyncMessage),
+        DutyType::PrepareSyncContribution => deserialize_json!(SyncCommitteeSelection),
+        DutyType::SyncContribution => deserialize_json!(SignedSyncContributionAndProof),
+        DutyType::Unknown | DutyType::InfoSync | DutyType::DutySentinel(_) => {
+            Err(ParSigExCodecError::UnsupportedDutyType)
+        }
+    }
+}

crates/core/src/signeddata.rs

@@ -48,6 +48,9 @@ pub enum SignedDataError {
     /// Invalid attestation wrapper JSON.
     #[error("unmarshal attestation")]
     AttestationJson,
+    /// Custom error.
+    #[error("{0}")]
+    Custom(Box<dyn std::error::Error>),
 }
 
 fn hash_root<T: TreeHash>(value: &T) -> [u8; 32] {

crates/core/src/types.rs

@@ -1,14 +1,19 @@
 //! Types for the Charon core.
 
-use std::{collections::HashMap, fmt::Display, iter};
+use std::{any::Any, collections::HashMap, fmt::Display, iter};
 
 use chrono::{DateTime, Duration, Utc};
 use dyn_clone::DynClone;
 use dyn_eq::DynEq;
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug as StdDebug;
 
-use crate::signeddata::SignedDataError;
+use crate::{
+    ParSigExCodecError,
+    corepb::v1::core as pbcore,
+    parsigex_codec::{deserialize_signed_data, serialize_signed_data},
+    signeddata::SignedDataError,
+};
 
 /// The type of duty.
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
@@ -66,6 +71,52 @@ impl DutyType {
     }
 }
 
+impl From<&DutyType> for i32 {
+    fn from(duty_type: &DutyType) -> Self {
+        match duty_type {
+            DutyType::Unknown => 0,
+            DutyType::Proposer => 1,
+            DutyType::Attester => 2,
+            DutyType::Signature => 3,
+            DutyType::Exit => 4,
+            DutyType::BuilderProposer => 5,
+            DutyType::BuilderRegistration => 6,
+            DutyType::Randao => 7,
+            DutyType::PrepareAggregator => 8,
+            DutyType::Aggregator => 9,
+            DutyType::SyncMessage => 10,
+            DutyType::PrepareSyncContribution => 11,
+            DutyType::SyncContribution => 12,
+            DutyType::InfoSync => 13,
+            DutyType::DutySentinel(_) => 14,
+        }
+    }
+}
+
+impl TryFrom<i32> for DutyType {
+    type Error = ParSigExCodecError;
+
+    fn try_from(value: i32) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(DutyType::Unknown),
+            1 => Ok(DutyType::Proposer),
+            2 => Ok(DutyType::Attester),
+            3 => Ok(DutyType::Signature),
+            4 => Ok(DutyType::Exit),
+            5 => Ok(DutyType::BuilderProposer),
+            6 => Ok(DutyType::BuilderRegistration),
+            7 => Ok(DutyType::Randao),
+            8 => Ok(DutyType::PrepareAggregator),
+            9 => Ok(DutyType::Aggregator),
+            10 => Ok(DutyType::SyncMessage),
+            11 => Ok(DutyType::PrepareSyncContribution),
+            12 => Ok(DutyType::SyncContribution),
+            13 => Ok(DutyType::InfoSync),
+            _ => Err(ParSigExCodecError::InvalidDuty),
+        }
+    }
+}
+
 /// SlotNumber struct
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct SlotNumber(u64);
@@ -192,6 +243,28 @@ impl Duty {
     }
 }
 
+impl From<&Duty> for pbcore::Duty {
+    fn from(duty: &Duty) -> Self {
+        Self {
+            slot: duty.slot.inner(),
+            r#type: i32::from(&duty.duty_type),
+        }
+    }
+}
+
+impl TryFrom<&pbcore::Duty> for Duty {
+    type Error = ParSigExCodecError;
+
+    fn try_from(duty: &pbcore::Duty) -> Result<Self, Self::Error> {
+        let duty_type = DutyType::try_from(duty.r#type)?;
+        if !duty_type.is_valid() {
+            return Err(ParSigExCodecError::InvalidDuty);
+        }
+
+        Ok(Self::new(duty.slot.into(), duty_type))
+    }
+}
+
 /// The type of proposal.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -452,7 +525,7 @@ impl AsRef<[u8; SIG_LEN]> for Signature {
 }
 
 /// Signed data type
-pub trait SignedData: DynClone + DynEq + StdDebug + Send + Sync {
+pub trait SignedData: Any + DynClone + DynEq + StdDebug + Send + Sync {
     /// signature returns the signed duty data's signature.
     fn signature(&self) -> Result<Signature, SignedDataError>;
 
@@ -517,6 +590,39 @@ impl ParSignedData {
     }
 }
 
+impl TryFrom<&ParSignedData> for pbcore::ParSignedData {
+    type Error = ParSigExCodecError;
+
+    fn try_from(data: &ParSignedData) -> Result<Self, Self::Error> {
+        let encoded = serialize_signed_data(data.signed_data.as_ref())?;
+        let share_idx =
+            i32::try_from(data.share_idx).map_err(|_| ParSigExCodecError::InvalidShareIndex)?;
+        let signature = data.signed_data.signature().map_err(|err| {
+            ParSigExCodecError::Serialize(serde_json::Error::io(std::io::Error::other(
+                err.to_string(),
+            )))
+        })?;
+
+        Ok(Self {
+            data: encoded.into(),
+            signature: signature.as_ref().to_vec().into(),
+            share_idx,
+        })
+    }
+}
+
+impl TryFrom<(&DutyType, &pbcore::ParSignedData)> for ParSignedData {
+    type Error = ParSigExCodecError;
+
+    fn try_from(value: (&DutyType, &pbcore::ParSignedData)) -> Result<Self, Self::Error> {
+        let (duty_type, data) = value;
+        let share_idx =
+            u64::try_from(data.share_idx).map_err(|_| ParSigExCodecError::InvalidShareIndex)?;
+        let signed_data = deserialize_signed_data(duty_type, &data.data)?;
+        Ok(Self::new_boxed(signed_data, share_idx))
+    }
+}
+
 /// ParSignedDataSet is a set of partially signed duty data only signed by a
 /// single threshold BLS share.
 #[derive(Debug, Clone, PartialEq, Eq, Default)]
@@ -554,6 +660,39 @@ impl ParSignedDataSet {
     }
 }
 
+impl TryFrom<&ParSignedDataSet> for pbcore::ParSignedDataSet {
+    type Error = ParSigExCodecError;
+
+    fn try_from(set: &ParSignedDataSet) -> Result<Self, Self::Error> {
+        let mut out = std::collections::BTreeMap::new();
+        for (pub_key, value) in set.inner() {
+            out.insert(pub_key.to_string(), pbcore::ParSignedData::try_from(value)?);
+        }
+
+        Ok(Self { set: out })
+    }
+}
+
+impl TryFrom<(&DutyType, &pbcore::ParSignedDataSet)> for ParSignedDataSet {
+    type Error = ParSigExCodecError;
+
+    fn try_from(value: (&DutyType, &pbcore::ParSignedDataSet)) -> Result<Self, Self::Error> {
+        let (duty_type, set) = value;
+        if set.set.is_empty() {
+            return Err(ParSigExCodecError::InvalidParSignedDataSetFields);
+        }
+
+        let mut out = Self::new();
+        for (pub_key, value) in &set.set {
+            let pub_key = PubKey::try_from(pub_key.as_str())
+                .map_err(|_| ParSigExCodecError::InvalidPubKey(pub_key.clone()))?;
+            out.insert(pub_key, ParSignedData::try_from((duty_type, value))?);
+        }
+
+        Ok(out)
+    }
+}
+
 /// SignedDataSet is a set of signed duty data.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SignedDataSet<T: SignedData>(HashMap<PubKey, T>);