feat: implement cluster/helpers

crates/cli/src/commands/create_enr.rs

@@ -37,7 +37,7 @@ pub fn run(args: CreateEnrArgs) -> Result<()> {
 
     let key = k1::new_saved_priv_key(&args.data_dir)?;
 
-    let record = Record::new(key, Vec::new())?;
+    let record = Record::new(&key, Vec::new())?;
     let key_path = k1::key_path(&args.data_dir);
 
     let mut writer = io::stdout();

crates/cli/src/commands/enr.rs

@@ -47,7 +47,7 @@ pub fn run(args: EnrArgs) -> Result<()> {
         }
     };
 
-    let record = Record::new(key.clone(), vec![])?;
+    let record = Record::new(&key, vec![])?;
 
     writeln!(writer, "{}", record)?;
 

crates/cli/src/commands/test/mod.rs

@@ -402,7 +402,7 @@ pub(crate) async fn publish_result_to_obol_api(
     private_key_file: impl AsRef<Path>,
 ) -> CliResult<()> {
     let private_key = load_or_generate_key(private_key_file.as_ref()).await?;
-    let enr = Record::new(private_key.clone(), vec![])?;
+    let enr = Record::new(&private_key, vec![])?;
     let sign_data_bytes = serde_json::to_vec(&data)?;
     let hash = hash_ssz(&sign_data_bytes)?;
     let sig = sign(&private_key, &hash)?;

crates/cluster/Cargo.toml

@@ -26,6 +26,8 @@ pluto-eth2util.workspace = true
 pluto-eth1wrap.workspace = true
 pluto-k1util.workspace = true
 k256.workspace = true
+tokio.workspace = true
+reqwest = { workspace = true, features = ["json"] }
 
 [build-dependencies]
 prost-build.workspace = true
@@ -34,7 +36,8 @@ prost-build.workspace = true
 test-case.workspace = true
 pluto-testutil.workspace = true
 rand.workspace = true
-tokio.workspace = true
+tempfile.workspace = true
+wiremock.workspace = true
 
 [lints]
 workspace = true

crates/cluster/src/definition.rs

@@ -11,7 +11,7 @@ use crate::{
     ssz_hasher::Hasher,
     version::{CURRENT_VERSION, DKG_ALGO, versions::*},
 };
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Timelike, Utc};
 use libp2p::PeerId;
 use pluto_eth1wrap::{EthClient, EthClientError};
 use pluto_eth2util::enr::{Record, RecordError};
@@ -394,7 +394,12 @@ impl Definition {
             uuid: uuid.to_string(),
             name,
             version: CURRENT_VERSION.to_string(),
-            timestamp: Utc::now().to_string(),
+            // TODO: This is very error prone and should be replaced with a controlled timestamp in
+            // UTC.
+            timestamp: chrono::Local::now()
+                .with_nanosecond(0)
+                .expect("nanoseconds = 0")
+                .to_rfc3339(),
             num_validators,
             threshold,
             dkg_algorithm: DKG_ALGO.to_string(),
@@ -441,7 +446,9 @@ impl Definition {
             return Err(InvalidGasLimitError::GasLimitNotSet.into());
         }
 
-        def.set_definition_hashes()
+        def.set_definition_hashes()?;
+
+        Ok(def)
     }
 
     /// Returns the timestamp of the definition.
@@ -479,7 +486,7 @@ impl Definition {
         // there are no EIP712 signatures before v1.3.0. For definition versions
         // earlier than v1.3.0, error if either config signature or enr signature for
         // any operator is present.
-        if !Self::support_eip712_sigs(self.version.as_str()) {
+        if !Self::support_eip712_sigs(&self.version) {
             return if Self::eip712_sigs_present(&self.operators) {
                 Err(DefinitionError::OlderVersionSignaturesNotSupported)
             } else {
@@ -664,18 +671,18 @@ impl Definition {
     }
 
     /// Sets the definition hashes.
-    pub fn set_definition_hashes(mut self) -> Result<Self, DefinitionError> {
+    pub fn set_definition_hashes(&mut self) -> Result<(), DefinitionError> {
         let config_hash =
-            hash_definition(&self, true).map_err(|e| DefinitionError::SSZError(Box::new(e)))?;
+            hash_definition(self, true).map_err(|e| DefinitionError::SSZError(Box::new(e)))?;
 
         self.config_hash = config_hash.to_vec();
 
         let definition_hash =
-            hash_definition(&self, false).map_err(|e| DefinitionError::SSZError(Box::new(e)))?;
+            hash_definition(self, false).map_err(|e| DefinitionError::SSZError(Box::new(e)))?;
 
         self.definition_hash = definition_hash.to_vec();
 
-        Ok(self)
+        Ok(())
     }
 
     /// `verify_hashes` returns an error if hashes populated from json object
@@ -707,8 +714,8 @@ impl Definition {
     /// Returns true if the provided definition version supports EIP712
     /// signatures. Note that Definition versions prior to v1.3.0 don't
     /// support EIP712 signatures.
-    pub(crate) fn support_eip712_sigs(version: &str) -> bool {
-        !matches!(version, V1_0 | V1_1 | V1_2)
+    pub fn support_eip712_sigs(version: impl AsRef<str>) -> bool {
+        !matches!(version.as_ref(), V1_0 | V1_1 | V1_2)
     }
 
     fn eip712_sigs_present(operators: &[Operator]) -> bool {

crates/cluster/src/eip712sigs.rs

@@ -164,7 +164,8 @@ pub(crate) fn digest_eip712(
     Ok(digest)
 }
 
-fn sign_eip712(
+/// Returns the EIP712 signature for the primary type.
+pub(crate) fn sign_eip712(
     secret_key: &SecretKey,
     typ: &EIP712Type,
     definition: &Definition,

crates/cluster/src/helpers.rs

@@ -1,13 +1,17 @@
 use chrono::{DateTime, Utc};
+use pluto_crypto::tbls::Tbls;
 use pluto_eth2util::helpers::{checksum_address, public_key_to_address};
 use pluto_k1util::K1UtilError;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_with::{DeserializeAs, SerializeAs};
-use std::borrow::Cow;
+use std::{borrow::Cow, path::PathBuf};
 
-use crate::{definition::ADDRESS_LEN, ssz::SSZError, ssz_hasher::HashWalker};
-
-type VerifySigResult<T> = std::result::Result<T, VerifySigError>;
+use crate::{
+    definition::{self, ADDRESS_LEN, Definition},
+    eip712sigs, operator,
+    ssz::SSZError,
+    ssz_hasher::HashWalker,
+};
 
 /// Error type returned by `verify_sig`.
 #[derive(Debug, thiserror::Error)]
@@ -22,13 +26,65 @@ pub enum VerifySigError {
 }
 
 /// Returns true if the signature matches the digest and expected address.
-pub fn verify_sig(expected_addr: &str, digest: &[u8], sig: &[u8]) -> VerifySigResult<bool> {
+pub fn verify_sig(
+    expected_addr: &str,
+    digest: &[u8],
+    sig: &[u8],
+) -> std::result::Result<bool, VerifySigError> {
     let expected_addr = checksum_address(expected_addr)?;
     let recovered = pluto_k1util::recover(digest, sig)?;
     let actual_addr = public_key_to_address(&recovered);
     Ok(expected_addr == actual_addr)
 }
 
+/// Error type returned by `fetch_definition`.
+#[derive(Debug, thiserror::Error)]
+pub enum FetchError {
+    /// Timeout while fetching the definition.
+    #[error("timeout {0}")]
+    Timeout(#[from] tokio::time::error::Elapsed),
+
+    /// HTTP error while fetching the definition.
+    #[error("HTTP error {0}")]
+    Http(#[from] reqwest::Error),
+}
+
+/// Fetch cluster definition file from a remote URI.
+pub async fn fetch_definition(
+    url: impl reqwest::IntoUrl,
+) -> std::result::Result<Definition, FetchError> {
+    let definition = tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        let response = reqwest::get(url).await?.error_for_status()?;
+        response.json::<Definition>().await
+    })
+    .await??;
+
+    Ok(definition)
+}
+
+/// Creates a new directory for validator keys.
+/// If the directory "validator_keys" exists, it checks if the directory is
+/// empty.
+pub async fn create_validator_keys_dir(parent_dir: &std::path::Path) -> std::io::Result<PathBuf> {
+    let vk_dir = parent_dir.join("validator_keys");
+
+    if let Err(e) = tokio::fs::create_dir(&vk_dir).await {
+        if e.kind() != std::io::ErrorKind::AlreadyExists {
+            return Err(e);
+        }
+
+        let mut entries = tokio::fs::read_dir(&vk_dir).await?;
+        if entries.next_entry().await?.is_some() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::AlreadyExists,
+                "validator_keys directory exists and is not empty",
+            ));
+        }
+    }
+
+    Ok(vk_dir)
+}
+
 /// EthHex represents byte slices that are json formatted as 0x prefixed hex.
 /// Can be used both as a standalone type and with serde_as.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -239,8 +295,69 @@ pub fn to_0x_hex(bytes: &[u8]) -> String {
     format!("0x{}", hex::encode(bytes))
 }
 
+/// Signs the creator's config hash.
+pub fn sign_creator(
+    secret: &k256::SecretKey,
+    definition: &mut definition::Definition,
+) -> Result<(), eip712sigs::EIP712Error> {
+    let config_signature = eip712sigs::sign_eip712(
+        secret,
+        &eip712sigs::eip712_creator_config_hash(),
+        definition,
+        &operator::Operator::default(),
+    )?;
+
+    definition.creator.config_signature = config_signature;
+
+    Ok(())
+}
+
+/// Signs the operator's config hash and enr.
+pub fn sign_operator(
+    secret: &k256::SecretKey,
+    definition: &definition::Definition,
+    operator: &mut operator::Operator,
+) -> Result<(), crate::eip712sigs::EIP712Error> {
+    let config_signature = crate::eip712sigs::sign_eip712(
+        secret,
+        &crate::eip712sigs::get_operator_eip712_type(&definition.version),
+        definition,
+        operator,
+    )?;
+
+    let enr_signature = crate::eip712sigs::sign_eip712(
+        secret,
+        &crate::eip712sigs::eip712_enr(),
+        definition,
+        operator,
+    )?;
+
+    operator.config_signature = config_signature;
+    operator.enr_signature = enr_signature;
+
+    Ok(())
+}
+
+/// Returns a BLS aggregate signature of the message signed by all the shares.
+pub fn agg_sign(
+    secrets: &[Vec<pluto_crypto::types::PrivateKey>],
+    message: &[u8],
+) -> Result<pluto_crypto::types::Signature, pluto_crypto::types::Error> {
+    let blst = pluto_crypto::blst_impl::BlstImpl;
+
+    let sigs = secrets
+        .iter()
+        .flat_map(|shares| shares.iter())
+        .map(|share| blst.sign(share, message))
+        .collect::<Result<Vec<_>, _>>()?;
+
+    blst.aggregate(&sigs)
+}
+
 #[cfg(test)]
 mod tests {
+    use crate::test_cluster;
+
     use super::*;
     use serde_with::serde_as;
 
@@ -330,4 +447,81 @@ mod tests {
         assert!(json.contains("\"0x010203\""));
         assert!(json.contains("[4,5,6]"));
     }
+
+    #[tokio::test]
+    async fn fetch_definition_valid() {
+        let (lock, ..) = test_cluster::new_for_test(1, 2, 3, 0);
+        let expected_definition = lock.definition.clone();
+
+        let server = wiremock::MockServer::start().await;
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/validDef"))
+            .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(lock.definition))
+            .mount(&server)
+            .await;
+
+        let actual_definition = super::fetch_definition(format!("{}/validDef", &server.uri()))
+            .await
+            .unwrap();
+
+        assert_eq!(actual_definition, expected_definition);
+    }
+
+    #[tokio::test]
+    async fn fetch_definition_invalid() {
+        let server = wiremock::MockServer::start().await;
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/invalidDef"))
+            .respond_with(
+                wiremock::ResponseTemplate::new(200).set_body_raw("r#{}#", "application/json"),
+            )
+            .mount(&server)
+            .await;
+
+        let response = super::fetch_definition(format!("{}/invalidDef", &server.uri())).await;
+
+        assert!(matches!(response, Err(super::FetchError::Http(e)) if e.is_decode()));
+    }
+
+    #[tokio::test]
+    async fn fetch_definition_non_200() {
+        let server = wiremock::MockServer::start().await;
+        wiremock::Mock::given(wiremock::matchers::method("GET"))
+            .and(wiremock::matchers::path("/non_ok"))
+            .respond_with(wiremock::ResponseTemplate::new(500))
+            .mount(&server)
+            .await;
+
+        let response = super::fetch_definition(format!("{}/non_ok", &server.uri())).await;
+
+        assert!(matches!(response, Err(super::FetchError::Http(e)) if e.is_status()));
+    }
+
+    #[tokio::test]
+    async fn create_validator_keys_dir() {
+        let tmp = tempfile::tempdir().unwrap();
+        let parent_dir = tmp.path();
+
+        // First attempt must succeed.
+        let dir = super::create_validator_keys_dir(parent_dir).await.unwrap();
+        assert!(dir.starts_with(parent_dir));
+        assert!(dir.ends_with("validator_keys"));
+
+        // Second attempt shall succeed as long as the dir is empty.
+        let dir2 = super::create_validator_keys_dir(parent_dir).await.unwrap();
+        assert_eq!(dir, dir2);
+
+        // Create a file in the directory to make it non-empty.
+        tokio::fs::write(dir.join("file"), b"data").await.unwrap();
+        let err = super::create_validator_keys_dir(parent_dir)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, e if e.kind() == std::io::ErrorKind::AlreadyExists));
+
+        // Parent directory does not exist
+        let err = super::create_validator_keys_dir(&parent_dir.join("nonexistent"))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, e if e.kind() == std::io::ErrorKind::NotFound));
+    }
 }

crates/cluster/src/lib.rs

@@ -29,6 +29,7 @@ pub mod ssz;
 /// Cluster SSZ hashing management and coordination.
 pub mod ssz_hasher;
 /// Cluster test cluster management and coordination.
+#[cfg(test)]
 pub mod test_cluster;
 /// Cluster version management and coordination.
 pub mod version;

crates/cluster/src/manifest/cluster.rs

@@ -426,7 +426,7 @@ mod tests {
 
         for i in 0..operator_amt {
             let k1_key = generate_insecure_k1_key(i);
-            let enr = Record::new(k1_key.clone(), vec![]).unwrap();
+            let enr = Record::new(&k1_key, vec![]).unwrap();
 
             operators.push(Operator {
                 address: format!("0x{:040x}", i),
@@ -457,7 +457,7 @@ mod tests {
         let k1_key0 = generate_insecure_k1_key(1);
         let k1_key_unknown = generate_insecure_k1_key(200);
 
-        let enr0 = Record::new(k1_key0, vec![]).unwrap();
+        let enr0 = Record::new(&k1_key0, vec![]).unwrap();
 
         let cluster = Cluster {
             operators: vec![Operator {