feat(service): enforce V2 data_service via DataServiceCheck

  - Add tap/checks/data_service_check.rs to validate V2
  receipt.message.data_service against an allowlist(from configuration)
  - Return CheckError::Failed with clear message; maps to HTTP 400
  - Extend IndexerTapContext::get_checks(..) to accept allowed_data_services
  and append the check when provided
  - Wire ServiceRouter to pass blockchain.subgraph_service_address (Horizon
  mode), activating the check only when configured
  - V1 receipts unaffected; persistence logic unchanged

Author: neithanmo

crates/service/src/service/router.rs

@@ -278,13 +278,19 @@ impl ServiceRouter {
                 let receipt_max_value = max_receipt_value_grt.get_value();
 
                 // Create checks
+                let allowed_data_services = self
+                    .blockchain
+                    .subgraph_service_address
+                    .map(|addr| vec![addr]);
+
                 let checks = IndexerTapContext::get_checks(
                     self.database,
                     allocations.clone(),
                     escrow_accounts_v1.clone(),
                     escrow_accounts_v2.clone(),
                     timestamp_error_tolerance,
                     receipt_max_value,
+                    allowed_data_services,
                 )
                 .await;
 

crates/service/src/tap.rs

@@ -16,9 +16,10 @@ use tokio::sync::{
 use tokio_util::sync::CancellationToken;
 
 use crate::tap::checks::{
-    allocation_eligible::AllocationEligible, deny_list_check::DenyListCheck,
-    receipt_max_val_check::ReceiptMaxValueCheck, sender_balance_check::SenderBalanceCheck,
-    timestamp_check::TimestampCheck, value_check::MinimumValue,
+    allocation_eligible::AllocationEligible, data_service_check::DataServiceCheck,
+    deny_list_check::DenyListCheck, receipt_max_val_check::ReceiptMaxValueCheck,
+    sender_balance_check::SenderBalanceCheck, timestamp_check::TimestampCheck,
+    value_check::MinimumValue,
 };
 
 mod checks;
@@ -56,8 +57,9 @@ impl IndexerTapContext {
         escrow_accounts_v2: Option<Receiver<EscrowAccounts>>,
         timestamp_error_tolerance: Duration,
         receipt_max_value: u128,
+        allowed_data_services: Option<Vec<Address>>,
     ) -> Vec<ReceiptCheck<TapReceipt>> {
-        vec![
+        let mut checks: Vec<ReceiptCheck<TapReceipt>> = vec![
             Arc::new(AllocationEligible::new(indexer_allocations)),
             Arc::new(SenderBalanceCheck::new(
                 escrow_accounts_v1,
@@ -67,7 +69,13 @@ impl IndexerTapContext {
             Arc::new(DenyListCheck::new(pgpool.clone()).await),
             Arc::new(ReceiptMaxValueCheck::new(receipt_max_value)),
             Arc::new(MinimumValue::new(pgpool, Duration::from_secs(GRACE_PERIOD)).await),
-        ]
+        ];
+
+        if let Some(addrs) = allowed_data_services {
+            checks.push(Arc::new(DataServiceCheck::new(addrs)));
+        }
+
+        checks
     }
 
     pub async fn new(

crates/service/src/tap/checks.rs

@@ -3,6 +3,7 @@
 
 pub mod allocation_eligible;
 pub mod deny_list_check;
+pub mod data_service_check;
 pub mod receipt_max_val_check;
 pub mod sender_balance_check;
 pub mod timestamp_check;

crates/service/src/tap/checks/data_service_check.rs

@@ -0,0 +1,49 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+use tap_core::receipt::checks::{Check, CheckError, CheckResult};
+use thegraph_core::alloy::{hex::ToHexExt, primitives::Address};
+
+use crate::tap::{CheckingReceipt, TapReceipt};
+
+/// Validates that the V2 receipt's `data_service` field matches an
+/// allowed SubgraphService address (or one of them).
+///
+/// - V1 receipts are ignored by this check (always Ok).
+/// - On mismatch, returns a CheckFailure with a descriptive message.
+pub struct DataServiceCheck {
+    allowed: Vec<Address>,
+}
+
+impl DataServiceCheck {
+    pub fn new(allowed: Vec<Address>) -> Self {
+        Self { allowed }
+    }
+}
+
+#[async_trait::async_trait]
+impl Check<TapReceipt> for DataServiceCheck {
+    async fn check(
+        &self,
+        _: &tap_core::receipt::Context,
+        receipt: &CheckingReceipt,
+    ) -> CheckResult {
+        match receipt.signed_receipt() {
+            // Not applicable for V1
+            TapReceipt::V1(_) => Ok(()),
+
+            // Validate data_service for V2
+            TapReceipt::V2(r) => {
+                let got = r.message.data_service;
+                if self.allowed.contains(&got) {
+                    Ok(())
+                } else {
+                    Err(CheckError::Failed(anyhow::anyhow!(
+                        "Invalid data_service: {} is not allowed for this indexer",
+                        got.encode_hex()
+                    )))
+                }
+            }
+        }
+    }
+}