feat(dips): reject agreement proposals from untrusted senders

The payments service recovers each proposal's signer but accepts any valid signature, with
no check on who sent it. Require the signer to hold the on-chain agreement-manager role, read
from the indexing-payments subgraph, and reject senders that are not in that set.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: MoonBoi9001

Cargo.lock

@@ -473,6 +473,10 @@ impl MetricsConfig {
 #[allow(deprecated)] // Allow using deprecated EscrowSubgraphConfig type
 pub struct SubgraphsConfig {
     pub network: NetworkSubgraphConfig,
+    /// Indexing-payments subgraph, read for the DIPs agreement-manager role set.
+    /// Required when DIPs is enabled.
+    #[serde(default)]
+    pub indexing_payments: Option<SubgraphConfig>,
     #[deprecated(note = "V2 escrow accounts are in the network subgraph; this field is ignored.")]
     #[serde(default)]
     pub escrow: Option<EscrowSubgraphConfig>,
@@ -682,6 +686,14 @@ pub struct DipsConfig {
     /// RecurringCollector address: the EIP-712 verifying contract used to recover
     /// the signer of incoming RCA proposals. Required when DIPs is enabled.
     pub recurring_collector: Option<Address>,
+    /// How often to re-pull the agreement-manager role set from the subgraph.
+    pub role_refresh_interval_secs: u64,
+    /// Extra time past the refresh interval that a cached role set stays trusted
+    /// while refreshes fail, before the gate fails closed (the fail-open window).
+    pub role_failopen_grace_secs: u64,
+    /// Max seconds the role subgraph head may lag wall-clock before its data is
+    /// treated as unreliable. 0 disables the check.
+    pub role_subgraph_max_lag_secs: u64,
 }
 
 impl Default for DipsConfig {
@@ -694,6 +706,9 @@ impl Default for DipsConfig {
             min_grt_per_billion_entities_per_30_days: GRT::ZERO,
             additional_networks: BTreeMap::new(),
             recurring_collector: None,
+            role_refresh_interval_secs: 86_400,
+            role_failopen_grace_secs: 3_600,
+            role_subgraph_max_lag_secs: 1_800,
         }
     }
 }

crates/config/src/config.rs

@@ -23,6 +23,7 @@ db = [
     "dep:graph-networks-registry",
     "dep:serde",
     "dep:serde_yaml",
+    "dep:bytes",
 ]
 
 [dependencies]
@@ -37,7 +38,7 @@ build-info = { workspace = true, optional = true }
 indexer-monitor = { path = "../monitor", optional = true }
 thiserror.workspace = true
 graph-networks-registry = { workspace = true, optional = true }
-serde = { workspace = true, optional = true }
+serde = { workspace = true, optional = true, features = ["derive"] }
 serde_yaml = { version = "0.9", optional = true }
 
 # IPFS client dependencies
@@ -58,6 +59,9 @@ indexer-monitor = { path = "../monitor" }
 graph-networks-registry.workspace = true
 build-info.workspace = true
 rand = "0.8"
+reqwest = { workspace = true, features = ["json"] }
+serde_json.workspace = true
+wiremock.workspace = true
 
 [build-dependencies]
 tonic-build = { workspace = true, optional = true }

crates/dips/Cargo.toml

@@ -71,6 +71,7 @@ mod registry;
 #[cfg(feature = "rpc")]
 pub mod server;
 pub mod store;
+pub mod trusted_signers;
 
 use thiserror::Error;
 use uuid::Uuid;
@@ -195,6 +196,10 @@ pub enum DipsError {
     DeadlineExpired { deadline: u64, now: u64 },
     #[error("agreement end time {ends_at} has already passed (current time: {now})")]
     AgreementExpired { ends_at: u64, now: u64 },
+    #[error("sender {signer} is not authorised to send agreement proposals")]
+    SenderNotTrusted { signer: Address },
+    #[error("could not verify sender authorisation: {0}")]
+    TrustVerificationUnavailable(String),
 }
 
 #[cfg(feature = "rpc")]
@@ -304,17 +309,18 @@ pub async fn validate_and_create_rca(
         registry,
         additional_networks,
         rca_domain,
+        trusted_signers,
     } = ctx.as_ref();
 
     // Decode SignedRCA
     let signed_rca = SignedRecurringCollectionAgreement::abi_decode(rca_bytes.as_ref())
         .map_err(|e| DipsError::AbiDecoding(e.to_string()))?;
 
-    // Authenticate the sender before acting on the proposal: recover the EIP-712
-    // signer. PR B gates this address against the on-chain agreement-manager role
-    // set; for now any cryptographically valid signature passes.
+    // Authenticate then authorize the sender: recover the EIP-712 signer, then
+    // require it to hold the on-chain agreement-manager role before doing any work.
     let signer = signed_rca.recover_signer(rca_domain)?;
     tracing::debug!(%signer, "recovered RCA signer");
+    trusted_signers.verify_trusted(signer).await?;
 
     // Validate service provider
     signed_rca.validate(expected_service_provider)?;
@@ -465,6 +471,7 @@ mod test {
         rca_eip712_domain,
         server::DipsServerContext,
         store::{FailingRcaStore, InMemoryRcaStore},
+        trusted_signers::{StaticTrustedSigners, TrustedSignerSource},
         AcceptIndexingAgreementMetadata, DipsError, IndexingAgreementTermsV1,
         RecurringCollectionAgreement, SignedRecurringCollectionAgreement,
     };
@@ -487,6 +494,12 @@ mod test {
         rca_eip712_domain(1337, Address::repeat_byte(0xCC))
     }
 
+    fn trusted_signers_for_test() -> Arc<dyn TrustedSignerSource> {
+        Arc::new(StaticTrustedSigners(HashSet::from([
+            test_signer().address()
+        ])))
+    }
+
     fn create_test_context() -> Arc<DipsServerContext> {
         Arc::new(DipsServerContext {
             rca_store: Arc::new(InMemoryRcaStore::default()),
@@ -499,9 +512,41 @@ mod test {
             registry: Arc::new(crate::registry::test_registry()),
             additional_networks: Arc::new(BTreeMap::new()),
             rca_domain: test_rca_domain(),
+            trusted_signers: trusted_signers_for_test(),
         })
     }
 
+    #[tokio::test]
+    async fn test_validate_and_create_rca_untrusted_signer_rejected() {
+        let service_provider = Address::repeat_byte(0x11);
+        let rca = create_test_rca(
+            Address::repeat_byte(0x42),
+            service_provider,
+            U256::from(200),
+            U256::from(100),
+        );
+
+        // Trusted set excludes the test signer, so a valid signature is rejected.
+        let ctx = Arc::new(DipsServerContext {
+            rca_store: Arc::new(InMemoryRcaStore::default()),
+            ipfs_fetcher: Arc::new(MockIpfsFetcher::default()),
+            price_calculator: Arc::new(PriceCalculator::new(
+                HashSet::from(["mainnet".to_string()]),
+                BTreeMap::from([("mainnet".to_string(), U256::from(100))]),
+                U256::from(50),
+            )),
+            registry: Arc::new(crate::registry::test_registry()),
+            additional_networks: Arc::new(BTreeMap::new()),
+            rca_domain: test_rca_domain(),
+            trusted_signers: Arc::new(StaticTrustedSigners::default()),
+        });
+        let rca_bytes = rca_to_wire_bytes(rca);
+
+        let result = super::validate_and_create_rca(ctx, &service_provider, rca_bytes).await;
+
+        assert!(matches!(result, Err(DipsError::SenderNotTrusted { .. })));
+    }
+
     /// Sign an RCA with the fixed test key over the test domain and ABI-encode
     /// the wrapper, producing the wire bytes `validate_and_create_rca` decodes
     /// and recovers the signer from.
@@ -845,6 +890,7 @@ mod test {
             registry: Arc::new(crate::registry::test_registry()),
             additional_networks: Arc::new(BTreeMap::new()),
             rca_domain: test_rca_domain(),
+            trusted_signers: trusted_signers_for_test(),
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1019,6 +1065,7 @@ mod test {
             registry: Arc::new(crate::registry::test_registry()),
             additional_networks: Arc::new(BTreeMap::new()),
             rca_domain: test_rca_domain(),
+            trusted_signers: trusted_signers_for_test(),
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1054,6 +1101,7 @@ mod test {
             registry: Arc::new(crate::registry::test_registry()),
             additional_networks: Arc::new(BTreeMap::new()),
             rca_domain: test_rca_domain(),
+            trusted_signers: trusted_signers_for_test(),
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);
@@ -1089,6 +1137,7 @@ mod test {
             registry: Arc::new(crate::registry::test_registry()),
             additional_networks: Arc::new(BTreeMap::new()),
             rca_domain: test_rca_domain(),
+            trusted_signers: trusted_signers_for_test(),
         });
 
         let rca_bytes = rca_to_wire_bytes(rca);

crates/dips/src/lib.rs

@@ -58,6 +58,7 @@ use crate::{
         SubmitAgreementProposalRequest, SubmitAgreementProposalResponse,
     },
     store::RcaStore,
+    trusted_signers::TrustedSignerSource,
     DipsError,
 };
 
@@ -81,6 +82,8 @@ pub struct DipsServerContext {
     pub additional_networks: Arc<BTreeMap<String, String>>,
     /// EIP-712 domain for recovering the RCA signer (RecurringCollector).
     pub rca_domain: Eip712Domain,
+    /// Authorises the recovered signer against the on-chain agreement-manager role.
+    pub trusted_signers: Arc<dyn TrustedSignerSource>,
 }
 
 /// DIPS server implementing RCA protocol.
@@ -119,9 +122,12 @@ fn reject_reason_from_error(err: &DipsError) -> RejectReason {
         DipsError::AbiDecoding(_)
         | DipsError::InvalidSubgraphManifest(_)
         | DipsError::InvalidRca(_) => RejectReason::Unspecified,
-        // A store failure means the proposal was valid but the indexer couldn't persist
-        // it -- tell dipper this is transient so it retries rather than giving up.
-        DipsError::UnknownError(_) => RejectReason::IndexerUnavailable,
+        DipsError::SenderNotTrusted { .. } => RejectReason::SenderNotTrusted,
+        // A store failure or an unverifiable role set means a valid proposal the
+        // indexer couldn't act on -- tell dipper it's transient so it retries.
+        DipsError::TrustVerificationUnavailable(_) | DipsError::UnknownError(_) => {
+            RejectReason::IndexerUnavailable
+        }
     }
 }
 
@@ -222,8 +228,11 @@ mod tests {
     impl DipsServerContext {
         pub fn for_testing() -> Arc<Self> {
             use std::collections::{BTreeMap, HashSet};
+
             use thegraph_core::alloy::primitives::U256;
 
+            use crate::trusted_signers::StaticTrustedSigners;
+
             Arc::new(Self {
                 rca_store: Arc::new(InMemoryRcaStore::default()),
                 ipfs_fetcher: Arc::new(MockIpfsFetcher::default()),
@@ -235,6 +244,7 @@ mod tests {
                 registry: Arc::new(crate::registry::test_registry()),
                 additional_networks: Arc::new(BTreeMap::new()),
                 rca_domain: crate::rca_eip712_domain(1337, Address::repeat_byte(0xCC)),
+                trusted_signers: Arc::new(StaticTrustedSigners::default()),
             })
         }
     }
@@ -472,6 +482,28 @@ mod tests {
         assert_eq!(reason, RejectReason::IndexerUnavailable);
     }
 
+    #[test]
+    fn test_reject_reason_sender_not_trusted() {
+        let err = DipsError::SenderNotTrusted {
+            signer: Address::repeat_byte(0x55),
+        };
+
+        assert_eq!(
+            super::reject_reason_from_error(&err),
+            RejectReason::SenderNotTrusted
+        );
+    }
+
+    #[test]
+    fn test_reject_reason_trust_unverifiable_is_transient() {
+        let err = DipsError::TrustVerificationUnavailable("role subgraph down".to_string());
+
+        assert_eq!(
+            super::reject_reason_from_error(&err),
+            RejectReason::IndexerUnavailable
+        );
+    }
+
     #[test]
     fn test_reject_detail_sanitizes_store_errors() {
         // Arrange: the raw database error mentions schema internals.