feat: add tx encryption key rotation

bin/validator/src/commands/mod.rs

@@ -12,7 +12,6 @@ use miden_node_utils::clap::GrpcOptionsInternal;
 use miden_node_utils::logging::OpenTelemetry;
 use miden_node_utils::shutdown::CancellationToken;
 use miden_protocol::crypto::dsa::ecdsa_k256_keccak::SigningKey;
-use miden_protocol::crypto::dsa::eddsa_25519_sha512::KeyExchangeKey;
 use miden_protocol::utils::serde::Deserializable;
 use miden_validator::{
     DataDirectory,
@@ -132,10 +131,11 @@ pub enum ValidatorCommand {
         )]
         signing_key_kms_id: Option<String>,
 
-        /// Hex-encoded shared secret of the transaction encryption key.
+        /// Hex-encoded shared master secret of the transaction encryption key.
         ///
-        /// Unlike the per-validator signing key, this value must be identical across every
-        /// validator in the set.
+        /// The per-epoch encryption keys are derived from this secret, rotating automatically at
+        /// each epoch boundary. Unlike the per-validator signing key, this value must be
+        /// identical across every validator in the set.
         ///
         /// If not provided, a predefined insecure key is used.
         ///
@@ -149,8 +149,8 @@ pub enum ValidatorCommand {
         )]
         encryption_key: String,
 
-        /// Base64-encoded KMS ciphertext of the shared transaction encryption key, as returned
-        /// by `kms:Encrypt`.
+        /// Base64-encoded KMS ciphertext of the shared transaction encryption master secret, as
+        /// returned by `kms:Encrypt`.
         ///
         /// The wrapped key material is recovered at startup with `kms:Decrypt`. The ciphertext
         /// must have been produced by `kms:Encrypt` under a symmetric KMS key, whose ID is
@@ -232,10 +232,12 @@ impl ValidatorCommand {
                     hex::decode(encryption_key)
                         .context("failed to decode the encryption key hex")?
                 };
-                let encryption_key = KeyExchangeKey::read_from_bytes(&encryption_key_bytes)
-                    .context("failed to construct the encryption key")?;
+                let master_secret: [u8; 32] = encryption_key_bytes
+                    .as_slice()
+                    .try_into()
+                    .map_err(|_| anyhow::anyhow!("the encryption key must be exactly 32 bytes"))?;
                 let decrypter: Arc<dyn TransactionInputDecrypter> =
-                    Arc::new(LocalX25519TransactionInputDecrypter::new(encryption_key));
+                    Arc::new(LocalX25519TransactionInputDecrypter::new(master_secret));
 
                 let signer = if let Some(kms_key_id) = signing_key_kms_id {
                     ValidatorSigner::new_kms(kms_key_id).await?

bin/validator/src/db/migrations.rs

@@ -71,7 +71,7 @@ mod tests {
     use super::*;
 
     const EXPECTED_SCHEMA_HASHES: [SchemaHash; 1] = [SchemaHash::from_hex(
-        "100025e3daa05c2f7d5be2dc6ff096dbe916f1af4d95ae27cdfb2e23d6f0723a",
+        "c8fc914e8109e47844744acea6a590dc3364acc7cc489205143bc0ee69b54520",
     )];
 
     #[test]

bin/validator/src/db/migrations/001_initial.sql

@@ -17,3 +17,11 @@ CREATE TABLE block_headers (
     block_num    BIGINT PRIMARY KEY,
     block_header BLOB NOT NULL
 ) WITHOUT ROWID;
+
+CREATE TABLE encryption_keys (
+    epoch      BIGINT PRIMARY KEY,
+    scheme     BIGINT NOT NULL,
+    key_id     BLOB NOT NULL,
+    public_key BLOB NOT NULL,
+    secret_key BLOB NOT NULL
+) WITHOUT ROWID;

bin/validator/src/db/mod.rs

@@ -24,6 +24,11 @@ mod sql {
     pub(super) const COUNT_VALIDATED_TRANSACTIONS: &str =
         include_str!("sql/count_validated_transactions.sql");
     pub(super) const COUNT_SIGNED_BLOCKS: &str = include_str!("sql/count_signed_blocks.sql");
+    pub(super) const INSERT_ENCRYPTION_KEY: &str = include_str!("sql/insert_encryption_key.sql");
+    pub(super) const MAX_ARCHIVED_ENCRYPTION_KEY_EPOCH: &str =
+        include_str!("sql/max_archived_encryption_key_epoch.sql");
+    #[cfg(test)]
+    pub(super) const LOAD_ENCRYPTION_KEY: &str = include_str!("sql/load_encryption_key.sql");
 }
 
 /// Open a connection to the DB after verifying that it is at the latest schema version.
@@ -261,6 +266,93 @@ pub fn count_signed_blocks(tx: &ReadTx<'_>) -> Result<i64, DatabaseError> {
         .unwrap_or(0))
 }
 
+/// A transaction encryption key of one epoch, as archived in the database.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct ArchivedEncryptionKey {
+    /// Wire identifier of the encryption scheme.
+    pub scheme: u32,
+    /// Opaque identifier of the encryption key.
+    pub key_id: Vec<u8>,
+    /// Raw public key bytes of the encryption key.
+    pub public_key: Vec<u8>,
+    /// Raw secret key bytes of the encryption key.
+    pub secret_key: Vec<u8>,
+}
+
+/// Archives one epoch's encryption key.
+///
+/// A key is immutable once derived, so a row that already exists for the epoch is left
+/// untouched.
+#[miden_instrument(
+    target = COMPONENT,
+    skip(tx, key),
+    err,
+)]
+pub(crate) fn insert_encryption_key(
+    tx: &WriteTx<'_>,
+    epoch: u16,
+    key: &ArchivedEncryptionKey,
+) -> Result<(), DatabaseError> {
+    tx.execute(
+        sql::INSERT_ENCRYPTION_KEY,
+        &[
+            &i64::from(epoch),
+            &i64::from(key.scheme),
+            &key.key_id,
+            &key.public_key,
+            &key.secret_key,
+        ],
+    )?;
+    Ok(())
+}
+
+/// Returns the highest epoch whose encryption key has been archived, or `None` when the archive is
+/// empty.
+#[miden_instrument(
+    target = COMPONENT,
+    skip(tx),
+    err,
+)]
+pub(crate) fn max_archived_encryption_key_epoch(
+    tx: &ReadTx<'_>,
+) -> Result<Option<u16>, DatabaseError> {
+    tx.query(sql::MAX_ARCHIVED_ENCRYPTION_KEY_EPOCH, &[], |row| row.get::<Option<i64>>(0))?
+        .into_iter()
+        .next()
+        .flatten()
+        .map(|epoch| {
+            u16::try_from(epoch).map_err(|err| {
+                DatabaseError::deserialization("archived epoch out of the u16 range", err)
+            })
+        })
+        .transpose()
+}
+
+/// Loads the archived encryption key of the given epoch.
+///
+/// Returns `None` if no key has been archived for the epoch.
+///
+/// Test-only until an archive recovery path consumes it.
+#[cfg(test)]
+pub(crate) fn load_encryption_key(
+    tx: &ReadTx<'_>,
+    epoch: u16,
+) -> Result<Option<ArchivedEncryptionKey>, DatabaseError> {
+    Ok(tx
+        .query(sql::LOAD_ENCRYPTION_KEY, &[&i64::from(epoch)], |row| {
+            Ok(ArchivedEncryptionKey {
+                scheme: u32::try_from(row.get::<i64>(0)?).map_err(|err| {
+                    DatabaseError::deserialization("archived scheme out of the u32 range", err)
+                })?,
+                key_id: row.get(1)?,
+                public_key: row.get(2)?,
+                secret_key: row.get(3)?,
+            })
+        })?
+        .into_iter()
+        .next())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -323,4 +415,47 @@ mod tests {
             .unwrap();
         assert!(!unknown_exists, "an unknown transaction id should not be reported as existing");
     }
+
+    fn test_archived_key(marker: u8) -> ArchivedEncryptionKey {
+        ArchivedEncryptionKey {
+            scheme: 1,
+            key_id: vec![marker; 4],
+            public_key: vec![marker; 32],
+            secret_key: vec![marker; 32],
+        }
+    }
+
+    /// Archived keys round-trip, the max archived epoch tracks inserts, and re-inserting an epoch
+    /// leaves the original row untouched.
+    #[tokio::test]
+    async fn encryption_key_archive_roundtrip() {
+        let temp_dir = tempfile::tempdir().expect("failed to create temp directory");
+        let db = setup(temp_dir.path().join("validator.sqlite3")).await.unwrap();
+
+        // The archive starts empty.
+        let max = db.read("max_epoch", max_archived_encryption_key_epoch).await.unwrap();
+        assert_eq!(max, None);
+        let missing = db.read("load_key", |tx| load_encryption_key(tx, 0)).await.unwrap();
+        assert_eq!(missing, None);
+
+        // Insert two epochs and read them back.
+        for (epoch, marker) in [(0u16, 7u8), (3u16, 9u8)] {
+            let key = test_archived_key(marker);
+            db.write("insert_key", move |tx| insert_encryption_key(tx, epoch, &key))
+                .await
+                .unwrap();
+        }
+        let max = db.read("max_epoch", max_archived_encryption_key_epoch).await.unwrap();
+        assert_eq!(max, Some(3));
+        let loaded = db.read("load_key", |tx| load_encryption_key(tx, 3)).await.unwrap();
+        assert_eq!(loaded, Some(test_archived_key(9)));
+
+        // Re-inserting an archived epoch must not overwrite the existing row.
+        let conflicting = test_archived_key(5);
+        db.write("insert_key", move |tx| insert_encryption_key(tx, 3, &conflicting))
+            .await
+            .unwrap();
+        let loaded = db.read("load_key", |tx| load_encryption_key(tx, 3)).await.unwrap();
+        assert_eq!(loaded, Some(test_archived_key(9)), "archived keys must be immutable");
+    }
 }

bin/validator/src/db/sql/insert_encryption_key.sql

@@ -0,0 +1,2 @@
+INSERT OR IGNORE INTO encryption_keys (epoch, scheme, key_id, public_key, secret_key)
+VALUES (?1, ?2, ?3, ?4, ?5)

bin/validator/src/db/sql/load_encryption_key.sql

@@ -0,0 +1 @@
+SELECT scheme, key_id, public_key, secret_key FROM encryption_keys WHERE epoch = ?1

bin/validator/src/db/sql/max_archived_encryption_key_epoch.sql

@@ -0,0 +1 @@
+SELECT MAX(epoch) FROM encryption_keys

bin/validator/src/lib.rs

@@ -7,6 +7,7 @@ mod tx_validation;
 pub use data_directory::DataDirectory;
 pub use server::ValidatorServer;
 pub use signers::{
+    EncryptionKeySet,
     KmsSigner,
     LocalX25519TransactionInputDecrypter,
     NextEncryptionKeyInfo,

bin/validator/src/server/mod.rs

@@ -94,24 +94,28 @@ impl ValidatorServer {
             .build_v1()
             .context("failed to build reflection service")?;
 
+        let service = ValidatorService::new(
+            self.signer,
+            self.decrypter,
+            db,
+            block_store,
+            initial_chain_tip,
+            initial_tx_count,
+            initial_block_count,
+        )
+        .await
+        .context("failed to initialize validator server")?;
+
+        // Rotate and re-attest the transaction encryption key as the chain crosses epoch
+        // boundaries. The task follows the committed tip and stops on shutdown.
+        service.spawn_key_rotation_task(shutdown.clone());
+
         // Build the gRPC server with the API service and trace layer.
         tonic::transport::Server::builder()
             .layer(CatchPanicLayer::custom(catch_panic_layer_fn))
             .layer(TraceLayer::new_for_grpc().make_span_with(grpc_trace_fn))
             .timeout(self.grpc_options.request_timeout)
-            .add_service(validator_api::service(
-                ValidatorService::new(
-                    self.signer,
-                    self.decrypter,
-                    db,
-                    block_store,
-                    initial_chain_tip,
-                    initial_tx_count,
-                    initial_block_count,
-                )
-                .await
-                .context("failed to initialize validator server")?,
-            ))
+            .add_service(validator_api::service(service))
             .add_service(reflection_service)
             .serve_with_incoming_shutdown(
                 TcpListenerStream::new(listener),

bin/validator/src/server/validator_service/get_transaction_encryption_key.rs

@@ -4,17 +4,20 @@ use miden_tx::utils::serde::Serializable;
 
 use super::ValidatorService;
 use crate::COMPONENT;
+use crate::signers::TransactionEncryptionKeyInfo;
 
 #[tonic::async_trait]
 impl grpc::server::validator_api::GetTransactionEncryptionKey for ValidatorService {
     type Input = ();
-    type Output = grpc::transaction::TransactionEncryptionKey;
+    type Output = grpc::transaction::TransactionEncryptionKeyResponse;
 
     fn decode(request: ()) -> tonic::Result<Self::Input> {
         Ok(request)
     }
 
-    fn encode(output: Self::Output) -> tonic::Result<grpc::transaction::TransactionEncryptionKey> {
+    fn encode(
+        output: Self::Output,
+    ) -> tonic::Result<grpc::transaction::TransactionEncryptionKeyResponse> {
         Ok(output)
     }
 
@@ -30,25 +33,47 @@ impl grpc::server::validator_api::GetTransactionEncryptionKey for ValidatorServi
         _metadata: &tonic::metadata::MetadataMap,
         _extensions: &tonic::codegen::http::Extensions,
     ) -> tonic::Result<Self::Output> {
-        // Built entirely from state fixed at construction, so the endpoint stays available while a
+        // Built entirely from in-memory attested state, so the endpoint stays available while a
         // backup subscription holds the serve lock.
-        Ok(grpc::transaction::TransactionEncryptionKey {
-            scheme: i32::try_from(self.encryption_key_info.scheme)
-                .expect("scheme identifier must fit in i32"),
-            key_id: self.encryption_key_info.key_id.clone(),
-            public_key: self.encryption_key_info.public_key.clone(),
-            attestations: vec![grpc::transaction::ValidatorKeyAttestation {
-                validator_public_key: self.signer.public_key().to_bytes(),
-                signature: self.encryption_key_attestation.to_bytes(),
-            }],
-            next_key: self.encryption_key_info.next_key.as_ref().map(|next| {
-                grpc::transaction::NextTransactionEncryptionKey {
-                    scheme: i32::try_from(next.scheme).expect("scheme identifier must fit in i32"),
-                    key_id: next.key_id.clone(),
-                    public_key: next.public_key.clone(),
-                    rotation_block_num: next.rotation_block_num,
-                }
-            }),
+        let attested = self.attested_encryption_keys();
+        let validator_public_key = self.signer.public_key().to_bytes();
+
+        let current_key = encode_key(
+            &attested.keys.current,
+            &validator_public_key,
+            &attested.current_attestation.to_bytes(),
+        );
+        let next_key = attested.keys.next.as_ref().map(|next| {
+            let attestation = attested
+                .next_attestation
+                .as_ref()
+                .expect("a next key is always attested together with the current key");
+            grpc::transaction::NextTransactionEncryptionKey {
+                key: Some(encode_key(&next.key, &validator_public_key, &attestation.to_bytes())),
+                rotation_block_num: next.rotation_block_num,
+            }
+        });
+
+        Ok(grpc::transaction::TransactionEncryptionKeyResponse {
+            current_key: Some(current_key),
+            next_key,
         })
     }
 }
+
+/// Encodes one attested encryption key in wire format.
+fn encode_key(
+    key: &TransactionEncryptionKeyInfo,
+    validator_public_key: &[u8],
+    signature: &[u8],
+) -> grpc::transaction::TransactionEncryptionKey {
+    grpc::transaction::TransactionEncryptionKey {
+        scheme: i32::try_from(key.scheme).expect("scheme identifier must fit in i32"),
+        key_id: key.key_id.clone(),
+        public_key: key.public_key.clone(),
+        attestations: vec![grpc::transaction::ValidatorKeyAttestation {
+            validator_public_key: validator_public_key.to_vec(),
+            signature: signature.to_vec(),
+        }],
+    }
+}