Reload validator public keys from Web3Signer instances on each epoch if previously failed to load or CLI option is set

Author: povi

Cargo.lock

@@ -685,6 +685,10 @@ struct ValidatorOptions {
     #[clap(long, num_args = 1.., value_delimiter = ',')]
     web3signer_public_keys: Vec<PublicKeyBytes>,
 
+    /// Refetches keys from Web3Signer once every epoch. This overwrites changes done via Keymanager API
+    #[clap(long)]
+    web3signer_refresh_keys_every_epoch: bool,
+
     /// [DEPRECATED] List of Web3Signer API URLs
     #[clap(long, num_args = 1..)]
     web3signer_api_urls: Vec<Url>,
@@ -873,6 +877,7 @@ impl GrandineArgs {
             builder_max_skipped_slots_per_epoch,
             use_validator_key_cache,
             web3signer_public_keys,
+            web3signer_refresh_keys_every_epoch,
             web3signer_api_urls,
             web3signer_urls,
             slashing_protection_history_limit,
@@ -1152,6 +1157,7 @@ impl GrandineArgs {
 
         let web3signer_config = Web3SignerConfig {
             public_keys: web3signer_public_keys.into_iter().collect(),
+            allow_to_reload_keys: web3signer_refresh_keys_every_epoch,
             urls: web3signer_urls,
         };
 

grandine/src/grandine_args.rs

@@ -10,7 +10,6 @@ workspace = true
 anyhow = { workspace = true }
 bls = { workspace = true }
 builder_api = { workspace = true }
-derive_more = { workspace = true }
 futures = { workspace = true }
 itertools = { workspace = true }
 log = { workspace = true }

signer/Cargo.toml

@@ -10,6 +10,7 @@ use futures::{
     try_join,
 };
 use itertools::Itertools as _;
+use log::info;
 use prometheus_metrics::Metrics;
 use rayon::iter::{IntoParallelIterator as _, ParallelIterator as _};
 use reqwest::{Client, Url};
@@ -151,6 +152,38 @@ impl Signer {
         self.sign_methods.is_empty()
     }
 
+    pub async fn refresh_keys_from_web3signer(&mut self) {
+        for (url, remote_keys) in self.web3signer.load_public_keys().await {
+            self.sign_methods
+                .retain(|public_key, sign_method| match sign_method {
+                    SignMethod::SecretKey(_, _) => true,
+                    SignMethod::Web3Signer(api_url) => {
+                        let retain = url != api_url || remote_keys.contains(public_key);
+
+                        if !retain {
+                            info!(
+                                "Validator credentials with public key {:?} were removed from Web3Signer at {}",
+                                public_key, url,
+                            );
+                        }
+
+                        retain
+                    }
+                });
+
+            for public_key in remote_keys {
+                self.sign_methods.entry(public_key).or_insert_with(|| {
+                    info!(
+                        "Validator credentials with public key {:?} were added to Web3Signer at {}",
+                        public_key, url,
+                    );
+
+                    SignMethod::Web3Signer(url.clone())
+                });
+            }
+        }
+    }
+
     pub async fn sign<'block, P: Preset>(
         &self,
         message: SigningMessage<'block, P>,

signer/src/signer.rs

@@ -5,7 +5,6 @@ use std::{
 
 use anyhow::Result;
 use bls::{PublicKeyBytes, SignatureBytes};
-use derive_more::Constructor;
 use log::warn;
 use prometheus_metrics::Metrics;
 use reqwest::{Client, Url};
@@ -17,24 +16,36 @@ use super::types::{SigningRequest, SigningResponse};
 
 #[derive(Clone, Default, Debug)]
 pub struct Config {
+    pub allow_to_reload_keys: bool,
     pub public_keys: HashSet<PublicKeyBytes>,
     pub urls: Vec<Url>,
 }
 
-#[derive(Clone, Constructor)]
+#[derive(Clone)]
 pub struct Web3Signer {
     client: Client,
     config: Config,
     metrics: Option<Arc<Metrics>>,
+    keys_loaded: HashSet<Url>,
 }
 
 impl Web3Signer {
+    #[must_use]
+    pub fn new(client: Client, config: Config, metrics: Option<Arc<Metrics>>) -> Self {
+        Self {
+            client,
+            config,
+            metrics,
+            keys_loaded: HashSet::new(),
+        }
+    }
+
     #[must_use]
     pub const fn client(&self) -> &Client {
         &self.client
     }
 
-    pub async fn load_public_keys(&self) -> HashMap<&Url, HashSet<PublicKeyBytes>> {
+    pub async fn load_public_keys(&mut self) -> HashMap<&Url, HashSet<PublicKeyBytes>> {
         let _timer = self
             .metrics
             .as_ref()
@@ -43,13 +54,18 @@ impl Web3Signer {
         let mut keys = HashMap::new();
 
         for url in &self.config.urls {
+            if !self.config.allow_to_reload_keys && self.keys_loaded.contains(url) {
+                continue;
+            }
+
             match self.load_public_keys_from_url(url).await {
                 Ok(mut remote_keys) => {
                     if !self.config.public_keys.is_empty() {
                         remote_keys.retain(|pubkey| self.config.public_keys.contains(pubkey));
                     }
 
                     keys.insert(url, remote_keys);
+                    self.keys_loaded.insert(url.clone());
                 }
                 Err(error) => warn!("failed to load Web3Signer keys from {url}: {error:?}"),
             }
@@ -135,16 +151,53 @@ mod tests {
 
         let url = Url::parse(&server.url("/"))?;
         let config = super::Config {
+            allow_to_reload_keys: false,
             public_keys: HashSet::new(),
             urls: vec![url.clone()],
         };
-        let web3signer = Web3Signer::new(Client::new(), config, None);
+        let mut web3signer = Web3Signer::new(Client::new(), config, None);
 
         let response = web3signer.load_public_keys().await;
         let expected = HashMap::from([(&url, HashSet::from([SAMPLE_PUBKEY, SAMPLE_PUBKEY_2]))]);
 
         assert_eq!(response, expected);
 
+        let response = web3signer.load_public_keys().await;
+        // By default, do not load pubkeys from Web3Signer again if keys were loaded
+        let expected = HashMap::new();
+
+        assert_eq!(response, expected);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_load_public_keys_if_reload_is_allowed() -> Result<()> {
+        let server = MockServer::start();
+
+        server.mock(|when, then| {
+            when.method(Method::GET).path("/api/v1/eth2/publicKeys");
+            then.status(200)
+                .body(json!([SAMPLE_PUBKEY, SAMPLE_PUBKEY_2]).to_string());
+        });
+
+        let url = Url::parse(&server.url("/"))?;
+        let config = super::Config {
+            allow_to_reload_keys: true,
+            public_keys: HashSet::new(),
+            urls: vec![url.clone()],
+        };
+        let mut web3signer = Web3Signer::new(Client::new(), config, None);
+
+        let response = web3signer.load_public_keys().await;
+        let expected = HashMap::from([(&url, HashSet::from([SAMPLE_PUBKEY, SAMPLE_PUBKEY_2]))]);
+
+        assert_eq!(response, expected);
+
+        let response = web3signer.load_public_keys().await;
+
+        assert_eq!(response, expected);
+
         Ok(())
     }
 
@@ -160,10 +213,11 @@ mod tests {
 
         let url = Url::parse(&server.url("/"))?;
         let config = super::Config {
+            allow_to_reload_keys: false,
             public_keys: vec![SAMPLE_PUBKEY_2].into_iter().collect(),
             urls: vec![url.clone()],
         };
-        let web3signer = Web3Signer::new(Client::new(), config, None);
+        let mut web3signer = Web3Signer::new(Client::new(), config, None);
 
         let response = web3signer.load_public_keys().await;
         let expected = HashMap::from([(&url, HashSet::from([SAMPLE_PUBKEY_2]))]);
@@ -186,6 +240,7 @@ mod tests {
 
         let url = Url::parse(&server.url("/"))?;
         let config = super::Config {
+            allow_to_reload_keys: false,
             public_keys: HashSet::new(),
             urls: vec![url.clone()],
         };

signer/src/web3signer/api.rs

@@ -770,6 +770,7 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
                 if misc::is_epoch_start::<P>(slot) {
                     let current_epoch = misc::compute_epoch_at_slot::<P>(slot);
                     self.spawn_slashing_protection_pruning(current_epoch);
+                    self.refresh_signer_keys();
                 }
             }
             _ => {}
@@ -2947,6 +2948,18 @@ impl<P: Preset, W: Wait + Sync> Validator<P, W> {
         misc::compute_start_slot_at_epoch::<P>(epoch)
     }
 
+    fn refresh_signer_keys(&self) {
+        let signer_arc = self.signer.clone_arc();
+
+        tokio::spawn(async move {
+            signer_arc
+                .write()
+                .await
+                .refresh_keys_from_web3signer()
+                .await
+        });
+    }
+
     fn get_execution_payload_header(
         &self,
         slot_head: &SlotHead<P>,