feat: Escrow based signer validation

Author: mangas

Cargo.lock

@@ -10,26 +10,28 @@ thiserror.workspace = true
 anyhow.workspace = true
 alloy-rlp = "0.3.10"
 thegraph-core.workspace = true
-tonic.workspace = true 
-async-trait.workspace = true 
-prost.workspace = true 
-prost-types.workspace = true 
+tonic.workspace = true
+async-trait.workspace = true
+prost.workspace = true
+prost-types.workspace = true
 uuid.workspace = true
 base64.workspace = true
 tokio.workspace = true
 sqlx.workspace = true
 futures = "0.3"
+indexer-monitor = { path = "../monitor" }
 
 http = "0.2"
 derivative = "2.2.0"
-ipfs-api-backend-hyper = {version = "0.6.0", features = ["with-send-sync"] }
-ipfs-api-prelude = {version = "0.6.0", features = ["with-send-sync"] }
+ipfs-api-backend-hyper = { version = "0.6.0", features = ["with-send-sync"] }
+ipfs-api-prelude = { version = "0.6.0", features = ["with-send-sync"] }
 bytes = "1.10.0"
 serde_yaml.workspace = true
 serde.workspace = true
 
 [dev-dependencies]
 rand = "0.9.0"
+indexer-watcher = { path = "../watcher" }
 
 [build-dependencies]
 tonic-build = { workspace = true }

crates/dips/Cargo.toml

@@ -3,8 +3,7 @@
 
 use std::{str::FromStr, sync::Arc};
 
-use ipfs::IpfsFetcher;
-use price::PriceCalculator;
+use server::DipsServerContext;
 use thegraph_core::alloy::{
     core::primitives::Address,
     primitives::{b256, ChainId, PrimitiveSignature as Signature, Uint, B256},
@@ -18,6 +17,7 @@ pub mod ipfs;
 pub mod price;
 pub mod proto;
 pub mod server;
+pub mod signers;
 pub mod store;
 
 use store::AgreementStore;
@@ -190,14 +190,16 @@ impl SignedIndexingAgreementVoucher {
     // TODO: Validate all values
     pub fn validate(
         &self,
+        signer_validator: &Arc<dyn signers::SignerValidator>,
         domain: &Eip712Domain,
         expected_payee: &Address,
         allowed_payers: impl AsRef<[Address]>,
     ) -> Result<(), DipsError> {
         let sig = Signature::from_str(&self.signature.to_string())
             .map_err(|err| DipsError::InvalidSignature(err.to_string()))?;
 
-        let payer = sig
+        let payer = self.voucher.payer;
+        let signer = sig
             .recover_address_from_prehash(&self.voucher.eip712_signing_hash(domain))
             .map_err(|err| DipsError::InvalidSignature(err.to_string()))?;
 
@@ -207,6 +209,10 @@ impl SignedIndexingAgreementVoucher {
             return Err(DipsError::PayerNotAuthorised(payer));
         }
 
+        signer_validator
+            .validate(&payer, &signer)
+            .map_err(|_| DipsError::SignerNotAuthorised(signer))?;
+
         if !self.voucher.recipient.eq(expected_payee) {
             return Err(DipsError::UnexpectedPayee {
                 expected: *expected_payee,
@@ -284,14 +290,18 @@ impl CollectionRequest {
 }
 
 pub async fn validate_and_create_agreement(
-    store: Arc<dyn AgreementStore>,
+    ctx: Arc<DipsServerContext>,
     domain: &Eip712Domain,
     expected_payee: &Address,
     allowed_payers: impl AsRef<[Address]>,
     voucher: Vec<u8>,
-    price_calculator: &PriceCalculator,
-    ipfs_client: Arc<dyn IpfsFetcher>,
 ) -> Result<Uuid, DipsError> {
+    let DipsServerContext {
+        store,
+        ipfs_fetcher,
+        price_calculator,
+        signer_validator,
+    } = ctx.as_ref();
     let decoded_voucher = SignedIndexingAgreementVoucher::abi_decode(voucher.as_ref(), true)
         .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
     let metadata = SubgraphIndexingVoucherMetadata::abi_decode(
@@ -300,9 +310,9 @@ pub async fn validate_and_create_agreement(
     )
     .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
 
-    decoded_voucher.validate(domain, expected_payee, allowed_payers)?;
+    decoded_voucher.validate(signer_validator, domain, expected_payee, allowed_payers)?;
 
-    let manifest = ipfs_client.fetch(&metadata.subgraphDeploymentId).await?;
+    let manifest = ipfs_fetcher.fetch(&metadata.subgraphDeploymentId).await?;
     match manifest.network() {
         Some(chain_id) if chain_id == metadata.chainId => {}
         Some(chain_id) => {
@@ -370,10 +380,11 @@ pub async fn validate_and_cancel_agreement(
 #[cfg(test)]
 mod test {
     use std::{
-        sync::Arc,
+        collections::HashMap,
         time::{Duration, SystemTime, UNIX_EPOCH},
     };
 
+    use indexer_monitor::EscrowAccounts;
     use rand::{distr::Alphanumeric, Rng};
     use thegraph_core::alloy::{
         primitives::{Address, FixedBytes, U256},
@@ -384,9 +395,9 @@ mod test {
 
     pub use crate::store::{AgreementStore, InMemoryAgreementStore};
     use crate::{
-        dips_agreement_eip712_domain, dips_cancellation_eip712_domain, ipfs::TestIpfsClient,
-        price::PriceCalculator, CancellationRequest, DipsError, IndexingAgreementVoucher,
-        SignedIndexingAgreementVoucher, SubgraphIndexingVoucherMetadata,
+        dips_agreement_eip712_domain, dips_cancellation_eip712_domain, server::DipsServerContext,
+        CancellationRequest, DipsError, IndexingAgreementVoucher, SignedIndexingAgreementVoucher,
+        SubgraphIndexingVoucherMetadata,
     };
 
     #[tokio::test]
@@ -424,22 +435,19 @@ mod test {
         let abi_voucher = voucher.abi_encode();
         let id = Uuid::from_bytes(voucher.voucher.agreement_id.into());
 
-        let store = Arc::new(InMemoryAgreementStore::default());
-
+        let ctx = DipsServerContext::for_testing();
         let actual_id = super::validate_and_create_agreement(
-            store.clone(),
+            ctx.clone(),
             &domain,
             &payee_addr,
             vec![payer_addr],
             abi_voucher,
-            &PriceCalculator::for_testing(),
-            Arc::new(TestIpfsClient::mainnet()),
         )
         .await
         .unwrap();
         assert_eq!(actual_id, id);
 
-        let stored_agreement = store.get_by_id(actual_id).await.unwrap().unwrap();
+        let stored_agreement = ctx.store.get_by_id(actual_id).await.unwrap().unwrap();
 
         assert_eq!(voucher, stored_agreement.voucher);
         assert!(!stored_agreement.cancelled);
@@ -448,6 +456,7 @@ mod test {
 
     #[test]
     fn voucher_signature_verification() {
+        let ctx = DipsServerContext::for_testing();
         let deployment_id = "Qmbg1qF4YgHjiVfsVt6a13ddrVcRtWyJQfD4LA3CwHM29f".to_string();
         let payee = PrivateKeySigner::random();
         let payee_addr = payee.address();
@@ -480,23 +489,34 @@ mod test {
         let signed = voucher.sign(&domain, payer).unwrap();
         assert_eq!(
             signed
-                .validate(&domain, &payee_addr, vec![])
+                .validate(&ctx.signer_validator, &domain, &payee_addr, vec![])
                 .unwrap_err()
                 .to_string(),
             DipsError::PayerNotAuthorised(voucher.payer).to_string()
         );
         assert!(signed
-            .validate(&domain, &payee_addr, vec![payer_addr])
+            .validate(
+                &ctx.signer_validator,
+                &domain,
+                &payee_addr,
+                vec![payer_addr]
+            )
             .is_ok());
     }
 
-    #[test]
-    fn check_voucher_modified() {
-        let deployment_id = "Qmbg1qF4YgHjiVfsVt6a13ddrVcRtWyJQfD4LA3CwHM29f".to_string();
+    #[tokio::test]
+    async fn check_voucher_modified() {
         let payee = PrivateKeySigner::random();
         let payee_addr = payee.address();
         let payer = PrivateKeySigner::random();
         let payer_addr = payer.address();
+        let ctx = DipsServerContext::for_testing_mocked_accounts(EscrowAccounts::new(
+            HashMap::default(),
+            HashMap::from_iter(vec![(payer_addr, vec![payer_addr])]),
+        ))
+        .await;
+
+        let deployment_id = "Qmbg1qF4YgHjiVfsVt6a13ddrVcRtWyJQfD4LA3CwHM29f".to_string();
 
         let metadata = SubgraphIndexingVoucherMetadata {
             basePricePerEpoch: U256::from(10000_u64),
@@ -526,9 +546,14 @@ mod test {
 
         assert!(matches!(
             signed
-                .validate(&domain, &payee_addr, vec![payer_addr])
+                .validate(
+                    &ctx.signer_validator,
+                    &domain,
+                    &payee_addr,
+                    vec![payer_addr]
+                )
                 .unwrap_err(),
-            DipsError::PayerNotAuthorised(_)
+            DipsError::SignerNotAuthorised(_)
         ));
     }
 
@@ -630,8 +655,8 @@ mod test {
 
     #[tokio::test]
     async fn test_create_and_cancel_agreement() -> anyhow::Result<()> {
+        let ctx = DipsServerContext::for_testing();
         let voucher_ctx = VoucherContext::random();
-        let store = Arc::new(InMemoryAgreementStore::default());
 
         // Create metadata and voucher
         let metadata = SubgraphIndexingVoucherMetadata {
@@ -645,13 +670,11 @@ mod test {
 
         // Create agreement
         let agreement_id = super::validate_and_create_agreement(
-            store.clone(),
+            ctx.clone(),
             &voucher_ctx.domain(),
             &voucher_ctx.payee.address(),
             vec![voucher_ctx.payer.address()],
             signed_voucher.encode_vec(),
-            &PriceCalculator::for_testing(),
-            Arc::new(TestIpfsClient::mainnet()),
         )
         .await?;
 
@@ -664,7 +687,7 @@ mod test {
 
         // Cancel agreement
         let cancelled_id = super::validate_and_cancel_agreement(
-            store.clone(),
+            ctx.store.clone(),
             &cancel_domain,
             signed_cancel.encode_vec(),
         )
@@ -673,16 +696,16 @@ mod test {
         assert_eq!(agreement_id, cancelled_id);
 
         // Verify agreement is cancelled
-        let stored_agreement = store.get_by_id(agreement_id).await?.unwrap();
+        let stored_agreement = ctx.store.get_by_id(agreement_id).await?.unwrap();
         assert!(stored_agreement.cancelled);
 
         Ok(())
     }
 
     #[tokio::test]
     async fn test_create_validations_errors() -> anyhow::Result<()> {
+        let ctx = DipsServerContext::for_testing();
         let voucher_ctx = VoucherContext::random();
-        let store = Arc::new(InMemoryAgreementStore::default());
 
         let metadata = SubgraphIndexingVoucherMetadata {
             basePricePerEpoch: U256::from(10000_u64),
@@ -734,13 +757,11 @@ mod test {
         let cases = vec![wrong_network_voucher, low_price_voucher, valid_voucher];
         for (voucher, result) in cases.into_iter().zip(expected_result.into_iter()) {
             let out = super::validate_and_create_agreement(
-                store.clone(),
+                ctx.clone(),
                 &voucher_ctx.domain(),
                 &voucher_ctx.payee.address(),
                 vec![voucher_ctx.payer.address()],
                 voucher.encode_vec(),
-                &PriceCalculator::for_testing(),
-                Arc::new(TestIpfsClient::mainnet()),
             )
             .await;
 

crates/dips/src/lib.rs

@@ -15,18 +15,53 @@ use crate::{
         CancelAgreementResponse, ProposalResponse, SubmitAgreementProposalRequest,
         SubmitAgreementProposalResponse,
     },
+    signers::SignerValidator,
     store::AgreementStore,
     validate_and_cancel_agreement, validate_and_create_agreement,
 };
 
+#[derive(Debug)]
+pub struct DipsServerContext {
+    pub store: Arc<dyn AgreementStore>,
+    pub ipfs_fetcher: Arc<dyn IpfsFetcher>,
+    pub price_calculator: PriceCalculator,
+    pub signer_validator: Arc<dyn SignerValidator>,
+}
+
+impl DipsServerContext {
+    #[cfg(test)]
+    pub fn for_testing() -> Arc<Self> {
+        use std::sync::Arc;
+
+        use crate::{ipfs::TestIpfsClient, signers, test::InMemoryAgreementStore};
+
+        Arc::new(DipsServerContext {
+            store: Arc::new(InMemoryAgreementStore::default()),
+            ipfs_fetcher: Arc::new(TestIpfsClient::mainnet()),
+            price_calculator: PriceCalculator::for_testing(),
+            signer_validator: Arc::new(signers::NoopSignerValidator),
+        })
+    }
+
+    #[cfg(test)]
+    pub async fn for_testing_mocked_accounts(accounts: EscrowAccounts) -> Arc<Self> {
+        use crate::{ipfs::TestIpfsClient, signers, test::InMemoryAgreementStore};
+
+        Arc::new(DipsServerContext {
+            store: Arc::new(InMemoryAgreementStore::default()),
+            ipfs_fetcher: Arc::new(TestIpfsClient::mainnet()),
+            price_calculator: PriceCalculator::for_testing(),
+            signer_validator: Arc::new(signers::EscrowSignerValidator::mock(accounts).await),
+        })
+    }
+}
+
 #[derive(Debug)]
 pub struct DipsServer {
-    pub agreement_store: Arc<dyn AgreementStore>,
+    pub ctx: Arc<DipsServerContext>,
     pub expected_payee: Address,
     pub allowed_payers: Vec<Address>,
     pub domain: Eip712Domain,
-    pub ipfs_fetcher: Arc<dyn IpfsFetcher>,
-    pub price_calculator: PriceCalculator,
 }
 
 #[async_trait]
@@ -50,13 +85,11 @@ impl IndexerDipsService for DipsServer {
         // - The subgraph deployment is for a chain we support
         // - The subgraph deployment is available on IPFS
         validate_and_create_agreement(
-            self.agreement_store.clone(),
+            self.ctx.clone(),
             &self.domain,
             &self.expected_payee,
             &self.allowed_payers,
             signed_voucher,
-            &self.price_calculator,
-            self.ipfs_fetcher.clone(),
         )
         .await
         .map_err(Into::<tonic::Status>::into)?;
@@ -80,13 +113,9 @@ impl IndexerDipsService for DipsServer {
             return Err(Status::invalid_argument("invalid version"));
         }
 
-        validate_and_cancel_agreement(
-            self.agreement_store.clone(),
-            &self.domain,
-            signed_cancellation,
-        )
-        .await
-        .map_err(Into::<tonic::Status>::into)?;
+        validate_and_cancel_agreement(self.ctx.store.clone(), &self.domain, signed_cancellation)
+            .await
+            .map_err(Into::<tonic::Status>::into)?;
 
         Ok(tonic::Response::new(CancelAgreementResponse {}))
     }