Merge pull request 'security(server): bind grant revocation state (revoked_at) to integrity hash' (#83) from security-hash-revoke_at into main

Reviewed-on: #83

server/Cargo.toml

@@ -6,7 +6,7 @@ resolver = "3"
 
 
 [workspace.dependencies]
-alloy = "2.0.0"
+alloy = "2.0.4"
 async-trait = "0.1.89"
 base64 = "0.22.1"
 chrono = { version = "0.4.44", features = ["serde"] }
@@ -16,15 +16,15 @@ kameo = {git = "https://github.com/hdbg/kameo.git", rev = "805b417"}
 kameo_actors = {git = "https://github.com/hdbg/kameo.git", rev = "805b417"}
 hmac = "0.13.0"
 miette = { version = "7.6.0", features = ["fancy", "serde"] }
-ml-dsa = { version = "0.1.0-rc.8", features = ["zeroize"] }
+ml-dsa = { version = "0.1.0-rc.9", features = ["zeroize"] }
 mutants = "0.0.4"
 prost = "0.14.3"
 prost-types = { version = "0.14.3", features = ["chrono"] }
 rand = "0.10.1"
 rcgen = { version = "0.14.7", features = [ "aws_lc_rs",  "pem", "x509-parser", "zeroize" ], default-features = false }
 rstest = "0.26.1"
-rustls = { version = "0.23.38", features = ["aws-lc-rs", "logging", "prefer-post-quantum", "std"], default-features = false }
+rustls = { version = "0.23.40", features = ["aws-lc-rs", "logging", "prefer-post-quantum", "std"], default-features = false }
-rustls-pki-types = "1.14.0"
+rustls-pki-types = "1.14.1"
 sha2 = "0.11"
 smlang = "0.8.0"
 thiserror = "2.0.18"
@@ -76,6 +76,7 @@ needless_pass_by_ref_mut = "allow"
 pub_underscore_fields = "allow"
 redundant_pub_crate = "allow" 
 uninhabited_references = "allow" # safe with unsafe_code = "forbid" and standard uninhabited pattern (match *self {})
+too-many-lines = "allow" # this is a very common pattern in server code, and it's not always possible to break it down into smaller modules without hurting readability
 
 # restriction lints
 alloc_instead_of_core = "warn"

server/crates/arbiter-client/Cargo.toml

@@ -21,7 +21,7 @@ tokio.workspace = true
 tokio-stream.workspace = true
 thiserror.workspace = true
 http = "1.4.0"
-rustls-webpki = { version = "0.103.12", features = ["aws-lc-rs"] }
+rustls-webpki = { version = "0.103.13", features = ["aws-lc-rs"] }
 async-trait.workspace = true
 chrono.workspace = true
 

server/crates/arbiter-client/src/auth.rs

@@ -100,7 +100,7 @@ async fn send_auth_challenge_solution(
     key: &SigningKey,
     challenge: AuthChallenge,
 ) -> Result<(), AuthError> {
-    let timestamp = DateTime::from_timestamp_nanos(challenge.timestamp_nanos as i64);
+    let timestamp = DateTime::from_timestamp_nanos(challenge.timestamp_nanos.cast_signed());
     let challenge = authn::AuthChallenge {
         nonce: *challenge
             .random

server/crates/arbiter-server/Cargo.toml

@@ -9,8 +9,8 @@ license = "Apache-2.0"
 workspace = true
 
 [dependencies]
-diesel = { version = "2.3.7", features = ["chrono", "returning_clauses_for_sqlite_3_35", "serde_json", "time", "uuid"] }
+diesel = { version = "2.3.9", features = ["chrono", "returning_clauses_for_sqlite_3_35", "serde_json", "time", "uuid"] }
-diesel-async = { version = "0.8.0", features = [
+diesel-async = { version = "0.9.0", features = [
     "bb8",
     "migrations",
     "sqlite",
@@ -27,7 +27,7 @@ tokio.workspace = true
 rustls.workspace = true
 smlang.workspace = true
 thiserror.workspace = true
-diesel_migrations = { version = "2.3.1", features = ["sqlite"] }
+diesel_migrations = { version = "2.3.2", features = ["sqlite"] }
 async-trait.workspace = true
 tokio-stream.workspace = true
 rand.workspace = true

server/crates/arbiter-server/src/actors/evm/mod.rs

@@ -160,29 +160,14 @@ impl EvmActor {
     }
 
     #[message]
-    #[expect(clippy::unused_async, reason = "reserved for impl")]
+    pub async fn useragent_delete_grant(
-    pub async fn operator_delete_grant(&mut self, _grant_id: i32) -> Result<(), Error> {
+        &mut self,
-        // let mut conn = self.db.get().await.map_err(DatabaseError::from)?;
+        grant_id: i32,
-        // let vault = self.vault.clone();
+    ) -> Result<(), Error> {
-
+        self.engine
-        // diesel_async::AsyncConnection::transaction(&mut conn, |conn| {
+            .revoke_grant(grant_id)
-        //     Box::pin(async move {
+            .await
-        //         diesel::update(schema::evm_basic_grant::table)
+            .map_err(Error::from)
-        //             .filter(schema::evm_basic_grant::id.eq(grant_id))
-        //             .set(schema::evm_basic_grant::revoked_at.eq(SqliteTimestamp::now()))
-        //             .execute(conn)
-        //             .await?;
-
-        //         let signed = integrity::evm::load_signed_grant_by_basic_id(conn, grant_id).await?;
-
-        //         diesel::result::QueryResult::Ok(())
-        //     })
-        // })
-        // .await
-        // .map_err(DatabaseError::from)?;
-
-        // Ok(())
-        todo!()
     }
 
     #[message]

server/crates/arbiter-server/src/actors/vault/mod.rs

@@ -119,31 +119,29 @@ impl Vault {
         let mut conn = pool.get().await?;
 
         let nonce = conn
-            .exclusive_transaction(|conn| {
+            .exclusive_transaction(async |conn| {
-                Box::pin(async move {
+                let current_nonce: Vec<u8> = schema::root_key_history::table
-                    let current_nonce: Vec<u8> = schema::root_key_history::table
+                    .filter(schema::root_key_history::id.eq(root_key_id))
-                        .filter(schema::root_key_history::id.eq(root_key_id))
+                    .select(schema::root_key_history::data_encryption_nonce)
-                        .select(schema::root_key_history::data_encryption_nonce)
+                    .first(&mut *conn)
-                        .first(conn)
+                    .await?;
-                        .await?;
 
-                    let mut nonce = Nonce::try_from(current_nonce.as_slice()).map_err(|()| {
+                let mut nonce = Nonce::try_from(current_nonce.as_slice()).map_err(|()| {
-                        error!(
+                    error!(
-                            "Broken database: invalid nonce for root key history id={}",
+                        "Broken database: invalid nonce for root key history id={}",
-                            root_key_id
+                        root_key_id
-                        );
+                    );
-                        Error::BrokenDatabase
+                    Error::BrokenDatabase
-                    })?;
+                })?;
-                    nonce.increment();
+                nonce.increment();
 
-                    update(schema::root_key_history::table)
+                update(schema::root_key_history::table)
-                        .filter(schema::root_key_history::id.eq(root_key_id))
+                    .filter(schema::root_key_history::id.eq(root_key_id))
-                        .set(schema::root_key_history::data_encryption_nonce.eq(nonce.to_vec()))
+                    .set(schema::root_key_history::data_encryption_nonce.eq(nonce.to_vec()))
-                        .execute(conn)
+                    .execute(&mut *conn)
-                        .await?;
+                    .await?;
 
-                    Result::<_, Error>::Ok(nonce)
+                Result::<_, Error>::Ok(nonce)
-                })
             })
             .await?;
 
@@ -185,28 +183,26 @@ impl Vault {
 
         let data_encryption_nonce_bytes = data_encryption_nonce.to_vec();
         let root_key_history_id = conn
-            .transaction(|conn| {
+            .transaction(async |conn| {
-                Box::pin(async move {
+                let root_key_history_id: i32 = insert_into(schema::root_key_history::table)
-                    let root_key_history_id: i32 = insert_into(schema::root_key_history::table)
+                    .values(&models::NewRootKeyHistory {
-                        .values(&models::NewRootKeyHistory {
+                        ciphertext: root_key_ciphertext.clone(),
-                            ciphertext: root_key_ciphertext,
+                        tag: v1::ROOT_KEY_TAG.to_vec(),
-                            tag: v1::ROOT_KEY_TAG.to_vec(),
+                        root_key_encryption_nonce: root_key_nonce.to_vec(),
-                            root_key_encryption_nonce: root_key_nonce.to_vec(),
+                        data_encryption_nonce: data_encryption_nonce_bytes.clone(),
-                            data_encryption_nonce: data_encryption_nonce_bytes,
+                        schema_version: 1,
-                            schema_version: 1,
+                        salt: salt.to_vec(),
-                            salt: salt.to_vec(),
+                    })
-                        })
+                    .returning(schema::root_key_history::id)
-                        .returning(schema::root_key_history::id)
+                    .get_result(&mut *conn)
-                        .get_result(conn)
+                    .await?;
-                        .await?;
 
-                    update(schema::arbiter_settings::table)
+                update(schema::arbiter_settings::table)
-                        .set(schema::arbiter_settings::root_key_id.eq(root_key_history_id))
+                    .set(schema::arbiter_settings::root_key_id.eq(root_key_history_id))
-                        .execute(conn)
+                    .execute(&mut *conn)
-                        .await?;
+                    .await?;
 
-                    Result::<_, diesel::result::Error>::Ok(root_key_history_id)
+                Result::<_, diesel::result::Error>::Ok(root_key_history_id)
-                })
             })
             .await?;
 
@@ -350,12 +346,10 @@ impl Vault {
             root_key_history_id,
         } = Self::expect_unsealed(&mut self.state)?;
 
-        let mut hmac = root_key
+        let mut hmac = root_key.0.read_inline(|k| {
-            .0
+            HmacSha256::new_from_slice(k)
-            .read_inline(|k| match HmacSha256::new_from_slice(k) {
+                .unwrap_or_else(|_| unreachable!("HMAC accepts keys of any size"))
-                Ok(v) => v,
+        });
-                Err(_) => unreachable!("HMAC accepts keys of any size"),
-            });
         hmac.update(&root_key_history_id.to_be_bytes());
         hmac.update(&mac_input);
 
@@ -379,12 +373,10 @@ impl Vault {
             return Ok(false);
         }
 
-        let mut hmac = root_key
+        let mut hmac = root_key.0.read_inline(|k| {
-            .0
+            HmacSha256::new_from_slice(k)
-            .read_inline(|k| match HmacSha256::new_from_slice(k) {
+                .unwrap_or_else(|_| unreachable!("HMAC accepts keys of any size"))
-                Ok(v) => v,
+        });
-                Err(_) => unreachable!("HMAC accepts keys of any size"),
-            });
         hmac.update(&key_version.to_be_bytes());
         hmac.update(&mac_input);
 
@@ -427,12 +419,13 @@ mod tests {
     async fn nonce_monotonic_even_when_nonce_allocation_interleaves() {
         let db = db::create_test_pool().await;
         let mut actor = bootstrapped_actor(&db).await;
-        let root_key_history_id = match actor.state {
+
-            State::Unsealed(Unsealed {
+        let State::Unsealed(Unsealed {
-                root_key_history_id,
+            root_key_history_id,
-                ..
+            ..
-            }) => root_key_history_id,
+        }) = actor.state
-            _ => panic!("expected unsealed state"),
+        else {
+            panic!("expected unsealed state")
         };
 
         let n1 = Vault::get_new_nonce(&db, root_key_history_id)
@@ -444,8 +437,8 @@ mod tests {
         assert!(n2.to_vec() > n1.to_vec(), "nonce must increase");
 
         let mut conn = db.get().await.unwrap();
-        let root_row: models::RootKeyHistory = schema::root_key_history::table
+        let root_row: RootKeyHistory = schema::root_key_history::table
-            .select(models::RootKeyHistory::as_select())
+            .select(RootKeyHistory::as_select())
             .first(&mut conn)
             .await
             .unwrap();

server/crates/arbiter-server/src/context/tls.rs

@@ -174,28 +174,26 @@ impl TlsManager {
 
         {
             let mut conn = db.get().await?;
-            conn.transaction(|conn| {
+            conn.transaction(async |conn| {
-                Box::pin(async {
+                let new_tls_history = NewTlsHistory {
-                    let new_tls_history = NewTlsHistory {
+                    cert: new_cert.cert.pem(),
-                        cert: new_cert.cert.pem(),
+                    cert_key: new_cert.cert_key.serialize_pem(),
-                        cert_key: new_cert.cert_key.serialize_pem(),
+                    ca_cert: encode_cert_to_pem(&ca.cert),
-                        ca_cert: encode_cert_to_pem(&ca.cert),
+                    ca_key: ca.issuer.key().serialize_pem(),
-                        ca_key: ca.issuer.key().serialize_pem(),
+                };
-                    };
 
-                    let inserted_tls_history: i32 = diesel::insert_into(tls_history::table)
+                let inserted_tls_history: i32 = diesel::insert_into(tls_history::table)
-                        .values(&new_tls_history)
+                    .values(&new_tls_history)
-                        .returning(tls_history::id)
+                    .returning(tls_history::id)
-                        .get_result(conn)
+                    .get_result(&mut *conn)
-                        .await?;
+                    .await?;
 
-                    diesel::update(arbiter_settings::table)
+                diesel::update(arbiter_settings::table)
-                        .set(arbiter_settings::tls_id.eq(inserted_tls_history))
+                    .set(arbiter_settings::tls_id.eq(inserted_tls_history))
-                        .execute(conn)
+                    .execute(&mut *conn)
-                        .await?;
+                    .await?;
 
-                    Result::<_, diesel::result::Error>::Ok(())
+                Result::<_, diesel::result::Error>::Ok(())
-                })
             })
             .await?;
         }

server/crates/arbiter-server/src/evm/mod.rs

@@ -1,28 +1,34 @@
+use diesel_async::{AsyncConnection, RunQueryDsl};
+use kameo::actor::ActorRef;
+
 use crate::{
     actors::vault::Vault,
     crypto::integrity,
     db::{
         self, DatabaseError,
         models::{
-            EvmBasicGrant, EvmWalletAccess, NewEvmBasicGrant, NewEvmTransactionLog, SqliteTimestamp,
+            EvmBasicGrant, EvmEtherTransferGrant, EvmEtherTransferGrantTarget,
+            EvmEtherTransferLimit, EvmTokenTransferGrant, EvmTokenTransferVolumeLimit,
+            EvmWalletAccess, NewEvmBasicGrant, NewEvmTransactionLog, SqliteTimestamp,
         },
         schema::{self, evm_transaction_log},
     },
     evm::policies::{
         CombinedSettings, DatabaseID, EvalContext, EvalViolation, Grant, Policy,
-        SharedGrantSettings, SpecificGrant, SpecificMeaning, ether_transfer::EtherTransfer,
+        SharedGrantSettings, SpecificGrant, SpecificMeaning, VolumeRateLimit,
-        token_transfers::TokenTransfer,
+        ether_transfer::EtherTransfer, token_transfers::TokenTransfer,
     },
 };
 
 use alloy::{
     consensus::TxEip1559,
-    primitives::{TxKind, U256},
+    primitives::{Address, TxKind, U256},
 };
 use chrono::Utc;
-use diesel::{ExpressionMethods as _, QueryDsl as _, QueryResult, insert_into, sqlite::Sqlite};
+use diesel::{
-use diesel_async::{AsyncConnection, RunQueryDsl};
+    ExpressionMethods as _, OptionalExtension, QueryDsl as _, QueryResult, SelectableHelper,
-use kameo::actor::ActorRef;
+    insert_into, sqlite::Sqlite, update,
+};
 
 pub mod abi;
 pub mod safe_signer;
@@ -179,24 +185,22 @@ impl Engine {
         }
 
         if run_kind == RunKind::Execution {
-            conn.transaction(|conn| {
+            conn.transaction(async |conn| {
-                Box::pin(async move {
+                let log_id: i32 = insert_into(evm_transaction_log::table)
-                    let log_id: i32 = insert_into(evm_transaction_log::table)
+                    .values(&NewEvmTransactionLog {
-                        .values(&NewEvmTransactionLog {
+                        grant_id: grant.common_settings_id,
-                            grant_id: grant.common_settings_id,
+                        wallet_access_id: context.target.id,
-                            wallet_access_id: context.target.id,
+                        chain_id: context.chain.into(),
-                            chain_id: context.chain.into(),
+                        eth_value: utils::u256_to_bytes(context.value).to_vec(),
-                            eth_value: utils::u256_to_bytes(context.value).to_vec(),
+                        signed_at: Utc::now().into(),
-                            signed_at: Utc::now().into(),
+                    })
-                        })
+                    .returning(evm_transaction_log::id)
-                        .returning(evm_transaction_log::id)
+                    .get_result(&mut *conn)
-                        .get_result(conn)
+                    .await?;
-                        .await?;
 
-                    P::record_transaction(&context, meaning, log_id, &grant, conn).await?;
+                P::record_transaction(&context, meaning, log_id, &grant, &mut *conn).await?;
 
-                    QueryResult::Ok(())
+                QueryResult::Ok(())
-                })
             })
             .await
             .map_err(DatabaseError::from)?;
@@ -222,60 +226,203 @@ impl Engine {
         let vault = self.vault.clone();
 
         let id = conn
-            .transaction(|conn| {
+            .transaction(async |conn| {
-                Box::pin(async move {
+                use schema::evm_basic_grant;
-                    use schema::evm_basic_grant;
 
-                    #[expect(
+                #[expect(
-                        clippy::cast_possible_truncation,
+                    clippy::cast_possible_truncation,
-                        clippy::cast_possible_wrap,
+                    clippy::cast_possible_wrap,
-                        clippy::as_conversions,
+                    clippy::as_conversions,
-                        reason = "fixme! #86"
+                    reason = "fixme! #86"
-                    )]
+                )]
-                    let basic_grant: EvmBasicGrant = insert_into(evm_basic_grant::table)
+                let basic_grant: EvmBasicGrant = insert_into(evm_basic_grant::table)
-                        .values(&NewEvmBasicGrant {
+                    .values(&NewEvmBasicGrant {
-                            chain_id: full_grant.shared.chain.into(),
+                        chain_id: full_grant.shared.chain.into(),
-                            wallet_access_id: full_grant.shared.wallet_access_id,
+                        wallet_access_id: full_grant.shared.wallet_access_id,
-                            valid_from: full_grant.shared.valid_from.map(SqliteTimestamp),
+                        valid_from: full_grant.shared.valid_from.map(SqliteTimestamp),
-                            valid_until: full_grant.shared.valid_until.map(SqliteTimestamp),
+                        valid_until: full_grant.shared.valid_until.map(SqliteTimestamp),
-                            max_gas_fee_per_gas: full_grant
+                        max_gas_fee_per_gas: full_grant
-                                .shared
+                            .shared
-                                .max_gas_fee_per_gas
+                            .max_gas_fee_per_gas
-                                .map(|fee| utils::u256_to_bytes(fee).to_vec()),
+                            .map(|fee| utils::u256_to_bytes(fee).to_vec()),
-                            max_priority_fee_per_gas: full_grant
+                        max_priority_fee_per_gas: full_grant
-                                .shared
+                            .shared
-                                .max_priority_fee_per_gas
+                            .max_priority_fee_per_gas
-                                .map(|fee| utils::u256_to_bytes(fee).to_vec()),
+                            .map(|fee| utils::u256_to_bytes(fee).to_vec()),
-                            rate_limit_count: full_grant
+                        rate_limit_count: full_grant
-                                .shared
+                            .shared
-                                .rate_limit
+                            .rate_limit
-                                .as_ref()
+                            .as_ref()
-                                .map(|rl| rl.count as i32),
+                            .map(|rl| rl.count as i32),
-                            rate_limit_window_secs: full_grant
+                        rate_limit_window_secs: full_grant
-                                .shared
+                            .shared
-                                .rate_limit
+                            .rate_limit
-                                .as_ref()
+                            .as_ref()
-                                .map(|rl| rl.window.num_seconds() as i32),
+                            .map(|rl| rl.window.num_seconds() as i32),
-                            revoked_at: None,
+                        revoked_at: None,
-                        })
+                    })
-                        .returning(evm_basic_grant::all_columns)
+                    .returning(evm_basic_grant::all_columns)
-                        .get_result(conn)
+                    .get_result(&mut *conn)
-                        .await?;
+                    .await?;
 
-                    P::create_grant(&basic_grant, &full_grant.specific, conn).await?;
+                P::create_grant(&basic_grant, &full_grant.specific, &mut *conn).await?;
 
-                    integrity::sign_entity(conn, &vault, &full_grant, basic_grant.id)
+                integrity::sign_entity(&mut *conn, &vault, &full_grant, basic_grant.id)
-                        .await
+                    .await
-                        .map_err(|_| diesel::result::Error::RollbackTransaction)?;
+                    .map_err(|_| diesel::result::Error::RollbackTransaction)?;
 
-                    QueryResult::Ok(basic_grant.id)
+                QueryResult::Ok(basic_grant.id)
-                })
             })
             .await?;
 
         Ok(id)
     }
 
+    pub async fn revoke_grant(
+        &self,
+        basic_grant_id: i32,
+    ) -> Result<(), DatabaseError> {
+        let mut conn = self.db.get().await.map_err(DatabaseError::from)?;
+        let vault = self.vault.clone();
+
+        conn.transaction(async move |conn| {
+            use crate::db::schema::{
+                evm_basic_grant, evm_ether_transfer_grant, evm_ether_transfer_grant_target,
+                evm_ether_transfer_limit, evm_token_transfer_grant,
+                evm_token_transfer_volume_limit,
+            };
+
+            update(evm_basic_grant::table)
+                .filter(evm_basic_grant::id.eq(basic_grant_id))
+                .set(evm_basic_grant::revoked_at.eq(SqliteTimestamp(Utc::now())))
+                .execute(&mut *conn)
+                .await?;
+
+            let basic_grant: EvmBasicGrant = evm_basic_grant::table
+                .filter(evm_basic_grant::id.eq(basic_grant_id))
+                .select(EvmBasicGrant::as_select())
+                .first(&mut *conn)
+                .await?;
+
+            let shared = SharedGrantSettings::try_from_model(basic_grant)?;
+
+            if let Some(ether_grant) = evm_ether_transfer_grant::table
+                .filter(evm_ether_transfer_grant::basic_grant_id.eq(basic_grant_id))
+                .select(EvmEtherTransferGrant::as_select())
+                .first(&mut *conn)
+                .await
+                .optional()?
+            {
+                let target_rows: Vec<EvmEtherTransferGrantTarget> =
+                    evm_ether_transfer_grant_target::table
+                        .filter(evm_ether_transfer_grant_target::grant_id.eq(ether_grant.id))
+                        .select(EvmEtherTransferGrantTarget::as_select())
+                        .load(&mut *conn)
+                        .await?;
+                let targets: Vec<Address> = target_rows
+                    .into_iter()
+                    .filter_map(|target| {
+                        let arr: [u8; 20] = target.address.try_into().ok()?;
+                        Some(Address::from(arr))
+                    })
+                    .collect();
+
+                let limit: EvmEtherTransferLimit = evm_ether_transfer_limit::table
+                    .filter(evm_ether_transfer_limit::id.eq(ether_grant.limit_id))
+                    .select(EvmEtherTransferLimit::as_select())
+                    .first(&mut *conn)
+                    .await?;
+
+                let settings = CombinedSettings {
+                    shared: shared.clone(),
+                    specific: policies::ether_transfer::Settings {
+                        target: targets,
+                        limit: VolumeRateLimit {
+                            max_volume: utils::try_bytes_to_u256(&limit.max_volume).map_err(
+                                |err| {
+                                    diesel::result::Error::DeserializationError(Box::new(err))
+                                },
+                            )?,
+                            window: chrono::Duration::seconds(limit.window_secs.into()),
+                        },
+                    },
+                };
+
+                integrity::sign_entity(&mut *conn, &vault, &settings, basic_grant_id)
+                    .await
+                    .map_err(|_| diesel::result::Error::RollbackTransaction)?;
+
+                return QueryResult::Ok(());
+            }
+
+            if let Some(token_grant) = evm_token_transfer_grant::table
+                .filter(evm_token_transfer_grant::basic_grant_id.eq(basic_grant_id))
+                .select(EvmTokenTransferGrant::as_select())
+                .first(&mut *conn)
+                .await
+                .optional()?
+            {
+                let volume_limit_rows: Vec<EvmTokenTransferVolumeLimit> =
+                    evm_token_transfer_volume_limit::table
+                        .filter(evm_token_transfer_volume_limit::grant_id.eq(token_grant.id))
+                        .select(EvmTokenTransferVolumeLimit::as_select())
+                        .load(&mut *conn)
+                        .await?;
+                let volume_limits: Vec<VolumeRateLimit> = volume_limit_rows
+                    .into_iter()
+                    .map(|row| {
+                        Ok(VolumeRateLimit {
+                            max_volume: utils::try_bytes_to_u256(&row.max_volume).map_err(
+                                |err| {
+                                    diesel::result::Error::DeserializationError(Box::new(err))
+                                },
+                            )?,
+                            window: chrono::Duration::seconds(row.window_secs.into()),
+                        })
+                    })
+                    .collect::<QueryResult<Vec<_>>>()?;
+
+                let target: Option<Address> = match token_grant.receiver {
+                    None => None,
+                    Some(bytes) => {
+                        let arr: [u8; 20] = bytes.try_into().map_err(|_| {
+                            diesel::result::Error::DeserializationError(
+                                "Invalid receiver address length".into(),
+                            )
+                        })?;
+                        Some(Address::from(arr))
+                    }
+                };
+
+                let token_contract: [u8; 20] =
+                    token_grant.token_contract.clone().try_into().map_err(|_| {
+                        diesel::result::Error::DeserializationError(
+                            "Invalid token contract address length".into(),
+                        )
+                    })?;
+
+                let settings = CombinedSettings {
+                    shared,
+                    specific: policies::token_transfers::Settings {
+                        token_contract: Address::from(token_contract),
+                        target,
+                        volume_limits,
+                    },
+                };
+
+                integrity::sign_entity(&mut *conn, &vault, &settings, basic_grant_id)
+                    .await
+                    .map_err(|_| diesel::result::Error::RollbackTransaction)?;
+
+                return QueryResult::Ok(());
+            }
+
+            Err(diesel::result::Error::NotFound)
+        })
+        .await
+        .map_err(DatabaseError::from)
+    }
+
     async fn list_one_kind<Kind: Policy, Y>(
         &self,
         conn: &mut impl AsyncConnection<Backend = Sqlite>,
@@ -355,11 +502,15 @@ impl Engine {
 #[cfg(test)]
 mod tests {
     use alloy::primitives::{Address, Bytes, U256, address};
+    use arbiter_crypto::safecell::{SafeCell, SafeCellHandle as _};
     use chrono::{Duration, Utc};
     use diesel::{SelectableHelper, insert_into};
     use diesel_async::RunQueryDsl;
+    use kameo::{actor::ActorRef, prelude::Spawn};
     use rstest::rstest;
 
+    use crate::actors::{GlobalActors, vault::{Bootstrap, Vault}};
+    use crate::crypto::integrity;
     use crate::db::{
         self, DatabaseConnection,
         models::{
@@ -367,8 +518,10 @@ mod tests {
         },
         schema::{evm_basic_grant, evm_transaction_log},
     };
+    use crate::evm::policies::ether_transfer::EtherTransfer;
     use crate::evm::policies::{
-        EvalContext, EvalViolation, SharedGrantSettings, TransactionRateLimit,
+        CombinedSettings, EvalContext, EvalViolation, Policy, SharedGrantSettings,
+        TransactionRateLimit, VolumeRateLimit,
     };
 
     use super::check_shared_constraints;
@@ -400,6 +553,7 @@ mod tests {
             chain: CHAIN_ID,
             valid_from: None,
             valid_until: None,
+            revoked_at: None,
             max_gas_fee_per_gas: None,
             max_priority_fee_per_gas: None,
             rate_limit: None,
@@ -608,4 +762,115 @@ mod tests {
             assert!(violations.is_empty());
         }
     }
+
+    async fn bootstrapped_vault(db: &db::DatabasePool) -> ActorRef<Vault> {
+        let actor = Vault::spawn(
+            Vault::new(db.clone(), GlobalActors::spawn_message_bus())
+                .await
+                .unwrap(),
+        );
+        actor
+            .ask(Bootstrap {
+                seal_key_raw: SafeCell::new(b"integrity-test-seal-key".to_vec()),
+            })
+            .await
+            .unwrap();
+        actor
+    }
+
+    #[tokio::test]
+    async fn revoke_grant_preserves_revoked_integrity() {
+        use crate::db::schema::evm_basic_grant;
+        use diesel::ExpressionMethods as _;
+
+        let db = db::create_test_pool().await;
+        let vault = bootstrapped_vault(&db).await;
+        let engine = super::Engine::new(db.clone(), vault.clone());
+
+        let full_grant = CombinedSettings {
+            shared: SharedGrantSettings {
+                wallet_access_id: WALLET_ACCESS_ID,
+                chain: CHAIN_ID,
+                valid_from: None,
+                valid_until: None,
+                revoked_at: None,
+                max_gas_fee_per_gas: None,
+                max_priority_fee_per_gas: None,
+                rate_limit: None,
+            },
+            specific: super::policies::ether_transfer::Settings {
+                target: vec![RECIPIENT],
+                limit: VolumeRateLimit {
+                    max_volume: U256::from(100u64),
+                    window: Duration::hours(1),
+                },
+            },
+        };
+
+        let grant_id = engine
+            .create_grant::<EtherTransfer>(full_grant)
+            .await
+            .unwrap();
+
+        engine.revoke_grant(grant_id).await.unwrap();
+
+        let mut conn = db.get().await.unwrap();
+        diesel::update(evm_basic_grant::table)
+            .filter(evm_basic_grant::id.eq(grant_id))
+            .set(evm_basic_grant::revoked_at.eq::<Option<SqliteTimestamp>>(None))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+
+        let wallet_access = EvmWalletAccess {
+            id: WALLET_ACCESS_ID,
+            wallet_id: 10,
+            client_id: 20,
+            created_at: SqliteTimestamp(Utc::now()),
+        };
+        let context = EvalContext {
+            target: wallet_access,
+            chain: CHAIN_ID,
+            to: RECIPIENT,
+            value: U256::ONE,
+            calldata: Bytes::new(),
+            max_fee_per_gas: 1,
+            max_priority_fee_per_gas: 1,
+        };
+
+        let grant = EtherTransfer::try_find_grant(
+            &context, &mut conn,
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+        let result =
+            integrity::verify_entity(&mut conn, &vault, &grant.settings, grant.id).await;
+
+        assert!(matches!(
+            result,
+            Err(integrity::Error::MacMismatch { .. })
+        ));
+    }
+
+    #[test]
+    fn shared_settings_hash_changes_when_revoked_at_changes() {
+        use arbiter_crypto::hashing::Hashable;
+        use sha2::Digest;
+
+        let active = shared_settings();
+        let revoked = SharedGrantSettings {
+            revoked_at: Some(Utc::now()),
+            ..shared_settings()
+        };
+
+        let mut active_hash = sha2::Sha256::new();
+        active.hash(&mut active_hash);
+
+        let mut revoked_hash = sha2::Sha256::new();
+        revoked.hash(&mut revoked_hash);
+
+        assert_ne!(active_hash.finalize(), revoked_hash.finalize());
+    }
 }

server/crates/arbiter-server/src/evm/policies.rs

@@ -144,6 +144,7 @@ pub struct SharedGrantSettings {
 
     pub valid_from: Option<DateTime<Utc>>,
     pub valid_until: Option<DateTime<Utc>>,
+    pub revoked_at: Option<DateTime<Utc>>,
 
     pub max_gas_fee_per_gas: Option<U256>,
     pub max_priority_fee_per_gas: Option<U256>,
@@ -158,6 +159,7 @@ impl SharedGrantSettings {
             chain: model.chain_id.into(),
             valid_from: model.valid_from.map(Into::into),
             valid_until: model.valid_until.map(Into::into),
+            revoked_at: model.revoked_at.map(Into::into),
             max_gas_fee_per_gas: model
                 .max_gas_fee_per_gas
                 .map(|b| utils::try_bytes_to_u256(&b))

server/crates/arbiter-server/src/evm/policies/ether_transfer/tests.rs

@@ -79,6 +79,7 @@ fn shared() -> SharedGrantSettings {
         chain: CHAIN_ID,
         valid_from: None,
         valid_until: None,
+        revoked_at: None,
         max_gas_fee_per_gas: None,
         max_priority_fee_per_gas: None,
         rate_limit: None,

server/crates/arbiter-server/src/evm/policies/token_transfers/tests.rs

@@ -98,6 +98,7 @@ fn shared() -> SharedGrantSettings {
         chain: CHAIN_ID,
         valid_from: None,
         valid_until: None,
+        revoked_at: None,
         max_gas_fee_per_gas: None,
         max_priority_fee_per_gas: None,
         rate_limit: None,

server/crates/arbiter-server/src/grpc/client/auth.rs

@@ -200,7 +200,7 @@ impl Convert for auth::Outbound {
                         .timestamp
                         .timestamp_nanos_opt()
                         .expect("timestamp within range")
-                        as u64,
+                        .cast_unsigned(),
                     random: challenge.nonce.to_vec(),
                 })
             }

server/crates/arbiter-server/src/grpc/operator/auth.rs

@@ -80,7 +80,7 @@ impl Sender<Result<auth::Outbound, auth::Error>> for AuthTransportAdapter<'_> {
                         .timestamp
                         .timestamp_nanos_opt()
                         .expect("timestamp within range")
-                        as u64,
+                        .cast_unsigned(),
                     random: challenge.nonce.to_vec(),
                 })
             }

server/crates/arbiter-server/src/grpc/operator/inbound.rs

@@ -87,6 +87,7 @@ impl TryConvert for ProtoSharedSettings {
                 .valid_until
                 .map(ProtoTimestamp::try_convert)
                 .transpose()?,
+            revoked_at: None,
             max_gas_fee_per_gas: self
                 .max_gas_fee_per_gas
                 .as_deref()

server/crates/arbiter-server/src/peers/client/auth.rs

@@ -171,46 +171,42 @@ async fn insert_client(
         Error::DatabasePoolUnavailable
     })?;
 
-    conn.exclusive_transaction(|conn| {
+    conn.exclusive_transaction(async |conn| {
-        let vault = vault.clone();
+        let metadata_id = insert_into(client_metadata::table)
-        let pubkey = pubkey.clone();
+            .values((
-        Box::pin(async move {
+                client_metadata::name.eq(&metadata.name),
-            let metadata_id = insert_into(client_metadata::table)
+                client_metadata::description.eq(&metadata.description),
-                .values((
+                client_metadata::version.eq(&metadata.version),
-                    client_metadata::name.eq(&metadata.name),
+            ))
-                    client_metadata::description.eq(&metadata.description),
+            .returning(client_metadata::id)
-                    client_metadata::version.eq(&metadata.version),
+            .get_result::<i32>(&mut *conn)
-                ))
+            .await?;
-                .returning(client_metadata::id)
-                .get_result::<i32>(conn)
-                .await?;
 
-            let client_id = insert_into(program_client::table)
+        let client_id = insert_into(program_client::table)
-                .values((
+            .values((
-                    program_client::public_key.eq(pubkey.to_bytes()),
+                program_client::public_key.eq(pubkey.to_bytes()),
-                    program_client::metadata_id.eq(metadata_id),
+                program_client::metadata_id.eq(metadata_id),
-                ))
+            ))
-                .on_conflict_do_nothing()
+            .on_conflict_do_nothing()
-                .returning(program_client::id)
+            .returning(program_client::id)
-                .get_result::<i32>(conn)
+            .get_result::<i32>(&mut *conn)
-                .await?;
+            .await?;
 
-            integrity::sign_entity(
+        integrity::sign_entity(
-                conn,
+            &mut *conn,
-                &vault,
+            vault,
-                &ClientCredentials {
+            &ClientCredentials {
-                    pubkey: pubkey.clone(),
+                pubkey: pubkey.clone(),
-                },
+            },
-                client_id,
+            client_id,
-            )
+        )
-            .await
+        .await
-            .map_err(|e| {
+        .map_err(|e| {
-                error!(error = ?e, "Failed to sign integrity tag for new client key");
+            error!(error = ?e, "Failed to sign integrity tag for new client key");
-                Error::DatabaseOperationFailed
+            Error::DatabaseOperationFailed
-            })?;
+        })?;
 
-            Ok(client_id)
+        Ok(client_id)
-        })
     })
     .await
 }
@@ -229,55 +225,51 @@ async fn sync_client_metadata(
         Error::DatabasePoolUnavailable
     })?;
 
-    conn.exclusive_transaction(|conn| {
+    conn.exclusive_transaction(async |conn| {
-        let metadata = metadata.clone();
+        let (current_metadata_id, current): (i32, ProgramClientMetadata) = program_client::table
-        Box::pin(async move {
+            .find(client_id)
-            let (current_metadata_id, current): (i32, ProgramClientMetadata) =
+            .inner_join(client_metadata::table)
-                program_client::table
+            .select((
-                    .find(client_id)
+                program_client::metadata_id,
-                    .inner_join(client_metadata::table)
+                ProgramClientMetadata::as_select(),
-                    .select((
+            ))
-                        program_client::metadata_id,
+            .first(&mut *conn)
-                        ProgramClientMetadata::as_select(),
+            .await?;
-                    ))
-                    .first(conn)
-                    .await?;
 
-            let unchanged = current.name == metadata.name
+        let unchanged = current.name == metadata.name
-                && current.description == metadata.description
+            && current.description == metadata.description
-                && current.version == metadata.version;
+            && current.version == metadata.version;
-            if unchanged {
+        if unchanged {
-                return Ok(());
+            return Ok(());
-            }
+        }
 
-            insert_into(client_metadata_history::table)
+        insert_into(client_metadata_history::table)
-                .values((
+            .values((
-                    client_metadata_history::metadata_id.eq(current_metadata_id),
+                client_metadata_history::metadata_id.eq(current_metadata_id),
-                    client_metadata_history::client_id.eq(client_id),
+                client_metadata_history::client_id.eq(client_id),
-                ))
+            ))
-                .execute(conn)
+            .execute(&mut *conn)
-                .await?;
+            .await?;
 
-            let metadata_id = insert_into(client_metadata::table)
+        let metadata_id = insert_into(client_metadata::table)
-                .values((
+            .values((
-                    client_metadata::name.eq(&metadata.name),
+                client_metadata::name.eq(&metadata.name),
-                    client_metadata::description.eq(&metadata.description),
+                client_metadata::description.eq(&metadata.description),
-                    client_metadata::version.eq(&metadata.version),
+                client_metadata::version.eq(&metadata.version),
-                ))
+            ))
-                .returning(client_metadata::id)
+            .returning(client_metadata::id)
-                .get_result::<i32>(conn)
+            .get_result::<i32>(&mut *conn)
-                .await?;
+            .await?;
 
-            update(program_client::table.find(client_id))
+        update(program_client::table.find(client_id))
-                .set((
+            .set((
-                    program_client::metadata_id.eq(metadata_id),
+                program_client::metadata_id.eq(metadata_id),
-                    program_client::updated_at.eq(now),
+                program_client::updated_at.eq(now),
-                ))
+            ))
-                .execute(conn)
+            .execute(&mut *conn)
-                .await?;
+            .await?;
 
-            Ok::<(), diesel::result::Error>(())
+        Ok::<(), diesel::result::Error>(())
-        })
     })
     .await
     .map_err(|e| {
@@ -306,7 +298,7 @@ where
 
     let signature = expect_message(transport, |req: Inbound| match req {
         Inbound::AuthChallengeSolution { signature } => Some(signature),
-        _ => None,
+        Inbound::AuthChallengeRequest { .. } => None,
     })
     .await
     .map_err(|e| {

server/crates/arbiter-server/src/peers/operator/auth/state.rs

@@ -19,7 +19,7 @@ pub(super) struct ChallengeRequest {
     pub(super) bootstrap_token: Option<String>,
 }
 
-pub(super) struct ChallengeContext {
+pub struct ChallengeContext {
     pub(super) challenge: AuthChallenge,
     pub(super) pubkey: authn::PublicKey,
     pub(super) bootstrap_token: Option<String>,
@@ -127,8 +127,6 @@ where
         })
     }
 
-    #[allow(missing_docs)]
-    #[allow(clippy::unused_unit)]
     async fn verify_solution(
         &mut self,
         ChallengeContext {

server/crates/arbiter-server/src/peers/operator/session/handlers.rs

@@ -1,8 +1,8 @@
 use super::{Error, OperatorSession};
 use crate::{
     actors::evm::{
-        ClientSignTransaction, Generate, ListWallets, SignTransactionError as EvmSignError,
+        ClientSignTransaction, Generate, ListWallets, OperatorCreateGrant, OperatorListGrants,
-        OperatorCreateGrant, OperatorListGrants,
+        SignTransactionError as EvmSignError,
     },
     actors::flow_coordinator::client_connect_approval::ClientApprovalAnswer,
     actors::vault::VaultState,
@@ -169,20 +169,18 @@ impl OperatorSession {
         entries: Vec<NewEvmWalletAccess>,
     ) -> Result<(), Error> {
         let mut conn = self.props.db.get().await?;
-        conn.transaction(|conn| {
+        conn.transaction(async |conn| {
-            Box::pin(async move {
+            use crate::db::schema::evm_wallet_access;
-                use crate::db::schema::evm_wallet_access;
 
-                for entry in entries {
+            for entry in entries {
-                    diesel::insert_into(evm_wallet_access::table)
+                diesel::insert_into(evm_wallet_access::table)
-                        .values(&entry)
+                    .values(&entry)
-                        .on_conflict_do_nothing()
+                    .on_conflict_do_nothing()
-                        .execute(conn)
+                    .execute(&mut *conn)
-                        .await?;
+                    .await?;
-                }
+            }
 
-                Result::<_, Error>::Ok(())
+            Result::<_, Error>::Ok(())
-            })
         })
         .await?;
         Ok(())
@@ -194,18 +192,16 @@ impl OperatorSession {
         entries: Vec<i32>,
     ) -> Result<(), Error> {
         let mut conn = self.props.db.get().await?;
-        conn.transaction(|conn| {
+        conn.transaction(async |conn| {
-            Box::pin(async move {
+            use crate::db::schema::evm_wallet_access;
-                use crate::db::schema::evm_wallet_access;
+            for entry in entries {
-                for entry in entries {
+                diesel::delete(evm_wallet_access::table)
-                    diesel::delete(evm_wallet_access::table)
+                    .filter(evm_wallet_access::wallet_id.eq(entry))
-                        .filter(evm_wallet_access::wallet_id.eq(entry))
+                    .execute(&mut *conn)
-                        .execute(conn)
+                    .await?;
-                        .await?;
+            }
-                }
 
-                Result::<_, Error>::Ok(())
+            Result::<_, Error>::Ok(())
-            })
         })
         .await?;
         Ok(())
@@ -216,8 +212,7 @@ impl OperatorSession {
         &mut self,
     ) -> Result<Vec<EvmWalletAccess>, Error> {
         let mut conn = self.props.db.get().await?;
-        use crate::db::schema::evm_wallet_access;
+        let access_entries = crate::db::schema::evm_wallet_access::table
-        let access_entries = evm_wallet_access::table
             .select(EvmWalletAccess::as_select())
             .load::<_>(&mut conn)
             .await?;

server/crates/arbiter-server/src/peers/operator/session/mod.rs

@@ -63,7 +63,7 @@ impl OperatorSession {
         Self {
             props,
             sender,
-            pending_client_approvals: Default::default(),
+            pending_client_approvals: HashMap::default(),
         }
     }
 }

server/crates/arbiter-server/tests/operator/auth.rs

@@ -400,7 +400,7 @@ pub async fn challenge_auth_rejects_integrity_tag_mismatch_when_unsealed() {
     let challenge = match response {
         Ok(resp) => match resp {
             auth::Outbound::AuthChallenge { challenge } => challenge,
-            other => panic!("Expected AuthChallenge, got {other:?}"),
+            other @ auth::Outbound::AuthSuccess => panic!("Expected AuthChallenge, got {other:?}"),
         },
         Err(err) => panic!("Expected Ok response, got Err({err:?})"),
     };

server/crates/arbiter-server/tests/vault/lifecycle.rs

@@ -14,7 +14,7 @@ use diesel_async::RunQueryDsl;
 
 #[tokio::test]
 #[test_log::test]
-async fn test_bootstrap() {
+async fn bootstrap() {
     let db = db::create_test_pool().await;
     let mut actor = Vault::new(db.clone(), GlobalActors::spawn_message_bus())
         .await
@@ -39,7 +39,7 @@ async fn test_bootstrap() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_bootstrap_rejects_double() {
+async fn bootstrap_rejects_double() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 
@@ -50,7 +50,7 @@ async fn test_bootstrap_rejects_double() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_create_new_before_bootstrap_fails() {
+async fn create_new_before_bootstrap_fails() {
     let db = db::create_test_pool().await;
     let mut actor = Vault::new(db, GlobalActors::spawn_message_bus())
         .await
@@ -65,7 +65,7 @@ async fn test_create_new_before_bootstrap_fails() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_decrypt_before_bootstrap_fails() {
+async fn decrypt_before_bootstrap_fails() {
     let db = db::create_test_pool().await;
     let mut actor = Vault::new(db, GlobalActors::spawn_message_bus())
         .await
@@ -77,7 +77,7 @@ async fn test_decrypt_before_bootstrap_fails() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_new_restores_sealed_state() {
+async fn new_restores_sealed_state() {
     let db = db::create_test_pool().await;
     let actor = common::bootstrapped_vault(&db).await;
     drop(actor);
@@ -91,7 +91,7 @@ async fn test_new_restores_sealed_state() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_unseal_correct_password() {
+async fn unseal_correct_password() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 
@@ -114,7 +114,7 @@ async fn test_unseal_correct_password() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_unseal_wrong_then_correct_password() {
+async fn unseal_wrong_then_correct_password() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 

server/crates/arbiter-server/tests/vault/storage.rs

@@ -12,7 +12,7 @@ use std::collections::HashSet;
 
 #[tokio::test]
 #[test_log::test]
-async fn test_create_decrypt_roundtrip() {
+async fn create_decrypt_roundtrip() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 
@@ -28,7 +28,7 @@ async fn test_create_decrypt_roundtrip() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_decrypt_nonexistent_returns_not_found() {
+async fn decrypt_nonexistent_returns_not_found() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 
@@ -38,7 +38,7 @@ async fn test_decrypt_nonexistent_returns_not_found() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_ciphertext_differs_across_entries() {
+async fn ciphertext_differs_across_entries() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;
 
@@ -76,7 +76,7 @@ async fn test_ciphertext_differs_across_entries() {
 
 #[tokio::test]
 #[test_log::test]
-async fn test_nonce_never_reused() {
+async fn nonce_never_reused() {
     let db = db::create_test_pool().await;
     let mut actor = common::bootstrapped_vault(&db).await;