refactor(custody): drop the single-implementation CustodyStore trait #108

Open
CleverWild wants to merge 1 commits from cleverwild/sznwsuou into feat-shamir-custody
9 changed files with 139 additions and 253 deletions

View File

@@ -3,14 +3,9 @@ use crate::{
bootstrap::Bootstrapper, evm::EvmActor, flow_coordinator::FlowCoordinator,
operator_registry::OperatorRegistry, vault::Vault, vault_coordinator::VaultCoordinator,
},
db::{
self,
custody::{CustodyStore, DieselCustodyStore},
},
db,
};
use std::sync::Arc;
use kameo::actor::{ActorRef, Spawn};
use kameo_actors::{DeliveryStrategy, message_bus::MessageBus};
use thiserror::Error;
@@ -50,12 +45,10 @@ impl GlobalActors {
pub async fn spawn(db: db::DatabasePool) -> Result<Self, SpawnError> {
let events = Self::spawn_message_bus();
let custody: Arc<dyn CustodyStore> = Arc::new(DieselCustodyStore);
let vault =
Vault::spawn(Vault::new(db.clone(), events.clone(), Arc::clone(&custody)).await?);
let vault = Vault::spawn(Vault::new(db.clone(), events.clone()).await?);
let bootstrapper = Bootstrapper::spawn(Bootstrapper::new(&db, events.clone()).await?);
let vault_coordinator =
VaultCoordinator::spawn(VaultCoordinator::new(db.clone(), vault.clone(), custody));
VaultCoordinator::spawn(VaultCoordinator::new(db.clone(), vault.clone()));
let operator_registry = OperatorRegistry::spawn(OperatorRegistry::default());
Ok(Self {
bootstrapper,

View File

@@ -6,14 +6,13 @@ use crate::{
},
db::{
self,
custody::{CustodyRecord, CustodyStore},
custody::{self, CustodyRecord},
models::{self, RootKeyHistory, RootKeyHistoryId},
schema::{self},
},
};
use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
use std::sync::Arc;
use chrono::Utc;
use diesel::{
@@ -64,7 +63,7 @@ pub enum Error {
DatabaseTransaction(#[from] diesel::result::Error),
#[error("Custody storage error: {0}")]
Custody(#[from] db::custody::Error),
Custody(#[from] custody::Error),
#[error("Broken database")]
BrokenDatabase,
@@ -101,17 +100,12 @@ pub struct Vault {
db: db::DatabasePool,
state: State,
events: ActorRef<MessageBus>,
custody: Arc<dyn CustodyStore>,
unseal_failures: u32,
}
#[messages]
impl Vault {
pub async fn new(
db: db::DatabasePool,
events: ActorRef<MessageBus>,
custody: Arc<dyn CustodyStore>,
) -> Result<Self, Error> {
pub async fn new(db: db::DatabasePool, events: ActorRef<MessageBus>) -> Result<Self, Error> {
let state = {
let mut conn = db.get().await?;
@@ -133,7 +127,6 @@ impl Vault {
db,
state,
events,
custody,
unseal_failures: 0,
})
}
@@ -213,7 +206,6 @@ impl Vault {
let mut conn = self.db.get().await?;
let data_encryption_nonce_bytes = data_encryption_nonce.to_vec();
let custody_store = Arc::clone(&self.custody);
let root_key_history_id = conn
.transaction(async |conn| {
let root_key_history_id = insert_into(schema::root_key_history::table)
@@ -235,7 +227,7 @@ impl Vault {
.await?;
if let Some(record) = custody.as_ref() {
custody_store.write_record(&mut *conn, record).await?;
custody::write_record(&mut *conn, record).await?;
}
Result::<_, Error>::Ok(RootKeyHistoryId::from_raw(root_key_history_id))
@@ -457,16 +449,12 @@ impl Vault {
#[cfg(test)]
mod tests {
use crate::{actors::GlobalActors, db::custody::DieselCustodyStore};
use crate::actors::GlobalActors;
use super::*;
async fn bootstrapped_actor(db: &db::DatabasePool) -> Vault {
let mut actor = Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap();
let seal_key = KeyCell::from([0u8; 32]);

View File

@@ -2,9 +2,7 @@
//!
//! The coordinator collects one passphrase per committee member, then hands the
//! assembled material to [`Vault`] in a single message. It owns no Diesel code:
//! everything it reads or writes goes through [`CustodyStore`].
use std::sync::Arc;
//! everything it reads or writes goes through [`db::custody`].
use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
use argon2::RECOMMENDED_SALT_LEN;
@@ -17,7 +15,7 @@ use crate::{
crypto::{KeyCell, derive_key, encryption::v1::Nonce, shamir},
db::{
self,
custody::{CustodyRecord, CustodyStore, EncryptedShare},
custody::{self, CustodyRecord, EncryptedShare},
models::OperatorId,
},
};
@@ -52,7 +50,7 @@ pub enum Error {
#[error("Database connection error: {0}")]
DatabaseConnection(#[from] db::PoolError),
#[error("Custody storage error: {0}")]
Custody(#[from] db::custody::Error),
Custody(#[from] custody::Error),
#[error("Encryption error")]
Encryption,
#[error("The vault is already bootstrapped")]
@@ -111,20 +109,14 @@ enum CoordinatorState {
pub struct VaultCoordinator {
db: db::DatabasePool,
vault: ActorRef<Vault>,
custody: Arc<dyn CustodyStore>,
state: CoordinatorState,
}
impl VaultCoordinator {
pub fn new(
db: db::DatabasePool,
vault: ActorRef<Vault>,
custody: Arc<dyn CustodyStore>,
) -> Self {
pub const fn new(db: db::DatabasePool, vault: ActorRef<Vault>) -> Self {
Self {
db,
vault,
custody,
state: CoordinatorState::Idle,
}
}
@@ -217,16 +209,13 @@ async fn finalize_bootstrap(
/// Reconstruct the seal key from the contributed passphrases and unseal.
async fn finalize_unseal(
db: &db::DatabasePool,
custody: &Arc<dyn CustodyStore>,
vault: &ActorRef<Vault>,
threshold: usize,
contributions: &mut Contributions,
) -> Result<(), Error> {
let stored = {
let mut conn = db.get().await?;
custody
.shares(&mut conn, &contributions.operators())
.await?
custody::shares(&mut conn, &contributions.operators()).await?
};
let mut plaintext = Vec::with_capacity(stored.len());
@@ -336,7 +325,7 @@ impl VaultCoordinator {
if matches!(self.state, CoordinatorState::Idle) {
let threshold = {
let mut conn = self.db.get().await?;
self.custody.threshold(&mut conn).await?
custody::threshold(&mut conn).await?
};
self.state = CoordinatorState::Unsealing {
threshold,
@@ -374,15 +363,7 @@ impl VaultCoordinator {
unreachable!("state was matched as Unsealing above")
};
match finalize_unseal(
&self.db,
&self.custody,
&self.vault,
threshold,
&mut contributions,
)
.await
{
match finalize_unseal(&self.db, &self.vault, threshold, &mut contributions).await {
Ok(()) => Ok(true),
Err(error) => {
self.state = CoordinatorState::Unsealing {

View File

@@ -206,9 +206,6 @@ pub async fn is_signing_available(vault: &ActorRef<Vault>) -> Result<bool, Error
#[cfg(test)]
mod tests {
use std::sync::Arc;
use crate::db::custody::DieselCustodyStore;
use diesel::{ExpressionMethods as _, QueryDsl};
use diesel_async::RunQueryDsl;
use kameo::{actor::ActorRef, prelude::Spawn};
@@ -234,11 +231,7 @@ mod tests {
async fn bootstrapped_vault(db: &db::DatabasePool) -> ActorRef<Vault> {
let actor = Vault::spawn(
Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap(),
);

View File

@@ -1,15 +1,13 @@
//! Storage for Shamir custody material: the reconstruction threshold and the
//! per-operator encrypted shares of the vault seal key.
//!
//! Every query lives behind [`CustodyStore`] so that the actors above it hold
//! no Diesel code of their own. [`CustodyStore::write_record`] borrows the
//! caller's connection instead of taking one from the pool, which lets the
//! vault write custody material inside the same transaction that stores the
//! root key.
//! The queries live here so that the actors above hold no Diesel code of their
//! own. Every one of them borrows the caller's connection instead of taking one
//! from the pool, which is what lets the vault write custody material inside
//! the same transaction that stores the root key.
use std::collections::HashMap;
use async_trait::async_trait;
use diesel::{ExpressionMethods as _, QueryDsl};
use diesel_async::RunQueryDsl;
@@ -46,38 +44,12 @@ pub enum Error {
MissingShare(OperatorId),
}
#[async_trait]
pub trait CustodyStore: std::fmt::Debug + Send + Sync {
/// Persist threshold and shares on the caller's connection, joining any
/// transaction the caller has already opened.
async fn write_record(
&self,
/// Persist threshold and shares on the caller's connection, joining any
/// transaction the caller has already opened.
pub async fn write_record(
conn: &mut db::DatabaseConnection,
record: &CustodyRecord,
) -> Result<(), Error>;
/// Number of shares required to reconstruct the seal key.
async fn threshold(&self, conn: &mut db::DatabaseConnection) -> Result<usize, Error>;
/// Load the shares of `operators` in one query, in the order requested.
async fn shares(
&self,
conn: &mut db::DatabaseConnection,
operators: &[OperatorId],
) -> Result<Vec<EncryptedShare>, Error>;
}
/// The production [`CustodyStore`], backed by the `SQLite` schema.
#[derive(Debug, Clone, Copy, Default)]
pub struct DieselCustodyStore;
#[async_trait]
impl CustodyStore for DieselCustodyStore {
async fn write_record(
&self,
conn: &mut db::DatabaseConnection,
record: &CustodyRecord,
) -> Result<(), Error> {
) -> Result<(), Error> {
let threshold = i32::try_from(record.threshold).map_err(|_| Error::BrokenThreshold)?;
// SQLite has no batch form for REPLACE INTO in diesel, so the rows go
@@ -106,9 +78,10 @@ impl CustodyStore for DieselCustodyStore {
}
Ok(())
}
}
async fn threshold(&self, conn: &mut db::DatabaseConnection) -> Result<usize, Error> {
/// Number of shares required to reconstruct the seal key.
pub async fn threshold(conn: &mut db::DatabaseConnection) -> Result<usize, Error> {
let stored: Option<i32> = schema::arbiter_settings::table
.select(schema::arbiter_settings::shamir_threshold)
.first(conn)
@@ -118,16 +91,19 @@ impl CustodyStore for DieselCustodyStore {
.and_then(|value| usize::try_from(value).ok())
.filter(|threshold| *threshold > 0)
.ok_or(Error::BrokenThreshold)
}
}
async fn shares(
&self,
/// Load the shares of `operators` in one query, in the order requested.
pub async fn shares(
conn: &mut db::DatabaseConnection,
operators: &[OperatorId],
) -> Result<Vec<EncryptedShare>, Error> {
) -> Result<Vec<EncryptedShare>, Error> {
/// (id, then the three columns that make up [`EncryptedShare`])
type ShareRow = (Option<OperatorId>, Vec<u8>, Vec<u8>, Vec<u8>);
let wanted: Vec<Option<OperatorId>> = operators.iter().copied().map(Some).collect();
let rows: Vec<(Option<OperatorId>, Vec<u8>, Vec<u8>, Vec<u8>)> = schema::operator::table
let rows: Vec<ShareRow> = schema::operator::table
.filter(schema::operator::id.eq_any(wanted))
.select((
schema::operator::id,
@@ -158,5 +134,4 @@ impl CustodyStore for DieselCustodyStore {
.iter()
.map(|id| found.remove(id).ok_or(Error::MissingShare(*id)))
.collect()
}
}

View File

@@ -501,9 +501,6 @@ impl Engine {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use crate::db::custody::DieselCustodyStore;
use alloy::primitives::{Address, Bytes, U256, address};
use chrono::{Duration, Utc};
use diesel::{SelectableHelper, insert_into};
@@ -769,11 +766,7 @@ mod tests {
async fn bootstrapped_vault(db: &db::DatabasePool) -> ActorRef<Vault> {
let actor = Vault::spawn(
Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap(),
);

View File

@@ -7,21 +7,16 @@ use arbiter_proto::transport::{Bi, Error, Receiver, Sender};
use arbiter_server::{
actors::{GlobalActors, vault::Vault},
crypto::KeyCell,
db::{self, custody::DieselCustodyStore, schema},
db::{self, schema},
};
use async_trait::async_trait;
use diesel::QueryDsl;
use diesel_async::RunQueryDsl;
use std::sync::Arc;
use tokio::sync::mpsc;
pub(crate) async fn bootstrapped_vault(db: &db::DatabasePool) -> Vault {
let mut actor = Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap();
actor

View File

@@ -6,16 +6,13 @@ use arbiter_server::{
vault::{CreateNew, Error, Vault},
},
crypto::KeyCell,
db::{self, custody::DieselCustodyStore, models, schema},
db::{self, models, schema},
};
use diesel::{ExpressionMethods as _, QueryDsl, SelectableHelper, dsl::sql_query};
use diesel_async::RunQueryDsl;
use kameo::actor::{ActorRef, Spawn as _};
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use std::collections::{HashMap, HashSet};
use tokio::task::JoinSet;
const TEST_AAD: &[u8] = b"test-aad";
@@ -169,11 +166,7 @@ async fn decrypt_roundtrip_after_high_concurrency() {
let writes = write_concurrently(actor, "roundtrip", 40).await;
let expected: HashMap<i32, Vec<u8>> = writes.into_iter().collect();
let mut decryptor = Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut decryptor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap();
decryptor

View File

@@ -9,12 +9,11 @@ use arbiter_server::{
KeyCell,
encryption::v1::{Nonce, ROOT_KEY_TAG},
},
db::{self, custody::DieselCustodyStore, models, schema},
db::{self, models, schema},
};
use diesel::{QueryDsl, SelectableHelper};
use diesel_async::RunQueryDsl;
use std::sync::Arc;
const TEST_AAD: &[u8] = b"test-aad";
@@ -22,11 +21,7 @@ const TEST_AAD: &[u8] = b"test-aad";
#[test_log::test]
async fn bootstrap() {
let db = db::create_test_pool().await;
let mut actor = Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap();
@@ -62,11 +57,7 @@ async fn bootstrap_rejects_double() {
#[test_log::test]
async fn create_new_before_bootstrap_fails() {
let db = db::create_test_pool().await;
let mut actor = Vault::new(
db,
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db, GlobalActors::spawn_message_bus())
.await
.unwrap();
@@ -81,11 +72,7 @@ async fn create_new_before_bootstrap_fails() {
#[test_log::test]
async fn decrypt_before_bootstrap_fails() {
let db = db::create_test_pool().await;
let mut actor = Vault::new(
db,
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db, GlobalActors::spawn_message_bus())
.await
.unwrap();
@@ -100,11 +87,7 @@ async fn new_restores_sealed_state() {
let actor = common::bootstrapped_vault(&db).await;
drop(actor);
let mut actor2 = Vault::new(
db,
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor2 = Vault::new(db, GlobalActors::spawn_message_bus())
.await
.unwrap();
let err = actor2.decrypt(1, TEST_AAD.to_vec()).await.unwrap_err();
@@ -124,11 +107,7 @@ async fn unseal_correct_password() {
.unwrap();
drop(actor);
let mut actor = Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap();
let seal_key = KeyCell::from([0u8; 32]);
@@ -151,11 +130,7 @@ async fn unseal_wrong_then_correct_password() {
.unwrap();
drop(actor);
let mut actor = Vault::new(
db.clone(),
GlobalActors::spawn_message_bus(),
Arc::new(DieselCustodyStore),
)
let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
.await
.unwrap();