diff --git a/crates/config/maximal-config-example.toml b/crates/config/maximal-config-example.toml index a57e701fd..cb28ef497 100644 --- a/crates/config/maximal-config-example.toml +++ b/crates/config/maximal-config-example.toml @@ -248,6 +248,10 @@ role_subgraph_max_lag_secs = 1800 # rolling 24h; rejected and expired proposals don't count. Unset = no cap; 0 rejects all. # max_new_agreements_per_24h = 30 +# Escrow floor (GRT) admitting a sender that lacks the agreement-manager role: if its recovered +# signer holds at least this much escrow, the proposal proceeds to pricing/manifest checks. Default 1. +# min_escrow_grt = "1" + [dips.min_grt_per_30_days] # base rate component (per-network) # arbitrum-one = "450" # matic = "300" diff --git a/crates/config/src/config.rs b/crates/config/src/config.rs index 5db5d2f0c..3375a20ac 100644 --- a/crates/config/src/config.rs +++ b/crates/config/src/config.rs @@ -712,6 +712,10 @@ pub struct DipsConfig { /// window; rejected and expired proposals don't count. None (unset) disables the /// cap; a best-effort safeguard, not a security control. pub max_new_agreements_per_24h: Option, + /// Minimum escrow balance an untrusted sender's recovered signer must hold to + /// pass the escrow fallback. TODO: derive from the configured price minimum + /// plus a buffer for unknown entity counts instead of using a flat floor. + pub min_escrow_grt: GRT, } impl Default for DipsConfig { @@ -729,6 +733,7 @@ impl Default for DipsConfig { role_subgraph_max_lag_secs: 1_800, rpc_url: None, max_new_agreements_per_24h: None, + min_escrow_grt: GRT::ONE, } } } diff --git a/crates/config/src/grt.rs b/crates/config/src/grt.rs index b9facb1d5..448e7a3bd 100644 --- a/crates/config/src/grt.rs +++ b/crates/config/src/grt.rs @@ -12,6 +12,8 @@ pub struct GRT(u128); impl GRT { pub const ZERO: GRT = GRT(0); + /// One GRT in wei (10^18). + pub const ONE: GRT = GRT(1_000_000_000_000_000_000); /// Convert GRT string to wei for test construction. /// Panics on invalid input - only use in tests. diff --git a/crates/dips/src/escrow.rs b/crates/dips/src/escrow.rs new file mode 100644 index 000000000..5a7fd0c2b --- /dev/null +++ b/crates/dips/src/escrow.rs @@ -0,0 +1,87 @@ +// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs. +// SPDX-License-Identifier: Apache-2.0 + +//! Escrow fallback for agreement-proposal senders that are not on-chain agreement +//! managers: the role gate would reject them, but this lets one through when its +//! recovered signer resolves to a payer with enough escrow (a limited permissionless path). + +use thegraph_core::alloy::primitives::{Address, U256}; + +/// Outcome of looking a signer up in the escrow snapshot. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EscrowLookup { + /// The snapshot has no signers at all (cold start or total outage), so the + /// answer is "can't tell" rather than "unfunded". + Unavailable, + /// The signer resolved to a funded payer holding this balance (wei). + Balance(U256), + /// The snapshot is populated but the signer is not a funded payer. + NotFound, +} + +/// Read-side view of the escrow snapshot, keyed on the cryptographically +/// recovered signer (never a caller-written proposal field). +pub trait EscrowSource: std::fmt::Debug + Send + Sync { + fn lookup(&self, signer: Address) -> EscrowLookup; +} + +/// A fixed escrow snapshot. Used in tests and as a simple in-memory source; +/// production uses the shared TAP escrow watcher. +#[derive(Debug, Clone, Default)] +pub struct StaticEscrow(pub std::collections::HashMap); + +impl EscrowSource for StaticEscrow { + fn lookup(&self, signer: Address) -> EscrowLookup { + if self.0.is_empty() { + return EscrowLookup::Unavailable; + } + match self.0.get(&signer) { + Some(balance) => EscrowLookup::Balance(*balance), + None => EscrowLookup::NotFound, + } + } +} + +#[cfg(feature = "db")] +impl EscrowSource for indexer_monitor::EscrowAccountsWatcher { + fn lookup(&self, signer: Address) -> EscrowLookup { + let snapshot = self.borrow(); + if snapshot.signer_count() == 0 { + return EscrowLookup::Unavailable; + } + match snapshot.get_balance_for_signer(&signer) { + Ok(balance) => EscrowLookup::Balance(balance), + Err(_) => EscrowLookup::NotFound, + } + } +} + +#[cfg(test)] +mod test { + use thegraph_core::alloy::primitives::{address, U256}; + + use super::*; + + const SIGNER: Address = address!("f39fd6e51aad88f6f4ce6ab8827279cfffb92266"); + + #[test] + fn empty_snapshot_is_unavailable() { + assert_eq!( + StaticEscrow::default().lookup(SIGNER), + EscrowLookup::Unavailable + ); + } + + #[test] + fn known_signer_returns_balance() { + let escrow = StaticEscrow([(SIGNER, U256::from(5))].into_iter().collect()); + assert_eq!(escrow.lookup(SIGNER), EscrowLookup::Balance(U256::from(5))); + } + + #[test] + fn populated_snapshot_without_signer_is_not_found() { + let other = Address::repeat_byte(0x11); + let escrow = StaticEscrow([(other, U256::from(5))].into_iter().collect()); + assert_eq!(escrow.lookup(SIGNER), EscrowLookup::NotFound); + } +} diff --git a/crates/dips/src/lib.rs b/crates/dips/src/lib.rs index 9028232c5..7f9d67369 100644 --- a/crates/dips/src/lib.rs +++ b/crates/dips/src/lib.rs @@ -50,6 +50,7 @@ use std::sync::Arc; +use escrow::{EscrowLookup, EscrowSource}; use server::DipsServerContext; use thegraph_core::alloy::{ core::primitives::Address, @@ -62,6 +63,7 @@ use thegraph_core::alloy::{ #[cfg(feature = "db")] pub mod database; pub mod eip5267; +pub mod escrow; pub mod inflight; pub mod ipfs; #[cfg(any(feature = "rpc", feature = "db"))] @@ -209,6 +211,10 @@ pub enum DipsError { TrustVerificationUnavailable(String), #[error("indexer is at its DIPs agreement capacity ({limit} per 24h)")] CapacityExceeded { limit: u64 }, + #[error("sender {signer} has insufficient escrow to send agreement proposals")] + InsufficientEscrow { signer: Address }, + #[error("could not verify sender escrow: {0}")] + EscrowVerificationUnavailable(String), } #[cfg(feature = "rpc")] @@ -297,6 +303,25 @@ pub(crate) fn try_extract_deployment_id(rca_bytes: &[u8]) -> Option { /// Window over which the cap counts live agreements (pending or accepted). const CAPACITY_WINDOW: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60); +/// Authorize an untrusted sender by escrow: a resolved balance at or above +/// `min_wei` passes; below or absent is InsufficientEscrow; a wholesale-empty +/// snapshot is transient, so a funded sender isn't branded unfunded during an outage. +fn verify_escrow( + escrow: &dyn EscrowSource, + signer: Address, + min_wei: U256, +) -> Result<(), DipsError> { + match escrow.lookup(signer) { + EscrowLookup::Balance(balance) if balance >= min_wei => Ok(()), + EscrowLookup::Balance(_) | EscrowLookup::NotFound => { + Err(DipsError::InsufficientEscrow { signer }) + } + EscrowLookup::Unavailable => Err(DipsError::EscrowVerificationUnavailable( + "escrow snapshot is empty".to_string(), + )), + } +} + /// Validate and create a RecurringCollectionAgreement. /// /// Performs validation: @@ -323,6 +348,8 @@ pub async fn validate_and_create_rca( rca_domain, trusted_signers, max_new_agreements_per_24h, + escrow_source, + min_escrow_wei, } = ctx.as_ref(); // Decode SignedRCA @@ -330,10 +357,18 @@ pub async fn validate_and_create_rca( .map_err(|e| DipsError::AbiDecoding(e.to_string()))?; // 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. + // require it to be an on-chain agreement manager, or else a payer with escrow. let signer = signed_rca.recover_signer(rca_domain)?; tracing::debug!(%signer, "recovered RCA signer"); - trusted_signers.verify_trusted(signer).await?; + match trusted_signers.verify_trusted(signer).await { + Ok(()) => {} + // Not a manager: fall back to an escrow check on the recovered signer + // rather than rejecting; a transient role error stays transient. + Err(DipsError::SenderNotTrusted { .. }) => { + verify_escrow(escrow_source.as_ref(), signer, *min_escrow_wei)?; + } + Err(transient) => return Err(transient), + } // Validate service provider signed_rca.validate(expected_service_provider)?; @@ -499,6 +534,7 @@ mod test { use crate::{ derive_agreement_id, + escrow::StaticEscrow, ipfs::{EmptyNetworkIpfsFetcher, FailingIpfsFetcher, MockIpfsFetcher}, price::PriceCalculator, rca_eip712_domain, @@ -547,21 +583,16 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: trusted_signers_for_test(), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }) } - #[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 { + /// A context whose trusted set is empty (so the recovered signer is never a + /// role holder), with a fixed escrow snapshot and minimum, to exercise C's + /// escrow fallback path. + fn untrusted_escrow_ctx(escrow: StaticEscrow, min_escrow_wei: U256) -> Arc { + Arc::new(DipsServerContext { rca_store: Arc::new(InMemoryRcaStore::default()), ipfs_fetcher: Arc::new(MockIpfsFetcher::default()), price_calculator: Arc::new(PriceCalculator::new( @@ -574,12 +605,97 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: Arc::new(StaticTrustedSigners::default()), max_new_agreements_per_24h: None, - }); - let rca_bytes = rca_to_wire_bytes(rca); + escrow_source: Arc::new(escrow), + min_escrow_wei, + }) + } - let result = super::validate_and_create_rca(ctx, &service_provider, rca_bytes).await; + #[tokio::test] + async fn untrusted_sender_with_sufficient_escrow_passes() { + let service_provider = Address::repeat_byte(0x11); + let rca = create_test_rca( + Address::repeat_byte(0x42), + service_provider, + U256::from(200), + U256::from(100), + ); + // The recovered signer (test_signer) holds 2 wei; the minimum is 1. + let escrow = StaticEscrow( + [(test_signer().address(), U256::from(2))] + .into_iter() + .collect(), + ); + let ctx = untrusted_escrow_ctx(escrow, U256::from(1)); + + let result = + super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await; + + assert!(result.is_ok(), "got: {:?}", result); + } + + #[tokio::test] + async fn untrusted_sender_below_minimum_is_insufficient() { + let service_provider = Address::repeat_byte(0x11); + let rca = create_test_rca( + Address::repeat_byte(0x42), + service_provider, + U256::from(200), + U256::from(100), + ); + let escrow = StaticEscrow( + [(test_signer().address(), U256::from(1))] + .into_iter() + .collect(), + ); + let ctx = untrusted_escrow_ctx(escrow, U256::from(5)); + + let result = + super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await; + + assert!(matches!(result, Err(DipsError::InsufficientEscrow { .. }))); + } + + #[tokio::test] + async fn untrusted_sender_absent_from_escrow_is_insufficient() { + let service_provider = Address::repeat_byte(0x11); + let rca = create_test_rca( + Address::repeat_byte(0x42), + service_provider, + U256::from(200), + U256::from(100), + ); + // Snapshot has a different address, so it is populated but not the signer. + let escrow = StaticEscrow( + [(Address::repeat_byte(0xEE), U256::from(99))] + .into_iter() + .collect(), + ); + let ctx = untrusted_escrow_ctx(escrow, U256::from(1)); - assert!(matches!(result, Err(DipsError::SenderNotTrusted { .. }))); + let result = + super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await; + + assert!(matches!(result, Err(DipsError::InsufficientEscrow { .. }))); + } + + #[tokio::test] + async fn untrusted_sender_with_empty_snapshot_is_transient() { + let service_provider = Address::repeat_byte(0x11); + let rca = create_test_rca( + Address::repeat_byte(0x42), + service_provider, + U256::from(200), + U256::from(100), + ); + let ctx = untrusted_escrow_ctx(StaticEscrow::default(), U256::from(1)); + + let result = + super::validate_and_create_rca(ctx, &service_provider, rca_to_wire_bytes(rca)).await; + + assert!(matches!( + result, + Err(DipsError::EscrowVerificationUnavailable(_)) + )); } /// Sign an RCA with the fixed test key over the test domain and ABI-encode @@ -927,6 +1043,8 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: trusted_signers_for_test(), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }); let rca_bytes = rca_to_wire_bytes(rca); @@ -1103,6 +1221,8 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: trusted_signers_for_test(), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }); let rca_bytes = rca_to_wire_bytes(rca); @@ -1140,6 +1260,8 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: trusted_signers_for_test(), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }); let rca_bytes = rca_to_wire_bytes(rca); @@ -1177,6 +1299,8 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: trusted_signers_for_test(), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }); let rca_bytes = rca_to_wire_bytes(rca); @@ -1214,6 +1338,8 @@ mod test { rca_domain: test_rca_domain(), trusted_signers: trusted_signers_for_test(), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }); let rca_bytes = rca_to_wire_bytes(rca); diff --git a/crates/dips/src/server.rs b/crates/dips/src/server.rs index 28872040a..a5a334234 100644 --- a/crates/dips/src/server.rs +++ b/crates/dips/src/server.rs @@ -45,10 +45,14 @@ use std::{collections::BTreeMap, sync::Arc}; use async_trait::async_trait; -use thegraph_core::alloy::{primitives::Address, sol_types::Eip712Domain}; +use thegraph_core::alloy::{ + primitives::{Address, U256}, + sol_types::Eip712Domain, +}; use tonic::{Request, Response, Status}; use crate::{ + escrow::EscrowSource, inflight::{InflightCounter, InflightGuard}, ipfs::IpfsFetcher, price::PriceCalculator, @@ -86,6 +90,10 @@ pub struct DipsServerContext { pub trusted_signers: Arc, /// Max live DIPs agreements (pending or accepted) per rolling 24h window. None disables the cap. pub max_new_agreements_per_24h: Option, + /// Escrow snapshot used to admit senders that are not agreement managers. + pub escrow_source: Arc, + /// Minimum escrow balance (wei) an untrusted sender's signer must hold. + pub min_escrow_wei: U256, } /// DIPS server implementing RCA protocol. @@ -127,11 +135,13 @@ fn reject_reason_from_error(err: &DipsError) -> RejectReason { | DipsError::InvalidSubgraphManifest(_) | DipsError::InvalidRca(_) => RejectReason::Unspecified, 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 - } + DipsError::InsufficientEscrow { .. } => RejectReason::InsufficientEscrow, + // A store failure, an unverifiable role set, or an empty escrow snapshot + // means a valid proposal the indexer couldn't act on -- tell dipper it's + // transient so it retries. + DipsError::TrustVerificationUnavailable(_) + | DipsError::EscrowVerificationUnavailable(_) + | DipsError::UnknownError(_) => RejectReason::IndexerUnavailable, } } @@ -251,7 +261,7 @@ mod tests { use thegraph_core::alloy::primitives::U256; - use crate::trusted_signers::StaticTrustedSigners; + use crate::{escrow::StaticEscrow, trusted_signers::StaticTrustedSigners}; Arc::new(Self { rca_store: Arc::new(InMemoryRcaStore::default()), @@ -266,6 +276,8 @@ mod tests { rca_domain: crate::rca_eip712_domain(1337, Address::repeat_byte(0xCC)), trusted_signers: Arc::new(StaticTrustedSigners::default()), max_new_agreements_per_24h: None, + escrow_source: Arc::new(StaticEscrow::default()), + min_escrow_wei: U256::ZERO, }) } } @@ -552,6 +564,28 @@ mod tests { ); } + #[test] + fn test_reject_reason_insufficient_escrow() { + let err = DipsError::InsufficientEscrow { + signer: Address::repeat_byte(0x55), + }; + + assert_eq!( + super::reject_reason_from_error(&err), + RejectReason::InsufficientEscrow + ); + } + + #[test] + fn test_reject_reason_escrow_unverifiable_is_transient() { + let err = DipsError::EscrowVerificationUnavailable("escrow snapshot empty".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. diff --git a/crates/service/src/service.rs b/crates/service/src/service.rs index 097a41618..7187f660e 100644 --- a/crates/service/src/service.rs +++ b/crates/service/src/service.rs @@ -22,7 +22,7 @@ use indexer_dips::{ server::{DipsServer, DipsServerContext}, trusted_signers::{SubgraphTrustedSigners, TrustedSignerSource}, }; -use indexer_monitor::{DeploymentDetails, SubgraphClient}; +use indexer_monitor::{DeploymentDetails, EscrowAccountsWatcher, SubgraphClient}; use release::IndexerServiceRelease; use reqwest::Url; use tap_core::tap_eip712_domain; @@ -194,6 +194,7 @@ pub async fn run() -> anyhow::Result<()> { blockchain_chain_id, indexing_payments_subgraph .expect("indexing_payments client is built whenever DIPs is enabled"), + v2_watcher.clone(), &ipfs_url, database.clone(), indexer_address, @@ -265,10 +266,12 @@ pub async fn run() -> anyhow::Result<()> { /// Initialize and start the DIPS gRPC server. Errors return to the caller, /// which runs the service without DIPs rather than letting an optional /// subsystem take down query serving. +#[allow(clippy::too_many_arguments)] async fn start_dips( dips: &DipsConfig, blockchain_chain_id: u64, indexing_payments_subgraph: &'static SubgraphClient, + v2_watcher: EscrowAccountsWatcher, ipfs_url: &Url, database: sqlx::PgPool, indexer_address: Address, @@ -287,6 +290,7 @@ async fn start_dips( role_failopen_grace_secs, role_subgraph_max_lag_secs, max_new_agreements_per_24h, + min_escrow_grt, } = dips; // The RecurringCollector address is the EIP-712 verifying contract used to @@ -399,6 +403,12 @@ async fn start_dips( rca_domain, trusted_signers, max_new_agreements_per_24h: *max_new_agreements_per_24h, + // Untrusted senders fall back to the same V2 escrow snapshot the + // query-fee path already watches; no extra subgraph query. + escrow_source: Arc::new(v2_watcher), + // Config carries this as GRT (operator-friendly); compare in U256 wei + // because escrow balances from the snapshot are U256 wei. + min_escrow_wei: U256::from(min_escrow_grt.wei()), }); // Create DIPS server