feat(aggregator)!: support multiple payer-signer pairs

crates/bin/aggregator/README.md

@@ -23,15 +23,17 @@ As described in the [gateway README section on Graph Tally](https://github.com/e
 A JSON-RPC service for Graph Tally that lets clients request an aggregate receipt from a list of
 individual receipts.
 
-Usage: graph_tally_aggregator [OPTIONS] --private-key <PRIVATE_KEY>
+Usage: graph_tally_aggregator [OPTIONS]
 
 Options:
       --port <PORT>
           Port to listen on for JSON-RPC requests [env: GRAPH_TALLY_PORT=] [default: 8080]
-      --private-key <PRIVATE_KEY>
-          Sender private key for signing Receipt Aggregate Vouchers, as a hex string [env: GRAPH_TALLY_PRIVATE_KEY=]
-      --public-keys <PUBLIC_KEYS>
-          Signer public keys for incoming receipts/RAVs [env: GRAPH_TALLY_PUBLIC_KEYS=]
+      --signers <SIGNERS>
+          Signing key per payer, as `;`-separated `<payer>=<signer private key>` entries
+          [env: GRAPH_TALLY_SIGNERS=]
+      --accepted-signers <ACCEPTED_SIGNERS>
+          Extra accepted receipt signers per payer, as `;`-separated `<payer>=<signer>` entries
+          [env: GRAPH_TALLY_ACCEPTED_SIGNERS=]
       --max-request-body-size <MAX_REQUEST_BODY_SIZE>
           Maximum request body size in bytes. Defaults to 10MB [env: GRAPH_TALLY_MAX_REQUEST_BODY_SIZE=] [default: 10485760]
       --max-response-body-size <MAX_RESPONSE_BODY_SIZE>
@@ -56,6 +58,61 @@ Options:
           Print version
 ```
 
+## Signing keys and payers
+
+A RAV carries the `payer` named in the receipts it aggregates, and `GraphTallyCollector` recovers the
+RAV's signer and requires it to be authorized for **that** payer. A RAV signed by a key belonging to a
+different payer is rejected by indexers and uncollectable on chain.
+
+Because the collector binds each signer to exactly one authorizer, one key cannot serve two payers.
+Serving more than one payer therefore takes one signing key per payer:
+
+```sh
+GRAPH_TALLY_SIGNERS="0xPAYER_A=0xSIGNER_KEY_A;0xPAYER_B=0xSIGNER_KEY_B"
+```
+
+Left of `=` is a payer **address**; right of `=` is the private key of a **signer** authorized on chain
+for that payer. It is not the payer's own key -- that controls the escrow balance and belongs with the
+escrow manager, not with a public-facing service.
+
+The aggregator picks the signing key from the payer on the incoming receipts, and refuses requests for
+a payer it holds no key for rather than signing with whatever key is at hand.
+
+### Rotating a signer within one payer
+
+Receipts already issued under the previous signer stay valid as long as that address remains accepted
+**for its own payer**. List it explicitly, repeating the payer for several:
+
+```sh
+GRAPH_TALLY_ACCEPTED_SIGNERS="0xPAYER_A=0xOLD_SIGNER_A;0xPAYER_A=0xOLDER_SIGNER_A"
+```
+
+Each payer's own signing address is always accepted, so it need not be listed. Accepted signers are
+scoped per payer: a signer accepted for payer A does not vouch for payer B's receipts.
+
+The old signer must also stay authorized on chain until the last RAV it signed has been collected.
+Revocation is `thawSigner` → wait out `REVOKE_AUTHORIZATION_THAWING_PERIOD` → `revokeAuthorizedSigner`.
+
+### Migrating to a new payer
+
+A payer change is not a signer rotation. Keep a `GRAPH_TALLY_SIGNERS` entry for the old payer, with a
+key authorized to *it*, until every receipt issued under that payer has been aggregated and collected.
+Dropping the old entry early leaves those receipts unaggregatable.
+
+### Single-payer deployments
+
+Use `GRAPH_TALLY_SIGNERS` with one entry. There is no separate single-payer mode: naming the payer is
+what lets receipts for any *other* payer be refused, and a key on its own cannot say which payer it
+belongs to.
+
+`GRAPH_TALLY_PRIVATE_KEY` and `GRAPH_TALLY_PUBLIC_KEYS` were removed. Nothing reads them any more, so
+a deployment that still sets them starts normally and ignores them -- remove them from the manifest
+rather than relying on an error. Startup fails only if `GRAPH_TALLY_SIGNERS` is unset.
+
+To find the payer for an existing key, read the `authorizer` field of
+`GraphTallyCollector.authorizations(<that key's address>)`, or take it from the escrow manager's
+`payer = 0x…` startup line.
+
 Please refer to [GraphTallyCollector](https://github.com/graphprotocol/contracts/blob/main/packages/horizon/contracts/payments/collectors/GraphTallyCollector.sol) for more information about Receipt Aggregate Voucher signing keys.
 
 ## Operational recommendations

crates/bin/aggregator/src/aggregator.rs

@@ -1,26 +1,49 @@
 use std::collections::HashSet;
 
-use anyhow::{bail, Ok, Result};
+use anyhow::{anyhow, bail, Ok, Result};
 use graph_tally_core::{receipt::WithUniqueId, signed_message::Eip712SignedMessage};
 use graph_tally_graph::{Receipt, ReceiptAggregateVoucher};
 use rayon::prelude::*;
 use thegraph_core::alloy::{
     dyn_abi::Eip712Domain,
     primitives::{Address, FixedBytes},
-    signers::local::PrivateKeySigner,
     sol_types::SolStruct,
 };
 
+use crate::signers::SignerRegistry;
+
 pub fn check_and_aggregate_receipts(
     domain_separator: &Eip712Domain,
     receipts: &[Eip712SignedMessage<Receipt>],
     previous_rav: Option<Eip712SignedMessage<ReceiptAggregateVoucher>>,
-    wallet: &PrivateKeySigner,
-    accepted_addresses: &HashSet<Address>,
+    signers: &SignerRegistry,
 ) -> Result<Eip712SignedMessage<ReceiptAggregateVoucher>> {
     check_signatures_unique(receipts)?;
 
-    // Check that the receipts are signed by an accepted signer address
+    // Get the allocation id from the first receipt, return error if there are no receipts
+    let (collection_id, payer, data_service, service_provider) = match receipts.first() {
+        Some(receipt) => (
+            receipt.message.collection_id,
+            receipt.message.payer,
+            receipt.message.data_service,
+            receipt.message.service_provider,
+        ),
+        None => return Err(graph_tally_core::Error::NoValidReceiptsForRavRequest.into()),
+    };
+
+    // The payer is read before any signature is checked because it decides both halves of
+    // the check: which key signs the RAV, and which signers are acceptable on the way in.
+    // `check_collection_id` below proves the remaining receipts carry this same payer.
+    let (wallet, accepted_addresses) = signers.resolve(payer).ok_or_else(|| {
+        anyhow!(
+            "no signing key configured for payer {payer}; \
+             signing with another payer's key would produce an uncollectable RAV"
+        )
+    })?;
+
+    // Check that the receipts are signed by a signer accepted for *this payer*. A signer
+    // accepted for some other payer is not interchangeable: the RAV carries this payer, and
+    // the collector requires its signer to be authorized for it.
     receipts.par_iter().try_for_each(|receipt| {
         check_signature_is_from_one_of_addresses(receipt, domain_separator, accepted_addresses)
     })?;
@@ -37,17 +60,6 @@ pub fn check_and_aggregate_receipts(
     // Check that the receipts timestamp is greater than the previous rav
     check_receipt_timestamps(receipts, previous_rav.as_ref())?;
 
-    // Get the allocation id from the first receipt, return error if there are no receipts
-    let (collection_id, payer, data_service, service_provider) = match receipts.first() {
-        Some(receipt) => (
-            receipt.message.collection_id,
-            receipt.message.payer,
-            receipt.message.data_service,
-            receipt.message.service_provider,
-        ),
-        None => return Err(graph_tally_core::Error::NoValidReceiptsForRavRequest.into()),
-    };
-
     // Check that the receipts all have the same collection id
     check_collection_id(
         receipts,
@@ -193,6 +205,8 @@ mod tests {
         signers::local::PrivateKeySigner,
     };
 
+    use crate::signers::SignerRegistry;
+
     #[fixture]
     fn keys() -> (PrivateKeySigner, Address) {
         let wallet = PrivateKeySigner::random();
@@ -229,6 +243,126 @@ mod tests {
         graph_tally_eip712_domain(1, Address::from([0x11u8; 20]))
     }
 
+    /// Two payers, each with its own signing key -- the shape that a payer migration needs.
+    fn two_payer_registry(
+        payer_a: Address,
+        signer_a: &PrivateKeySigner,
+        payer_b: Address,
+        signer_b: &PrivateKeySigner,
+    ) -> SignerRegistry {
+        SignerRegistry::build(
+            [(payer_a, signer_a.clone()), (payer_b, signer_b.clone())],
+            [],
+        )
+        .unwrap()
+    }
+
+    fn receipt_for(
+        domain_separator: &Eip712Domain,
+        payer: Address,
+        signer: &PrivateKeySigner,
+        value: u128,
+    ) -> Eip712SignedMessage<Receipt> {
+        Eip712SignedMessage::new(
+            domain_separator,
+            Receipt::new(
+                collection_id(),
+                payer,
+                data_service(),
+                service_provider(),
+                value,
+            )
+            .unwrap(),
+            signer,
+        )
+        .unwrap()
+    }
+
+    #[rstest]
+    #[test]
+    /// The RAV must be signed by the key belonging to the payer named in the receipts.
+    ///
+    /// The collector recovers the RAV signer and requires it to be authorized for the RAV's
+    /// payer, so signing payer A's receipts with payer B's key yields a RAV that is rejected
+    /// by the indexer and uncollectable on chain.
+    fn signs_each_payer_with_its_own_key(domain_separator: Eip712Domain) {
+        let (payer_a, payer_b) = (Address::repeat_byte(0xa1), Address::repeat_byte(0xb2));
+        let (signer_a, signer_b) = (PrivateKeySigner::random(), PrivateKeySigner::random());
+        let registry = two_payer_registry(payer_a, &signer_a, payer_b, &signer_b);
+
+        for (payer, expected) in [(payer_a, &signer_a), (payer_b, &signer_b)] {
+            let receipts = vec![receipt_for(&domain_separator, payer, expected, 42)];
+            let rav =
+                super::check_and_aggregate_receipts(&domain_separator, &receipts, None, &registry)
+                    .unwrap();
+            assert_eq!(rav.message.payer, payer);
+            assert_eq!(
+                rav.recover_signer(&domain_separator).unwrap(),
+                expected.address(),
+            );
+        }
+    }
+
+    #[rstest]
+    #[test]
+    /// A payer with no configured key is refused rather than signed with whatever key is at
+    /// hand. Refusing is a loud failure; signing anyway is a silent one discovered days later.
+    fn refuses_a_payer_it_holds_no_key_for(domain_separator: Eip712Domain) {
+        let (payer_a, payer_b) = (Address::repeat_byte(0xa1), Address::repeat_byte(0xb2));
+        let (signer_a, signer_b) = (PrivateKeySigner::random(), PrivateKeySigner::random());
+        let registry = two_payer_registry(payer_a, &signer_a, payer_b, &signer_b);
+
+        let unknown = Address::repeat_byte(0xcc);
+        let receipts = vec![receipt_for(&domain_separator, unknown, &signer_a, 42)];
+        let err =
+            super::check_and_aggregate_receipts(&domain_separator, &receipts, None, &registry)
+                .unwrap_err();
+        assert!(err.to_string().contains("no signing key configured"));
+    }
+
+    #[rstest]
+    #[test]
+    /// Accepted signers are scoped per payer, so one payer's signer cannot vouch for another's
+    /// receipts. This is the case that turns a payer migration into uncollectable RAVs: the
+    /// receipts are accepted, the RAV carries the old payer, and the new key signs it.
+    fn rejects_a_signer_belonging_to_a_different_payer(domain_separator: Eip712Domain) {
+        let (payer_a, payer_b) = (Address::repeat_byte(0xa1), Address::repeat_byte(0xb2));
+        let (signer_a, signer_b) = (PrivateKeySigner::random(), PrivateKeySigner::random());
+        let registry = two_payer_registry(payer_a, &signer_a, payer_b, &signer_b);
+
+        // Receipts claiming payer A, signed by payer B's signer.
+        let receipts = vec![receipt_for(&domain_separator, payer_a, &signer_b, 42)];
+        let err =
+            super::check_and_aggregate_receipts(&domain_separator, &receipts, None, &registry)
+                .unwrap_err();
+        assert!(err.to_string().contains(&signer_b.address().to_string()));
+    }
+
+    #[rstest]
+    #[test]
+    /// A rotation *within* one payer is still supported: receipts from the payer's previous
+    /// signer aggregate into a RAV signed by its current one. Both keys are authorized for
+    /// that same payer on chain, so the result is valid.
+    fn accepts_a_previous_signer_of_the_same_payer(domain_separator: Eip712Domain) {
+        let payer = Address::repeat_byte(0xa1);
+        let (old_signer, new_signer) = (PrivateKeySigner::random(), PrivateKeySigner::random());
+        let registry = SignerRegistry::build(
+            [(payer, new_signer.clone())],
+            [(payer, old_signer.address())],
+        )
+        .unwrap();
+
+        let receipts = vec![receipt_for(&domain_separator, payer, &old_signer, 42)];
+        let rav =
+            super::check_and_aggregate_receipts(&domain_separator, &receipts, None, &registry)
+                .unwrap();
+        assert_eq!(rav.message.payer, payer);
+        assert_eq!(
+            rav.recover_signer(&domain_separator).unwrap(),
+            new_signer.address(),
+        );
+    }
+
     #[rstest]
     #[test]
     fn check_signatures_unique_fail(

crates/bin/aggregator/src/lib.rs

@@ -5,3 +5,4 @@ pub mod grpc;
 pub mod jsonrpsee_helpers;
 pub mod metrics;
 pub mod server;
+pub mod signers;