feat(server): integrity envelope engine for EVM grants with HMAC verification

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

@@ -0,0 +1,259 @@
+use alloy::primitives::Address;
+use arbiter_proto::proto::integrity::{
+    IntegrityEtherTransferSettings, IntegrityEvmGrantPayloadV1, IntegritySharedGrantSettings,
+    IntegritySpecificGrant, IntegrityTokenTransferSettings, IntegrityTransactionRateLimit,
+    IntegrityVolumeRateLimit, integrity_specific_grant,
+};
+use chrono::{DateTime, Utc};
+use diesel::sqlite::Sqlite;
+use diesel::{ExpressionMethods as _, OptionalExtension as _, QueryDsl, SelectableHelper as _};
+use diesel_async::{AsyncConnection, RunQueryDsl};
+use prost::Message;
+use prost_types::Timestamp;
+
+use crate::{
+    db::{models, schema},
+    evm::policies::{Grant, SharedGrantSettings, SpecificGrant, VolumeRateLimit},
+    integrity::IntegrityEntity,
+};
+
+pub const EVM_GRANT_ENTITY_KIND: &str = "evm_grant";
+
+#[derive(Debug, Clone)]
+pub struct SignedEvmGrant {
+    pub basic_grant_id: i32,
+    pub shared: SharedGrantSettings,
+    pub specific: SpecificGrant,
+    pub revoked_at: Option<DateTime<Utc>>,
+}
+
+impl SignedEvmGrant {
+    pub fn from_active_grant(grant: &Grant<SpecificGrant>) -> Self {
+        Self {
+            basic_grant_id: grant.shared_grant_id,
+            shared: grant.shared.clone(),
+            specific: grant.settings.clone(),
+            revoked_at: None,
+        }
+    }
+}
+
+fn timestamp(value: DateTime<Utc>) -> Timestamp {
+    Timestamp {
+        seconds: value.timestamp(),
+        nanos: 0,
+    }
+}
+
+fn encode_shared(shared: &SharedGrantSettings) -> IntegritySharedGrantSettings {
+    IntegritySharedGrantSettings {
+        wallet_access_id: shared.wallet_access_id,
+        chain_id: shared.chain,
+        valid_from: shared.valid_from.map(timestamp),
+        valid_until: shared.valid_until.map(timestamp),
+        max_gas_fee_per_gas: shared
+            .max_gas_fee_per_gas
+            .map(|v| v.to_le_bytes::<32>().to_vec()),
+        max_priority_fee_per_gas: shared
+            .max_priority_fee_per_gas
+            .map(|v| v.to_le_bytes::<32>().to_vec()),
+        rate_limit: shared
+            .rate_limit
+            .as_ref()
+            .map(|rl| IntegrityTransactionRateLimit {
+                count: rl.count,
+                window_secs: rl.window.num_seconds(),
+            }),
+    }
+}
+
+fn encode_volume_limit(limit: &VolumeRateLimit) -> IntegrityVolumeRateLimit {
+    IntegrityVolumeRateLimit {
+        max_volume: limit.max_volume.to_le_bytes::<32>().to_vec(),
+        window_secs: limit.window.num_seconds(),
+    }
+}
+
+fn try_bytes_to_u256(bytes: &[u8]) -> diesel::result::QueryResult<alloy::primitives::U256> {
+    let bytes: [u8; 32] = bytes.try_into().map_err(|_| {
+        diesel::result::Error::DeserializationError(
+            format!("Expected 32-byte U256 payload, got {}", bytes.len()).into(),
+        )
+    })?;
+    Ok(alloy::primitives::U256::from_le_bytes(bytes))
+}
+
+fn encode_specific(specific: &SpecificGrant) -> IntegritySpecificGrant {
+    let grant = match specific {
+        SpecificGrant::EtherTransfer(settings) => {
+            let mut targets: Vec<Vec<u8>> =
+                settings.target.iter().map(|addr| addr.to_vec()).collect();
+            targets.sort_unstable();
+
+            integrity_specific_grant::Grant::EtherTransfer(IntegrityEtherTransferSettings {
+                targets,
+                limit: Some(encode_volume_limit(&settings.limit)),
+            })
+        }
+        SpecificGrant::TokenTransfer(settings) => {
+            let mut volume_limits: Vec<IntegrityVolumeRateLimit> = settings
+                .volume_limits
+                .iter()
+                .map(encode_volume_limit)
+                .collect();
+            volume_limits.sort_by(|left, right| {
+                left.window_secs
+                    .cmp(&right.window_secs)
+                    .then_with(|| left.max_volume.cmp(&right.max_volume))
+            });
+
+            integrity_specific_grant::Grant::TokenTransfer(IntegrityTokenTransferSettings {
+                token_contract: settings.token_contract.to_vec(),
+                target: settings.target.map(|a| a.to_vec()),
+                volume_limits,
+            })
+        }
+    };
+
+    IntegritySpecificGrant { grant: Some(grant) }
+}
+
+impl IntegrityEntity for SignedEvmGrant {
+    fn entity_kind(&self) -> &'static str {
+        EVM_GRANT_ENTITY_KIND
+    }
+
+    fn entity_id_bytes(&self) -> Vec<u8> {
+        self.basic_grant_id.to_be_bytes().to_vec()
+    }
+
+    fn payload_version(&self) -> i32 {
+        1
+    }
+
+    fn canonical_payload_bytes(&self) -> Vec<u8> {
+        IntegrityEvmGrantPayloadV1 {
+            basic_grant_id: self.basic_grant_id,
+            shared: Some(encode_shared(&self.shared)),
+            specific: Some(encode_specific(&self.specific)),
+            revoked_at: self.revoked_at.map(timestamp),
+        }
+        .encode_to_vec()
+    }
+}
+
+pub async fn load_signed_grant_by_basic_id(
+    conn: &mut impl AsyncConnection<Backend = Sqlite>,
+    basic_grant_id: i32,
+) -> diesel::result::QueryResult<SignedEvmGrant> {
+    let basic: models::EvmBasicGrant = schema::evm_basic_grant::table
+        .filter(schema::evm_basic_grant::id.eq(basic_grant_id))
+        .select(models::EvmBasicGrant::as_select())
+        .first(conn)
+        .await?;
+
+    let specific_token: Option<models::EvmTokenTransferGrant> =
+        schema::evm_token_transfer_grant::table
+            .filter(schema::evm_token_transfer_grant::basic_grant_id.eq(basic_grant_id))
+            .select(models::EvmTokenTransferGrant::as_select())
+            .first(conn)
+            .await
+            .optional()?;
+
+    let revoked_at = basic.revoked_at.clone().map(Into::into);
+    let shared = SharedGrantSettings::try_from_model(basic)?;
+
+    if let Some(token) = specific_token {
+        let limits: Vec<models::EvmTokenTransferVolumeLimit> =
+            schema::evm_token_transfer_volume_limit::table
+                .filter(schema::evm_token_transfer_volume_limit::grant_id.eq(token.id))
+                .select(models::EvmTokenTransferVolumeLimit::as_select())
+                .load(conn)
+                .await?;
+
+        let token_contract: [u8; 20] = token.token_contract.try_into().map_err(|_| {
+            diesel::result::Error::DeserializationError(
+                "Invalid token contract address length".into(),
+            )
+        })?;
+
+        let target = match token.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 volume_limits = limits
+            .into_iter()
+            .map(|row| {
+                Ok(VolumeRateLimit {
+                    max_volume: try_bytes_to_u256(&row.max_volume)?,
+                    window: chrono::Duration::seconds(row.window_secs as i64),
+                })
+            })
+            .collect::<diesel::result::QueryResult<Vec<_>>>()?;
+
+        return Ok(SignedEvmGrant {
+            basic_grant_id,
+            shared,
+            specific: SpecificGrant::TokenTransfer(
+                crate::evm::policies::token_transfers::Settings {
+                    token_contract: Address::from(token_contract),
+                    target,
+                    volume_limits,
+                },
+            ),
+            revoked_at,
+        });
+    }
+
+    let ether: models::EvmEtherTransferGrant = schema::evm_ether_transfer_grant::table
+        .filter(schema::evm_ether_transfer_grant::basic_grant_id.eq(basic_grant_id))
+        .select(models::EvmEtherTransferGrant::as_select())
+        .first(conn)
+        .await?;
+
+    let targets_rows: Vec<models::EvmEtherTransferGrantTarget> =
+        schema::evm_ether_transfer_grant_target::table
+            .filter(schema::evm_ether_transfer_grant_target::grant_id.eq(ether.id))
+            .select(models::EvmEtherTransferGrantTarget::as_select())
+            .load(conn)
+            .await?;
+
+    let limit: models::EvmEtherTransferLimit = schema::evm_ether_transfer_limit::table
+        .filter(schema::evm_ether_transfer_limit::id.eq(ether.limit_id))
+        .select(models::EvmEtherTransferLimit::as_select())
+        .first(conn)
+        .await?;
+
+    let targets = targets_rows
+        .into_iter()
+        .map(|row| {
+            let arr: [u8; 20] = row.address.try_into().map_err(|_| {
+                diesel::result::Error::DeserializationError(
+                    "Invalid ether target address length".into(),
+                )
+            })?;
+            Ok(Address::from(arr))
+        })
+        .collect::<diesel::result::QueryResult<Vec<_>>>()?;
+
+    Ok(SignedEvmGrant {
+        basic_grant_id,
+        shared,
+        specific: SpecificGrant::EtherTransfer(crate::evm::policies::ether_transfer::Settings {
+            target: targets,
+            limit: VolumeRateLimit {
+                max_volume: try_bytes_to_u256(&limit.max_volume)?,
+                window: chrono::Duration::seconds(limit.window_secs as i64),
+            },
+        }),
+        revoked_at,
+    })
+}

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

@@ -0,0 +1,307 @@
+use diesel::{ExpressionMethods as _, QueryDsl, dsl::insert_into, sqlite::Sqlite};
+use diesel_async::{AsyncConnection, RunQueryDsl};
+use kameo::actor::ActorRef;
+use sha2::{Digest as _, Sha256};
+
+use crate::{
+    actors::keyholder::{KeyHolder, SignIntegrity, VerifyIntegrity},
+    db::{
+        self,
+        models::{IntegrityEnvelope, NewIntegrityEnvelope},
+        schema::integrity_envelope,
+    },
+};
+
+pub const CURRENT_PAYLOAD_VERSION: i32 = 1;
+
+pub mod evm;
+
+pub trait IntegrityEntity {
+    fn entity_kind(&self) -> &'static str;
+    fn entity_id_bytes(&self) -> Vec<u8>;
+    fn payload_version(&self) -> i32;
+    fn canonical_payload_bytes(&self) -> Vec<u8>;
+}
+
+#[derive(Debug, thiserror::Error, miette::Diagnostic)]
+pub enum Error {
+    #[error("Database error: {0}")]
+    #[diagnostic(code(arbiter::integrity::database))]
+    Database(#[from] db::DatabaseError),
+
+    #[error("KeyHolder error: {0}")]
+    #[diagnostic(code(arbiter::integrity::keyholder))]
+    Keyholder(#[from] crate::actors::keyholder::Error),
+
+    #[error("KeyHolder mailbox error")]
+    #[diagnostic(code(arbiter::integrity::keyholder_send))]
+    KeyholderSend,
+
+    #[error("Integrity envelope is missing for entity {entity_kind}")]
+    #[diagnostic(code(arbiter::integrity::missing_envelope))]
+    MissingEnvelope { entity_kind: &'static str },
+
+    #[error(
+        "Integrity payload version mismatch for entity {entity_kind}: expected {expected}, found {found}"
+    )]
+    #[diagnostic(code(arbiter::integrity::payload_version_mismatch))]
+    PayloadVersionMismatch {
+        entity_kind: &'static str,
+        expected: i32,
+        found: i32,
+    },
+
+    #[error("Integrity MAC mismatch for entity {entity_kind}")]
+    #[diagnostic(code(arbiter::integrity::mac_mismatch))]
+    MacMismatch { entity_kind: &'static str },
+}
+
+fn payload_hash(payload: &[u8]) -> [u8; 32] {
+    Sha256::digest(payload).into()
+}
+
+fn push_len_prefixed(out: &mut Vec<u8>, bytes: &[u8]) {
+    out.extend_from_slice(&(bytes.len() as u32).to_be_bytes());
+    out.extend_from_slice(bytes);
+}
+
+fn build_mac_input(
+    entity_kind: &str,
+    entity_id: &[u8],
+    payload_version: i32,
+    payload_hash: &[u8; 32],
+) -> Vec<u8> {
+    let mut out = Vec::with_capacity(8 + entity_kind.len() + entity_id.len() + 32);
+    push_len_prefixed(&mut out, entity_kind.as_bytes());
+    push_len_prefixed(&mut out, entity_id);
+    out.extend_from_slice(&payload_version.to_be_bytes());
+    out.extend_from_slice(payload_hash);
+    out
+}
+
+pub async fn sign_entity(
+    conn: &mut impl AsyncConnection<Backend = Sqlite>,
+    keyholder: &ActorRef<KeyHolder>,
+    entity: &impl IntegrityEntity,
+) -> Result<(), Error> {
+    let entity_kind = entity.entity_kind();
+    let entity_id = entity.entity_id_bytes();
+    let payload_version = entity.payload_version();
+    let payload = entity.canonical_payload_bytes();
+    let payload_hash = payload_hash(&payload);
+    let mac_input = build_mac_input(entity_kind, &entity_id, payload_version, &payload_hash);
+
+    let (key_version, mac) = keyholder
+        .ask(SignIntegrity { mac_input })
+        .await
+        .map_err(|err| match err {
+            kameo::error::SendError::HandlerError(inner) => Error::Keyholder(inner),
+            _ => Error::KeyholderSend,
+        })?;
+
+    diesel::delete(integrity_envelope::table)
+        .filter(integrity_envelope::entity_kind.eq(entity_kind))
+        .filter(integrity_envelope::entity_id.eq(&entity_id))
+        .execute(conn)
+        .await
+        .map_err(db::DatabaseError::from)?;
+
+    insert_into(integrity_envelope::table)
+        .values(NewIntegrityEnvelope {
+            entity_kind: entity_kind.to_string(),
+            entity_id,
+            payload_version,
+            key_version,
+            mac,
+        })
+        .execute(conn)
+        .await
+        .map_err(db::DatabaseError::from)?;
+
+    Ok(())
+}
+
+pub async fn verify_entity(
+    conn: &mut impl AsyncConnection<Backend = Sqlite>,
+    keyholder: &ActorRef<KeyHolder>,
+    entity: &impl IntegrityEntity,
+) -> Result<(), Error> {
+    let entity_kind = entity.entity_kind();
+    let entity_id = entity.entity_id_bytes();
+    let expected_payload_version = entity.payload_version();
+
+    let envelope: IntegrityEnvelope = integrity_envelope::table
+        .filter(integrity_envelope::entity_kind.eq(entity_kind))
+        .filter(integrity_envelope::entity_id.eq(&entity_id))
+        .first(conn)
+        .await
+        .map_err(|err| match err {
+            diesel::result::Error::NotFound => Error::MissingEnvelope { entity_kind },
+            other => Error::Database(db::DatabaseError::from(other)),
+        })?;
+
+    if envelope.payload_version != expected_payload_version {
+        return Err(Error::PayloadVersionMismatch {
+            entity_kind,
+            expected: expected_payload_version,
+            found: envelope.payload_version,
+        });
+    }
+
+    let payload = entity.canonical_payload_bytes();
+    let payload_hash = payload_hash(&payload);
+    let mac_input = build_mac_input(
+        entity_kind,
+        &entity_id,
+        envelope.payload_version,
+        &payload_hash,
+    );
+
+    let ok = keyholder
+        .ask(VerifyIntegrity {
+            mac_input,
+            expected_mac: envelope.mac,
+            key_version: envelope.key_version,
+        })
+        .await
+        .map_err(|err| match err {
+            kameo::error::SendError::HandlerError(inner) => Error::Keyholder(inner),
+            _ => Error::KeyholderSend,
+        })?;
+
+    if !ok {
+        return Err(Error::MacMismatch { entity_kind });
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use diesel::{ExpressionMethods as _, QueryDsl};
+    use diesel_async::RunQueryDsl;
+    use kameo::{actor::ActorRef, prelude::Spawn};
+
+    use crate::{
+        actors::keyholder::{Bootstrap, KeyHolder},
+        db::{self, schema},
+        safe_cell::{SafeCell, SafeCellHandle as _},
+    };
+
+    use super::{Error, IntegrityEntity, sign_entity, verify_entity};
+
+    #[derive(Clone)]
+    struct DummyEntity {
+        id: i32,
+        payload_version: i32,
+        payload: Vec<u8>,
+    }
+
+    impl IntegrityEntity for DummyEntity {
+        fn entity_kind(&self) -> &'static str {
+            "dummy_entity"
+        }
+
+        fn entity_id_bytes(&self) -> Vec<u8> {
+            self.id.to_be_bytes().to_vec()
+        }
+
+        fn payload_version(&self) -> i32 {
+            self.payload_version
+        }
+
+        fn canonical_payload_bytes(&self) -> Vec<u8> {
+            self.payload.clone()
+        }
+    }
+
+    async fn bootstrapped_keyholder(db: &db::DatabasePool) -> ActorRef<KeyHolder> {
+        let actor = KeyHolder::spawn(KeyHolder::new(db.clone()).await.unwrap());
+        actor
+            .ask(Bootstrap {
+                seal_key_raw: SafeCell::new(b"integrity-test-seal-key".to_vec()),
+            })
+            .await
+            .unwrap();
+        actor
+    }
+
+    #[tokio::test]
+    async fn sign_writes_envelope_and_verify_passes() {
+        let db = db::create_test_pool().await;
+        let keyholder = bootstrapped_keyholder(&db).await;
+        let mut conn = db.get().await.unwrap();
+
+        let entity = DummyEntity {
+            id: 7,
+            payload_version: 1,
+            payload: b"payload-v1".to_vec(),
+        };
+
+        sign_entity(&mut conn, &keyholder, &entity).await.unwrap();
+
+        let count: i64 = schema::integrity_envelope::table
+            .filter(schema::integrity_envelope::entity_kind.eq("dummy_entity"))
+            .filter(schema::integrity_envelope::entity_id.eq(entity.entity_id_bytes()))
+            .count()
+            .get_result(&mut conn)
+            .await
+            .unwrap();
+
+        assert_eq!(count, 1, "envelope row must be created exactly once");
+        verify_entity(&mut conn, &keyholder, &entity).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn tampered_mac_fails_verification() {
+        let db = db::create_test_pool().await;
+        let keyholder = bootstrapped_keyholder(&db).await;
+        let mut conn = db.get().await.unwrap();
+
+        let entity = DummyEntity {
+            id: 11,
+            payload_version: 1,
+            payload: b"payload-v1".to_vec(),
+        };
+
+        sign_entity(&mut conn, &keyholder, &entity).await.unwrap();
+
+        diesel::update(schema::integrity_envelope::table)
+            .filter(schema::integrity_envelope::entity_kind.eq("dummy_entity"))
+            .filter(schema::integrity_envelope::entity_id.eq(entity.entity_id_bytes()))
+            .set(schema::integrity_envelope::mac.eq(vec![0u8; 32]))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+
+        let err = verify_entity(&mut conn, &keyholder, &entity)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::MacMismatch { .. }));
+    }
+
+    #[tokio::test]
+    async fn changed_payload_fails_verification() {
+        let db = db::create_test_pool().await;
+        let keyholder = bootstrapped_keyholder(&db).await;
+        let mut conn = db.get().await.unwrap();
+
+        let entity = DummyEntity {
+            id: 21,
+            payload_version: 1,
+            payload: b"payload-v1".to_vec(),
+        };
+
+        sign_entity(&mut conn, &keyholder, &entity).await.unwrap();
+
+        let tampered = DummyEntity {
+            payload: b"payload-v1-but-tampered".to_vec(),
+            ..entity
+        };
+
+        let err = verify_entity(&mut conn, &keyholder, &tampered)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::MacMismatch { .. }));
+    }
+}