diff --git a/server/crates/arbiter-server/src/actors/evm/mod.rs b/server/crates/arbiter-server/src/actors/evm/mod.rs index 481c9fa..77a0dd7 100644 --- a/server/crates/arbiter-server/src/actors/evm/mod.rs +++ b/server/crates/arbiter-server/src/actors/evm/mod.rs @@ -1,9 +1,13 @@ use crate::{ - actors::vault::{CreateNew, Decrypt, Vault}, + actors::{ + proposal_manager::events::ProposalApproved, + vault::{CreateNew, Decrypt, Vault}, + }, crypto::integrity, db::{ DatabaseError, DatabasePool, - models::{self, EvmWalletId}, + models::{self, EvmWalletId, ProposalId}, + proposal::{ProposalKind, grant_wallet_access, one_off_transaction, persistent_grant}, schema, }, evm::{ @@ -23,8 +27,9 @@ use diesel::{ ExpressionMethods, OptionalExtension as _, QueryDsl, SelectableHelper as _, dsl::insert_into, }; use diesel_async::RunQueryDsl; -use kameo::{Actor, actor::ActorRef, messages}; +use kameo::{Actor, actor::ActorRef, messages, prelude::Message}; use rand::{SeedableRng, rng, rngs::StdRng}; +use tracing::error; pub use crate::evm::safe_signer; @@ -62,6 +67,9 @@ pub enum Error { #[error("Integrity violation: {0}")] Integrity(#[from] integrity::Error), + + #[error("Signing error: {0}")] + Sign(#[from] SignTransactionError), } #[derive(Actor)] @@ -267,3 +275,142 @@ impl EvmActor { Ok(signer.sign_transaction_sync(&mut transaction)?) } } + +impl Message for EvmActor { + type Reply = (); + + /// Every subscriber sees every approval and acts only on the kinds it owns. + async fn handle( + &mut self, + msg: ProposalApproved, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + let result = match msg.kind { + ProposalKind::GrantWalletAccess(settings) => self.grant_wallet_access(&settings).await, + ProposalKind::ApprovePersistentGrant(settings) => { + self.create_persistent_grant(*settings).await + } + ProposalKind::ApproveOneOffTransaction(settings) => { + self.sign_one_off_transaction(msg.id, *settings).await + } + _ => return, + }; + + if let Err(error) = result { + error!( + ?error, + proposal_id = msg.id.to_raw(), + "Failed to execute an approved proposal" + ); + } + } +} + +impl EvmActor { + async fn grant_wallet_access( + &mut self, + settings: &grant_wallet_access::Settings, + ) -> Result<(), Error> { + let mut conn = self.db.get().await.map_err(DatabaseError::from)?; + + insert_into(schema::evm_wallet_access::table) + .values(( + schema::evm_wallet_access::wallet_id.eq(EvmWalletId::from_raw(settings.wallet_id)), + schema::evm_wallet_access::client_id.eq(settings.client_id), + )) + .execute(&mut conn) + .await + .map_err(DatabaseError::from)?; + + Ok(()) + } + + async fn create_persistent_grant( + &mut self, + grant: persistent_grant::Settings, + ) -> Result<(), Error> { + use crate::evm::policies::{ + TransactionRateLimit, VolumeRateLimit, ether_transfer, token_transfers, + }; + use alloy::primitives::U256; + use chrono::Duration; + + let volume = |limit: persistent_grant::VolumeLimit| VolumeRateLimit { + max_volume: U256::from_be_bytes(limit.max_volume), + window: Duration::seconds(limit.window_secs), + }; + + let basic = SharedGrantSettings { + wallet_access_id: grant.wallet_access_id, + chain: grant.chain_id, + valid_from: grant + .valid_from_secs + .and_then(|s| chrono::DateTime::from_timestamp(s, 0)), + valid_until: grant + .valid_until_secs + .and_then(|s| chrono::DateTime::from_timestamp(s, 0)), + max_gas_fee_per_gas: grant.max_gas_fee_per_gas.map(U256::from_be_bytes), + max_priority_fee_per_gas: grant.max_priority_fee_per_gas.map(U256::from_be_bytes), + rate_limit: grant.rate_limit.map(|r| TransactionRateLimit { + count: r.count, + window: Duration::seconds(r.window_secs), + }), + }; + + let specific = match grant.specific { + persistent_grant::Specific::EtherTransfer { targets, limit } => { + SpecificGrant::EtherTransfer(ether_transfer::Settings { + target: targets.into_iter().map(Address::from).collect(), + limit: volume(limit), + }) + } + persistent_grant::Specific::TokenTransfer { + token_contract, + receiver, + volume_limits, + } => SpecificGrant::TokenTransfer(token_transfers::Settings { + token_contract: Address::from(token_contract), + target: receiver.map(Address::from), + volume_limits: volume_limits.into_iter().map(volume).collect(), + }), + }; + + self.operator_create_grant(basic, specific).await?; + + Ok(()) + } + + async fn sign_one_off_transaction( + &mut self, + proposal_id: ProposalId, + tx: one_off_transaction::Settings, + ) -> Result<(), Error> { + use alloy::{ + eips::eip2930::AccessList, + primitives::{Bytes, TxKind, U256}, + }; + + let transaction = TxEip1559 { + chain_id: tx.chain_id, + nonce: tx.nonce, + gas_limit: tx.gas_limit, + max_fee_per_gas: tx.max_fee_per_gas, + max_priority_fee_per_gas: tx.max_priority_fee_per_gas, + to: TxKind::Call(Address::from(tx.to)), + value: U256::from_be_bytes(tx.value), + input: Bytes::from(tx.input), + access_list: AccessList::default(), + }; + + let signature = self + .client_sign_transaction(tx.client_id, Address::from(tx.wallet_address), transaction) + .await?; + + let mut conn = self.db.get().await.map_err(DatabaseError::from)?; + one_off_transaction::store_signature(proposal_id, &signature, &mut conn) + .await + .map_err(DatabaseError::from)?; + + Ok(()) + } +} diff --git a/server/crates/arbiter-server/src/actors/mod.rs b/server/crates/arbiter-server/src/actors/mod.rs index 0412374..812b1be 100644 --- a/server/crates/arbiter-server/src/actors/mod.rs +++ b/server/crates/arbiter-server/src/actors/mod.rs @@ -1,14 +1,21 @@ use crate::{ actors::{ - bootstrap::Bootstrapper, evm::EvmActor, flow_coordinator::FlowCoordinator, - operator_registry::OperatorRegistry, proposal_manager::ProposalManager, vault::Vault, + bootstrap::Bootstrapper, + evm::EvmActor, + flow_coordinator::FlowCoordinator, + operator_registry::OperatorRegistry, + proposal_manager::{ProposalManager, events::ProposalApproved}, + vault::Vault, vault_coordinator::VaultCoordinator, }, db, }; use kameo::actor::{ActorRef, Spawn}; -use kameo_actors::{DeliveryStrategy, message_bus::MessageBus}; +use kameo_actors::{ + DeliveryStrategy, + message_bus::{MessageBus, Register}, +}; use thiserror::Error; pub mod bootstrap; @@ -55,14 +62,18 @@ impl GlobalActors { db.clone(), key_holder.clone(), )); + // Approved proposals are executed by whoever owns the kind, not by ProposalManager. + for recipient in [ + evm.clone().recipient::(), + vault_coordinator.clone().recipient::(), + key_holder.clone().recipient::(), + ] { + let _ = message_bus.tell(Register(recipient)).await; + } + Ok(Self { bootstrapper: Bootstrapper::spawn(Bootstrapper::new(&db).await?), - proposal_manager: ProposalManager::spawn(ProposalManager::new( - db, - key_holder.clone(), - evm.clone(), - vault_coordinator.clone(), - )), + proposal_manager: ProposalManager::spawn(ProposalManager::new(db, message_bus.clone())), vault: key_holder, vault_coordinator, flow_coordinator: FlowCoordinator::spawn(FlowCoordinator::new( diff --git a/server/crates/arbiter-server/src/actors/proposal_manager.rs b/server/crates/arbiter-server/src/actors/proposal_manager.rs index ebaf81f..62b741a 100644 --- a/server/crates/arbiter-server/src/actors/proposal_manager.rs +++ b/server/crates/arbiter-server/src/actors/proposal_manager.rs @@ -1,9 +1,5 @@ use crate::{ - actors::{ - evm::EvmActor, - vault::Vault, - vault_coordinator::{StartRekey, VaultCoordinator}, - }, + actors::proposal_manager::events::ProposalApproved, crypto::governance, db::{ self, @@ -13,7 +9,7 @@ use crate::{ OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId, SqliteTimestamp, }, - proposal::{ProposalKind, ProposalKindTag, one_off_transaction, persistent_grant}, + proposal::{ProposalKind, ProposalKindTag}, schema, }, }; @@ -24,9 +20,12 @@ use diesel::{ }; use diesel_async::{AsyncConnection as _, RunQueryDsl}; use kameo::{Actor, actor::ActorRef, messages}; +use kameo_actors::message_bus::{MessageBus, Publish}; use std::collections::HashMap; use strum::IntoDiscriminant as _; -use tracing::{error, warn}; +use tracing::warn; + +pub mod events; pub const DEFAULT_TTL_SECS: u32 = 7 * 24 * 60 * 60; // 7 days pub const MAX_TTL_SECS: u32 = DEFAULT_TTL_SECS; @@ -58,8 +57,8 @@ pub enum Error { DatabaseConnection(#[from] db::PoolError), #[error("Database query error: {0}")] DatabaseQuery(#[from] diesel::result::Error), - #[error("Execution failed: {0}")] - ExecutionFailed(String), + #[error("Proposal manager is unavailable")] + Unavailable, #[error("Recovery operators are sleeping")] RecoveryNotActive, #[error("Recovery operators may only vote on operator replacement")] @@ -83,24 +82,12 @@ pub struct ProposalSummary { #[derive(Actor)] pub struct ProposalManager { pub(crate) db: db::DatabasePool, - pub(crate) vault: ActorRef, - pub(crate) evm: ActorRef, - pub(crate) vault_coordinator: ActorRef, + pub(crate) events: ActorRef, } impl ProposalManager { - pub const fn new( - db: db::DatabasePool, - vault: ActorRef, - evm: ActorRef, - vault_coordinator: ActorRef, - ) -> Self { - Self { - db, - vault, - evm, - vault_coordinator, - } + pub const fn new(db: db::DatabasePool, events: ActorRef) -> Self { + Self { db, events } } } @@ -343,12 +330,7 @@ impl ProposalManager { let threshold_i64 = threshold as i64; if approve_count >= threshold_i64 { - diesel::update(schema::proposal::table.find(proposal_id)) - .set(schema::proposal::status.eq(ProposalStatus::Approved)) - .execute(&mut conn) - .await?; - drop(conn); // release connection before async execution - self.execute_proposal(&proposal).await?; + self.announce_approval(&mut conn, &proposal).await?; return Ok(VoteOutcome::Approved); } @@ -505,12 +487,7 @@ impl ProposalManager { let approve_count = ordinary_approve + recovery_approve; if approve_count >= threshold_i64 { - diesel::update(schema::proposal::table.find(proposal_id)) - .set(schema::proposal::status.eq(ProposalStatus::Approved)) - .execute(&mut conn) - .await?; - drop(conn); - self.execute_proposal(&proposal).await?; + self.announce_approval(&mut conn, &proposal).await?; return Ok(VoteOutcome::Approved); } @@ -568,222 +545,30 @@ impl ProposalManager { .map_err(Error::from) } - async fn execute_proposal(&self, proposal: &Proposal) -> Result<(), Error> { - let mut conn = self.db.get().await?; - let kind = db::proposal::load_kind(&mut conn, proposal.id, proposal.kind).await?; - drop(conn); - - match kind { - ProposalKind::ApproveSdkClient(s) => self.execute_approve_sdk_client(s.client_id).await, - ProposalKind::GrantWalletAccess(s) => { - self.execute_grant_wallet_access(s.wallet_id, s.client_id) - .await - } - ProposalKind::ReplaceOperator(s) => { - self.execute_replace_operator(s.old_operator_id, s.new_pubkey) - .await - } - ProposalKind::TriggerRekey => self.execute_trigger_rekey().await, - ProposalKind::ApprovePersistentGrant(grant) => { - self.execute_approve_persistent_grant(*grant).await - } - ProposalKind::ApproveOneOffTransaction(tx) => { - self.execute_approve_one_off_transaction(proposal.id, *tx) - .await - } - } - } - - async fn execute_grant_wallet_access( + /// Marks the proposal approved and hands the outcome to whoever owns that kind. + /// + /// The outcome is published, not executed: this actor coordinates voting and nothing + /// else. Executors subscribe on the bus, so a vote is answered once the quorum is + /// recorded rather than once the effect has landed. + async fn announce_approval( &self, - wallet_id: i32, - client_id: i32, + conn: &mut db::DatabaseConnection, + proposal: &Proposal, ) -> Result<(), Error> { - use crate::db::models::EvmWalletId; + diesel::update(schema::proposal::table.find(proposal.id)) + .set(schema::proposal::status.eq(ProposalStatus::Approved)) + .execute(conn) + .await?; - let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?; - - diesel::insert_into(schema::evm_wallet_access::table) - .values(( - schema::evm_wallet_access::wallet_id.eq(EvmWalletId::from_raw(wallet_id)), - schema::evm_wallet_access::client_id.eq(client_id), - )) - .execute(&mut conn) - .await - .map_err(|e| Error::ExecutionFailed(format!("grant wallet access: {e}")))?; + let kind = db::proposal::load_kind(conn, proposal.id, proposal.kind).await?; + let _ = self + .events + .tell(Publish(ProposalApproved { + id: proposal.id, + kind, + })) + .await; Ok(()) } - - /// Updates the old operator's public key in-place (preserving their DB id and history), - /// removes their old Shamir share, then begins a coordinated re-key (§3.3). - async fn execute_replace_operator( - &self, - old_operator_id: OperatorIdentityId, - new_pubkey: Vec, - ) -> Result<(), Error> { - let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?; - - diesel::update(schema::operator_identity::table) - .filter(schema::operator_identity::id.eq(old_operator_id)) - .set(schema::operator_identity::public_key.eq(&new_pubkey)) - .execute(&mut conn) - .await - .map_err(|e| Error::ExecutionFailed(format!("update operator pubkey: {e}")))?; - - // Remove the old Shamir share; finalize_rekey will store a fresh one. - diesel::delete(schema::operator::table) - .filter(schema::operator::id.eq(Some(old_operator_id))) - .execute(&mut conn) - .await - .map_err(|e| Error::ExecutionFailed(format!("remove old operator share: {e}")))?; - - drop(conn); - - self.vault_coordinator - .ask(StartRekey {}) - .await - .map_err(|e| Error::ExecutionFailed(format!("start rekey: {e}")))?; - - Ok(()) - } - - /// Triggers a Shamir re-key with the current operator set (§3.3). - async fn execute_trigger_rekey(&self) -> Result<(), Error> { - self.vault_coordinator - .ask(StartRekey {}) - .await - .map_err(|e| Error::ExecutionFailed(format!("start rekey: {e}")))?; - Ok(()) - } - - async fn execute_approve_one_off_transaction( - &self, - proposal_id: ProposalId, - tx: one_off_transaction::Settings, - ) -> Result<(), Error> { - use crate::actors::evm::ClientSignTransaction; - use alloy::{ - consensus::TxEip1559, - eips::eip2930::AccessList, - primitives::{Address, Bytes, TxKind, U256}, - }; - - let transaction = TxEip1559 { - chain_id: tx.chain_id, - nonce: tx.nonce, - gas_limit: tx.gas_limit, - max_fee_per_gas: tx.max_fee_per_gas, - max_priority_fee_per_gas: tx.max_priority_fee_per_gas, - to: TxKind::Call(Address::from(tx.to)), - value: U256::from_be_bytes(tx.value), - input: Bytes::from(tx.input), - access_list: AccessList::default(), - }; - - let sig = self - .evm - .ask(ClientSignTransaction { - client_id: tx.client_id, - wallet_address: Address::from(tx.wallet_address), - transaction, - }) - .await - .map_err(|e| Error::ExecutionFailed(format!("sign one-off tx: {e}")))?; - - let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?; - one_off_transaction::store_signature(proposal_id, &sig, &mut conn) - .await - .map_err(|e| Error::ExecutionFailed(format!("store proposal result: {e}")))?; - - Ok(()) - } - - async fn execute_approve_persistent_grant( - &self, - grant: persistent_grant::Settings, - ) -> Result<(), Error> { - use crate::{ - actors::evm::OperatorCreateGrant, - evm::policies::{ - SharedGrantSettings, SpecificGrant, TransactionRateLimit, VolumeRateLimit, - ether_transfer, token_transfers, - }, - }; - use alloy::primitives::{Address, U256}; - use chrono::Duration; - - let volume = |limit: persistent_grant::VolumeLimit| VolumeRateLimit { - max_volume: U256::from_be_bytes(limit.max_volume), - window: Duration::seconds(limit.window_secs), - }; - - let basic = SharedGrantSettings { - wallet_access_id: grant.wallet_access_id, - chain: grant.chain_id, - valid_from: grant - .valid_from_secs - .and_then(|s| chrono::DateTime::from_timestamp(s, 0)), - valid_until: grant - .valid_until_secs - .and_then(|s| chrono::DateTime::from_timestamp(s, 0)), - max_gas_fee_per_gas: grant.max_gas_fee_per_gas.map(U256::from_be_bytes), - max_priority_fee_per_gas: grant.max_priority_fee_per_gas.map(U256::from_be_bytes), - rate_limit: grant.rate_limit.map(|r| TransactionRateLimit { - count: r.count, - window: Duration::seconds(r.window_secs), - }), - }; - - let grant = match grant.specific { - persistent_grant::Specific::EtherTransfer { targets, limit } => { - SpecificGrant::EtherTransfer(ether_transfer::Settings { - target: targets.into_iter().map(Address::from).collect(), - limit: volume(limit), - }) - } - persistent_grant::Specific::TokenTransfer { - token_contract, - receiver, - volume_limits, - } => SpecificGrant::TokenTransfer(token_transfers::Settings { - token_contract: Address::from(token_contract), - target: receiver.map(Address::from), - volume_limits: volume_limits.into_iter().map(volume).collect(), - }), - }; - - self.evm - .ask(OperatorCreateGrant { basic, grant }) - .await - .map_err(|e| Error::ExecutionFailed(format!("create grant: {e}")))?; - - Ok(()) - } - - async fn execute_approve_sdk_client(&self, client_id: i32) -> Result<(), Error> { - use crate::{crypto::integrity, peers::client::ClientCredentials}; - use arbiter_crypto::authn; - - let mut conn = self.db.get().await.map_err(Error::DatabaseConnection)?; - - let pubkey_bytes: Vec = schema::program_client::table - .find(client_id) - .select(schema::program_client::public_key) - .first(&mut conn) - .await - .map_err(|e| Error::ExecutionFailed(format!("client not found: {e}")))?; - - let pubkey = authn::PublicKey::try_from(pubkey_bytes.as_slice()) - .map_err(|()| Error::ExecutionFailed("invalid client public key".to_owned()))?; - - let creds = ClientCredentials { pubkey }; - - integrity::sign_entity(&mut conn, &self.vault, &creds, client_id) - .await - .map_err(|e| { - error!(?e, "Failed to sign integrity envelope for client"); - Error::ExecutionFailed(e.to_string()) - }) - } } diff --git a/server/crates/arbiter-server/src/actors/proposal_manager/events.rs b/server/crates/arbiter-server/src/actors/proposal_manager/events.rs new file mode 100644 index 0000000..2e2cfd7 --- /dev/null +++ b/server/crates/arbiter-server/src/actors/proposal_manager/events.rs @@ -0,0 +1,11 @@ +use crate::db::{models::ProposalId, proposal::ProposalKind}; + +/// Published once a proposal reaches its approval threshold. +/// +/// Executors subscribe on the global `MessageBus` and act on the kinds they own; +/// `ProposalManager` does not know who acts on an outcome, or whether anyone does. +#[derive(Debug, Clone)] +pub struct ProposalApproved { + pub id: ProposalId, + pub kind: ProposalKind, +} diff --git a/server/crates/arbiter-server/src/actors/vault/mod.rs b/server/crates/arbiter-server/src/actors/vault/mod.rs index ad9d721..14fde5d 100644 --- a/server/crates/arbiter-server/src/actors/vault/mod.rs +++ b/server/crates/arbiter-server/src/actors/vault/mod.rs @@ -1,12 +1,14 @@ use crate::{ + actors::proposal_manager::events::ProposalApproved, crypto::{ KeyCell, encryption::v1::{self, Nonce}, - integrity::v1::HmacSha256, + integrity::{self, v1::HmacSha256}, }, db::{ self, models::{self, RootKeyHistory, RootKeyHistoryId}, + proposal::ProposalKind, schema, }, }; @@ -19,7 +21,7 @@ use diesel::{ }; use diesel_async::{AsyncConnection, RunQueryDsl}; use hmac::{KeyInit as _, Mac as _}; -use kameo::{Actor, Reply, actor::ActorRef, messages}; +use kameo::{Actor, Reply, actor::ActorRef, messages, prelude::Message}; use kameo_actors::message_bus::{MessageBus, Publish}; use strum::{EnumDiscriminants, IntoDiscriminant}; use tracing::{error, info}; @@ -461,6 +463,62 @@ impl Vault { } } +impl Message for Vault { + type Reply = (); + + /// Every subscriber sees every approval and acts only on the kinds it owns. + async fn handle( + &mut self, + msg: ProposalApproved, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + let ProposalKind::ApproveSdkClient(settings) = msg.kind else { + return; + }; + + if let Err(error) = self.approve_sdk_client(settings.client_id).await { + error!( + ?error, + proposal_id = msg.id.to_raw(), + "Failed to execute an approved proposal" + ); + } + } +} + +impl Vault { + /// Attests an approved SDK client with the root key. + /// + /// Builds the envelope from its parts rather than calling `integrity::sign_entity`, + /// which would have this actor ask itself for a signature and deadlock. + async fn approve_sdk_client(&mut self, client_id: i32) -> Result<(), Error> { + use crate::peers::client::ClientCredentials; + use arbiter_crypto::authn; + + // Cloned so the connection does not hold a borrow of `self` across `sign_integrity`. + let db = self.db.clone(); + let mut conn = db.get().await?; + + let pubkey_bytes: Vec = schema::program_client::table + .find(client_id) + .select(schema::program_client::public_key) + .first(&mut conn) + .await?; + + let pubkey = + authn::PublicKey::try_from(pubkey_bytes.as_slice()).map_err(|()| Error::InvalidKey)?; + let credentials = ClientCredentials { pubkey }; + + let (entity_id, mac_input) = integrity::envelope_input(&credentials, client_id); + let (key_version, mac) = self.sign_integrity(mac_input)?; + + integrity::store_envelope::(&mut conn, entity_id, key_version, mac) + .await?; + + Ok(()) + } +} + #[cfg(test)] mod tests { use crate::actors::GlobalActors; diff --git a/server/crates/arbiter-server/src/actors/vault_coordinator/mod.rs b/server/crates/arbiter-server/src/actors/vault_coordinator/mod.rs index 1b723bd..ba403ca 100644 --- a/server/crates/arbiter-server/src/actors/vault_coordinator/mod.rs +++ b/server/crates/arbiter-server/src/actors/vault_coordinator/mod.rs @@ -3,14 +3,21 @@ use std::collections::HashMap; use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _}; use diesel::{ExpressionMethods as _, QueryDsl}; use diesel_async::RunQueryDsl; -use kameo::{Actor, actor::ActorRef, messages}; +use kameo::{Actor, actor::ActorRef, messages, prelude::Message}; use rand_core::{OsRng, RngCore as _}; use tracing::error; use crate::{ - actors::vault::{Bootstrap, RekeyRootKey, TryUnseal, Vault}, + actors::{ + proposal_manager::events::ProposalApproved, + vault::{Bootstrap, RekeyRootKey, TryUnseal, Vault}, + }, crypto::{KeyCell, derive_key, encryption::v1::Nonce, shamir, shamir::shamir_threshold}, - db::{self, models, schema}, + db::{ + self, models, + proposal::{ProposalKind, replace_operator}, + schema, + }, }; #[derive(Debug, thiserror::Error)] @@ -719,3 +726,55 @@ impl VaultCoordinator { self.do_finalize_rekey().await } } + +impl Message for VaultCoordinator { + type Reply = (); + + /// Every subscriber sees every approval and acts only on the kinds it owns. + async fn handle( + &mut self, + msg: ProposalApproved, + _ctx: &mut kameo::prelude::Context, + ) -> Self::Reply { + let result = match msg.kind { + ProposalKind::ReplaceOperator(settings) => self.replace_operator(&settings).await, + ProposalKind::TriggerRekey => self.start_rekey().await, + _ => return, + }; + + if let Err(error) = result { + error!( + ?error, + proposal_id = msg.id.to_raw(), + "Failed to execute an approved proposal" + ); + } + } +} + +impl VaultCoordinator { + /// Replaces the operator's public key in place, keeping their id and history, drops the + /// share that key no longer matches, then begins a coordinated re-key (§3.3). + async fn replace_operator( + &mut self, + settings: &replace_operator::Settings, + ) -> Result<(), Error> { + let mut conn = self.db.get().await?; + + diesel::update(schema::operator_identity::table) + .filter(schema::operator_identity::id.eq(settings.old_operator_id)) + .set(schema::operator_identity::public_key.eq(&settings.new_pubkey)) + .execute(&mut conn) + .await?; + + // Drop the stale Shamir share; finalize_rekey stores a fresh one. + diesel::delete(schema::operator::table) + .filter(schema::operator::id.eq(Some(settings.old_operator_id))) + .execute(&mut conn) + .await?; + + drop(conn); + + self.start_rekey().await + } +} diff --git a/server/crates/arbiter-server/src/crypto/integrity/v1.rs b/server/crates/arbiter-server/src/crypto/integrity/v1.rs index 9feb840..9154552 100644 --- a/server/crates/arbiter-server/src/crypto/integrity/v1.rs +++ b/server/crates/arbiter-server/src/crypto/integrity/v1.rs @@ -2,7 +2,7 @@ use crate::{ actors::vault::{self, GetState, SignIntegrity, Vault, VerifyIntegrity}, db::{ self, - models::{IntegrityEnvelope, NewIntegrityEnvelope}, + models::{IntegrityEnvelope, NewIntegrityEnvelope, RootKeyHistoryId}, schema::integrity_envelope, }, }; @@ -109,11 +109,7 @@ pub async fn sign_entity( entity: &E, entity_id: impl IntoId, ) -> Result<(), Error> { - let payload_hash = payload_hash(&entity); - - let entity_id = entity_id.into_id(); - - let mac_input = build_mac_input(E::KIND, &entity_id, E::VERSION, &payload_hash); + let (entity_id, mac_input) = envelope_input::(entity, entity_id); let (key_version, mac) = vault @@ -124,6 +120,31 @@ pub async fn sign_entity( _ => Error::VaultSend, })?; + store_envelope::(conn, entity_id, key_version, mac) + .await + .map_err(db::DatabaseError::from)?; + + Ok(()) +} + +/// The entity id and the bytes the root key covers, as a pair. +/// +/// Split out of [`sign_entity`] so the `Vault` actor can build an envelope from inside a +/// message handler, where asking itself for a signature would deadlock. +pub fn envelope_input(entity: &E, entity_id: impl IntoId) -> (Vec, Vec) { + let payload_hash = payload_hash(entity); + let entity_id = entity_id.into_id(); + let mac_input = build_mac_input(E::KIND, &entity_id, E::VERSION, &payload_hash); + (entity_id, mac_input) +} + +/// Stores the integrity envelope for one entity, replacing any envelope it already has. +pub async fn store_envelope( + conn: &mut impl AsyncConnection, + entity_id: Vec, + key_version: RootKeyHistoryId, + mac: Vec, +) -> Result<(), diesel::result::Error> { insert_into(integrity_envelope::table) .values(NewIntegrityEnvelope { entity_kind: E::KIND.to_owned(), @@ -143,8 +164,7 @@ pub async fn sign_entity( integrity_envelope::mac.eq(mac), )) .execute(conn) - .await - .map_err(db::DatabaseError::from)?; + .await?; Ok(()) } diff --git a/server/crates/arbiter-server/src/grpc/operator/governance.rs b/server/crates/arbiter-server/src/grpc/operator/governance.rs index f1f1475..145f7ee 100644 --- a/server/crates/arbiter-server/src/grpc/operator/governance.rs +++ b/server/crates/arbiter-server/src/grpc/operator/governance.rs @@ -217,6 +217,9 @@ async fn handle_vote( Err(kameo::error::SendError::HandlerError(ProposalError::ProposalNotFound)) => { return Err(Status::not_found("Proposal not found")); } + Err(kameo::error::SendError::HandlerError(ProposalError::Unavailable)) => { + return Err(Status::unavailable("Proposal manager is unavailable")); + } Err(e) => { warn!(?e, "cast_vote failed"); return Err(Status::internal("Failed to cast vote")); diff --git a/server/crates/arbiter-server/src/peers/operator/session/handlers.rs b/server/crates/arbiter-server/src/peers/operator/session/handlers.rs index 258ea2c..6fc8d59 100644 --- a/server/crates/arbiter-server/src/peers/operator/session/handlers.rs +++ b/server/crates/arbiter-server/src/peers/operator/session/handlers.rs @@ -318,7 +318,7 @@ impl OperatorSession { .await .map_err(|err| match err { SendError::HandlerError(e) => e, - _ => crate::actors::proposal_manager::Error::ExecutionFailed("actor unavailable".to_owned()), + _ => crate::actors::proposal_manager::Error::Unavailable, }) } diff --git a/server/crates/arbiter-server/tests/governance.rs b/server/crates/arbiter-server/tests/governance.rs index a25dea9..1819809 100644 --- a/server/crates/arbiter-server/tests/governance.rs +++ b/server/crates/arbiter-server/tests/governance.rs @@ -26,6 +26,26 @@ use arbiter_server::db::schema::{ }; use diesel::{ExpressionMethods, QueryDsl, insert_into}; use diesel_async::RunQueryDsl; +use std::future::Future; + +/// Retries `probe` until it yields a value, then returns it. +/// +/// Outcome execution is asynchronous: `CastVote` answers as soon as the quorum is +/// recorded, and the actor that owns the kind runs afterwards off the message bus. Tests +/// therefore wait for the effect instead of reading the database straight after the vote. +async fn eventually(what: &str, mut probe: F) -> T +where + F: FnMut() -> Fut, + Fut: Future>, +{ + for _ in 0..100 { + if let Some(value) = probe().await { + return value; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + panic!("{what} did not happen within 2s"); +} async fn register_operator(db: &db::DatabasePool, pubkey: &authn::PublicKey) -> OperatorIdentityId { let mut conn = db.get().await.unwrap(); @@ -628,14 +648,20 @@ async fn approve_sdk_client_writes_integrity_envelope() { assert_eq!(outcome, VoteOutcome::Approved); - let mut conn = db.get().await.unwrap(); - let count: i64 = integrity_envelope::table - .filter(integrity_envelope::entity_kind.eq("client_credentials")) - .count() - .get_result(&mut conn) - .await - .unwrap(); - assert_eq!(count, 1); + eventually("the client's integrity envelope", || { + let db = db.clone(); + async move { + let mut conn = db.get().await.unwrap(); + let count: i64 = integrity_envelope::table + .filter(integrity_envelope::entity_kind.eq("client_credentials")) + .count() + .get_result(&mut conn) + .await + .unwrap(); + (count == 1).then_some(()) + } + }) + .await; } #[tokio::test] @@ -682,15 +708,21 @@ async fn grant_wallet_access_on_quorum_approval() { assert_eq!(outcome, VoteOutcome::Approved); - let mut conn = db.get().await.unwrap(); - let count: i64 = evm_wallet_access::table - .filter(evm_wallet_access::wallet_id.eq(wallet_id)) - .filter(evm_wallet_access::client_id.eq(client_id)) - .count() - .get_result(&mut conn) - .await - .unwrap(); - assert_eq!(count, 1); + eventually("the wallet access row", || { + let db = db.clone(); + async move { + let mut conn = db.get().await.unwrap(); + let count: i64 = evm_wallet_access::table + .filter(evm_wallet_access::wallet_id.eq(wallet_id)) + .filter(evm_wallet_access::client_id.eq(client_id)) + .count() + .get_result(&mut conn) + .await + .unwrap(); + (count == 1).then_some(()) + } + }) + .await; } #[tokio::test] @@ -767,14 +799,20 @@ async fn approve_persistent_grant_creates_basic_grant_row() { assert_eq!(outcome, VoteOutcome::Approved); - let mut conn = db.get().await.unwrap(); - let count: i64 = evm_basic_grant::table - .filter(evm_basic_grant::wallet_access_id.eq(wallet_access_id)) - .count() - .get_result(&mut conn) - .await - .unwrap(); - assert_eq!(count, 1); + eventually("the basic grant row", || { + let db = db.clone(); + async move { + let mut conn = db.get().await.unwrap(); + let count: i64 = evm_basic_grant::table + .filter(evm_basic_grant::wallet_access_id.eq(wallet_access_id)) + .count() + .get_result(&mut conn) + .await + .unwrap(); + (count == 1).then_some(()) + } + }) + .await; } #[tokio::test] @@ -881,17 +919,23 @@ async fn approve_one_off_transaction_stores_result() { assert_eq!(outcome, VoteOutcome::Approved); - let mut conn = db.get().await.unwrap(); - let (r, s, y_parity): (Vec, Vec, i32) = proposal_one_off_transaction_result::table - .find(proposal_id) - .select(( - proposal_one_off_transaction_result::r, - proposal_one_off_transaction_result::s, - proposal_one_off_transaction_result::y_parity, - )) - .first(&mut conn) - .await - .expect("an approved transaction must leave its signature"); + let (r, s, y_parity): (Vec, Vec, i32) = eventually("the transaction signature", || { + let db = db.clone(); + async move { + let mut conn = db.get().await.unwrap(); + proposal_one_off_transaction_result::table + .find(proposal_id) + .select(( + proposal_one_off_transaction_result::r, + proposal_one_off_transaction_result::s, + proposal_one_off_transaction_result::y_parity, + )) + .first(&mut conn) + .await + .ok() + } + }) + .await; assert_eq!(r.len(), 32, "r must be a 32-byte scalar"); assert_eq!(s.len(), 32, "s must be a 32-byte scalar"); @@ -944,23 +988,30 @@ async fn replace_operator_updates_pubkey_and_starts_rekey() { assert_eq!(outcome, VoteOutcome::Approved); + eventually("the operator's public key to be replaced", || { + let db = db.clone(); + let new_pubkey = new_pubkey.clone(); + async move { + let mut conn = db.get().await.unwrap(); + let stored: Vec = operator_identity::table + .filter(operator_identity::id.eq(op_id)) + .select(operator_identity::public_key) + .first(&mut conn) + .await + .unwrap(); + (stored == new_pubkey).then_some(()) + } + }) + .await; + + // The old identity row is updated in place, so no second operator appears. let mut conn = db.get().await.unwrap(); - // The old identity row is updated in-place; count stays the same. let count: i64 = operator_identity::table .count() .get_result(&mut conn) .await .unwrap(); assert_eq!(count, 1); - - // Verify the public key was updated to the new one. - let stored_pubkey: Vec = operator_identity::table - .filter(operator_identity::id.eq(op_id)) - .select(operator_identity::public_key) - .first(&mut conn) - .await - .unwrap(); - assert_eq!(stored_pubkey, new_pubkey.clone()); } #[tokio::test]