feat: refactor safe client to use multichain gateway

.github/workflows/test.yaml

@@ -89,7 +89,11 @@ jobs:
             pip install .[test]
 
         - name: Run Functional Tests
+          env:
+            APE_SAFE_GATEWAY_API_KEY: ${{ secrets.APE_SAFE_GATEWAY_API_KEY }}
           run: ape test -n 0 tests/functional/ -s --cov -v WARNING
 
         - name: Run Integration Tests
+          env:
+            APE_SAFE_GATEWAY_API_KEY: ${{ secrets.APE_SAFE_GATEWAY_API_KEY }}
           run: ape test -n 0 tests/integration/ -s --cov -v WARNING

ape_safe/_cli/safe_mgmt.py

@@ -10,6 +10,7 @@
 from eth_typing import ChecksumAddress
 
 from ape_safe._cli.click_ext import SafeCliContext, safe_argument, safe_cli_ctx
+from ape_safe.client import SafeClient
 
 
 @click.command(name="list")
@@ -145,9 +146,8 @@ def all_txns(cli_ctx: SafeCliContext, account, confirmed):
         account = cli_ctx.account_manager.load(account)
 
     address = cli_ctx.conversion_manager.convert(account, AddressType)
-
-    # NOTE: Create a client to support non-local safes.
-    client = cli_ctx.safes.create_client(address)
+    chain_id = cli_ctx.provider.chain_id
+    client = SafeClient(address=address, chain_id=chain_id)
 
     for txn in client.get_transactions(confirmed=confirmed):
         if isinstance(txn, ExecutedTxData):

ape_safe/accounts.py

@@ -169,27 +169,6 @@ def delete_account(self, alias: str):
         """
         self._get_path(alias).unlink(missing_ok=True)
 
-    def create_client(self, key: str) -> BaseSafeClient:
-        if key in self.aliases:
-            safe = self.load_account(key)
-            return safe.client
-
-        elif key in self.addresses:
-            account = cast(SafeAccount, self[cast(AddressType, key)])
-            return account.client
-
-        elif key in self.aliases:
-            return self.load_account(key).client
-
-        else:
-            address = self.conversion_manager.convert(key, AddressType)
-            if address in self.addresses:
-                account = cast(SafeAccount, self[cast(AddressType, key)])
-                return account.client
-
-            # Is not locally managed.
-            return SafeClient(address=address, chain_id=self.chain_manager.provider.chain_id)
-
     def _get_path(self, alias: str) -> Path:
         return self.data_folder.joinpath(f"{alias}.json")
 
@@ -220,17 +199,21 @@ def alias(self) -> str:
         return self.account_file_path.stem
 
     @property
-    def account_file(self) -> SafeCacheData:
+    def account_data(self) -> SafeCacheData:
         return SafeCacheData.model_validate_json(self.account_file_path.read_text(encoding="utf-8"))
 
+    @property
+    def deployed_chain_ids(self) -> list[int]:
+        return self.account_data.deployed_chain_ids
+
     @cached_property
     def address(self) -> AddressType:
         try:
             ecosystem = self.provider.network.ecosystem
         except ProviderNotConnectedError:
             ecosystem = self.network_manager.ethereum
 
-        return ecosystem.decode_address(self.account_file.address)
+        return ecosystem.decode_address(self.account_data.address)
 
     @cached_property
     def contract(self) -> "ContractInstance":
@@ -288,34 +271,32 @@ def set_guard(
 
         return self.contract.setGuard(new_guard, **tx_args)
 
-    @cached_property
-    def client(self) -> BaseSafeClient:
-        chain_id = self.provider.chain_id
-        override_url = os.environ.get("SAFE_TRANSACTION_SERVICE_URL")
+    def get_client(
+        self, chain_id: Optional[int] = None, override_url: Optional[str] = None
+    ) -> BaseSafeClient:
+        if chain_id is None:
+            chain_id = self.provider.chain_id
+
+        if override_url is None:
+            env_override = os.environ.get("SAFE_TRANSACTION_SERVICE_URL")
+            if env_override:
+                override_url = env_override
 
-        if self.provider.network.is_local:
+        if chain_id == 0 or (self.provider.network.is_local and self.provider.chain_id == chain_id):
             return MockSafeClient(contract=self.contract)
 
-        elif chain_id in self.account_file.deployed_chain_ids:
-            return SafeClient(
-                address=self.address, chain_id=self.provider.chain_id, override_url=override_url
-            )
+        return SafeClient(address=self.address, chain_id=chain_id, override_url=override_url)
 
-        elif (
+    @cached_property
+    def client(self) -> BaseSafeClient:
+        if (
             self.provider.network.name.endswith("-fork")
             and isinstance(self.provider.network, ForkedNetworkAPI)
-            and self.provider.network.upstream_chain_id in self.account_file.deployed_chain_ids
+            and self.provider.network.upstream_chain_id in self.deployed_chain_ids
         ):
-            return SafeClient(
-                address=self.address,
-                chain_id=self.provider.network.upstream_chain_id,
-                override_url=override_url,
-            )
-
-        elif self.provider.network.is_dev:
-            return MockSafeClient(contract=self.contract)
+            return self.get_client(chain_id=self.provider.network.upstream_chain_id)
 
-        return SafeClient(address=self.address, chain_id=self.provider.chain_id)
+        return self.get_client()
 
     @property
     def version(self) -> Version:
@@ -444,6 +425,7 @@ def propose_safe_tx(
 
         if submitter is not None and not isinstance(submitter, AccountAPI):
             submitter = self.load_submitter(submitter)
+            assert isinstance(submitter, AccountAPI)  # NOTE: mypy happy
 
         if (
             submitter is not None
@@ -701,6 +683,7 @@ def submit_safe_tx(
 
         if not isinstance(submitter, AccountAPI):
             submitter = self.load_submitter(submitter)
+            assert isinstance(submitter, AccountAPI)  # NOTE: mypy happy
 
         return submitter.call(txn)
 

ape_safe/client/__init__.py

@@ -1,4 +1,5 @@
 import json
+import os
 from collections.abc import Iterator
 from datetime import datetime
 from functools import reduce
@@ -26,37 +27,45 @@
 from ape_safe.exceptions import (
     ActionNotPerformedError,
     ClientResponseError,
-    ClientUnsupportedChainError,
     MultisigTransactionNotFoundError,
 )
 from ape_safe.utils import get_safe_tx_hash, order_by_signer
 
 if TYPE_CHECKING:
     from ape.api import AccountAPI
+    from requests import Response
 
 APE_SAFE_VERSION = get_package_version(__name__)
 APE_SAFE_USER_AGENT = f"Ape-Safe/{APE_SAFE_VERSION} {USER_AGENT}"
 # NOTE: Origin must be a string, but can be json that contains url & name fields
 ORIGIN = json.dumps(dict(url="https://apeworx.io", name="Ape Safe", ua=APE_SAFE_USER_AGENT))
 assert len(ORIGIN) <= 200  # NOTE: Must be less than 200 chars
 
-TRANSACTION_SERVICE_URL = {
-    # NOTE: If URLs need to be updated, a list of available service URLs can be found at
-    # https://docs.safe.global/safe-core-api/available-services.
-    # NOTE: There should be no trailing slashes at the end of the URL.
-    1: "https://safe-transaction-mainnet.safe.global",
-    10: "https://safe-transaction-optimism.safe.global",
-    56: "https://safe-transaction-bsc.safe.global",
-    100: "https://safe-transaction-gnosis-chain.safe.global",
-    137: "https://safe-transaction-polygon.safe.global",
-    250: "https://safe-txservice.fantom.network",
-    288: "https://safe-transaction.mainnet.boba.network",
-    8453: "https://safe-transaction-base.safe.global",
-    42161: "https://safe-transaction-arbitrum.safe.global",
-    43114: "https://safe-transaction-avalanche.safe.global",
-    84531: "https://safe-transaction-base-testnet.safe.global",
-    11155111: "https://safe-transaction-sepolia.safe.global",
-    81457: "https://transaction.blast-safe.io",
+# URL for the multichain client gateway
+SAFE_CLIENT_GATEWAY_URL = "https://api.safe.global/tx-service"
+GATEWAY_API_KEY = os.environ.get("APE_SAFE_GATEWAY_API_KEY")
+EIP3770_BLOCKCHAIN_NAMES_BY_CHAIN_ID = {
+    1: "eth",  # Ethereum Mainnet
+    11155111: "sep",  # Ethereum Sepolia
+    10: "oeth",  # Optimism Mainnet
+    42161: "arb1",  # Arbitrum One Mainnet
+    56: "bnb",  # Binance Smart Chain
+    146: "sonic",
+    5000: "mantle",
+    43114: "avax",
+    1313161554: "aurora",
+    8453: "base",  # Base Mainnet
+    84532: "basesep",  # Base Sepolia
+    42220: "celo",  # Celo Mainnet
+    100: "gno",  # Gnosis Chain
+    59144: "linea",  # Linea Mainnet
+    137: "pol",  # Polygon
+    534352: "scr",  # Scroll
+    130: "unichain",  # Unichain Mainnet
+    480: "wc",  # Worldchain Mainnet
+    324: "zksync",  # zkSync Mainnet
+    57073: "ink",  # Ink Mainnet
+    800094: "berachain",  # Berachain Mainnet
 }
 
 
@@ -68,24 +77,34 @@ def __init__(
         chain_id: Optional[int] = None,
     ) -> None:
         self.address = address
+        self.chain_id = chain_id
 
         if override_url:
-            tx_service_url = override_url
-
+            base_url = override_url
         elif chain_id:
-            if chain_id not in TRANSACTION_SERVICE_URL:
-                raise ClientUnsupportedChainError(chain_id)
+            if chain_id not in EIP3770_BLOCKCHAIN_NAMES_BY_CHAIN_ID:
+                raise ValueError(f"Chain ID {chain_id} is not a supported chain.")
 
-            tx_service_url = TRANSACTION_SERVICE_URL[chain_id]
+            elif not GATEWAY_API_KEY:
+                raise ValueError("Must provide API key via 'APE_SAFE_GATEWAY_API_KEY='.")
 
+            base_url = (
+                f"{SAFE_CLIENT_GATEWAY_URL}/{EIP3770_BLOCKCHAIN_NAMES_BY_CHAIN_ID[chain_id]}/api"
+            )
         else:
             raise ValueError("Must provide one of chain_id or override_url.")
 
-        super().__init__(tx_service_url)
+        super().__init__(base_url)
+
+    def _request(self, method: str, url: str, json: Optional[dict] = None, **kwargs) -> "Response":
+        # NOTE: Add authorization header
+        headers = kwargs.pop("headers", {})
+        headers.update(dict(Authorization=f"Bearer {GATEWAY_API_KEY}"))
+        return super()._request(method, url, json=json, headers=headers, **kwargs)
 
     @property
     def safe_details(self) -> SafeDetails:
-        response = self._get(f"safes/{self.address}/")
+        response = self._get(f"/safes/{self.address}")
         return SafeDetails.model_validate(response.json())
 
     def get_next_nonce(self) -> int:
@@ -96,7 +115,7 @@ def _all_transactions(self) -> Iterator[SafeApiTxData]:
         Get all transactions from safe, both confirmed and unconfirmed
         """
 
-        url = f"safes/{self.address}/all-transactions/"
+        url = f"/safes/{self.address}/multisig-transactions"
         while url:
             response = self._get(url)
             data = response.json()
@@ -115,12 +134,9 @@ def _all_transactions(self) -> Iterator[SafeApiTxData]:
             url = data.get("next")
 
     def get_confirmations(self, safe_tx_hash: SafeTxID) -> Iterator[SafeTxConfirmation]:
-        url = f"multisig-transactions/{str(safe_tx_hash)}/confirmations/"
-        while url:
-            response = self._get(url)
-            data = response.json()
-            yield from map(SafeTxConfirmation.model_validate, data.get("results"))
-            url = data.get("next")
+        response = self._get(f"/multisig-transactions/{safe_tx_hash}/raw")
+        data = response.json()
+        yield from map(SafeTxConfirmation.model_validate, data.get("confirmations", []))
 
     def post_transaction(
         self, safe_tx: SafeTx, signatures: dict[AddressType, MessageSignature], **kwargs
@@ -153,7 +169,7 @@ def post_transaction(
             # Signature handled above.
             post_dict.pop("signatures")
 
-        url = f"safes/{tx_data.safe}/multisig-transactions/"
+        url = f"/transactions/{tx_data.safe}/propose"
         response = self._post(url, json=post_dict)
         return response
 
@@ -169,7 +185,7 @@ def post_signatures(
             safe_tx_hash = safe_tx_or_hash
 
         safe_tx_hash = cast(SafeTxID, to_hex(HexBytes(safe_tx_hash)))
-        url = f"multisig-transactions/{safe_tx_hash}/confirmations/"
+        url = f"/transactions/{safe_tx_hash}/confirmations"
         signature = to_hex(
             HexBytes(b"".join([x.encode_rsv() for x in order_by_signer(signatures)]))
         )
@@ -184,7 +200,7 @@ def post_signatures(
     def estimate_gas_cost(
         self, receiver: AddressType, value: int, data: bytes, operation: int = 0
     ) -> Optional[int]:
-        url = f"safes/{self.address}/multisig-transactions/estimations/"
+        url = f"/safes/{self.address}/multisig-transactions/estimations"
         request: dict = {
             "to": receiver,
             "value": value,
@@ -196,7 +212,7 @@ def estimate_gas_cost(
         return int(to_hex(HexBytes(gas)), 16)
 
     def get_delegates(self) -> dict["AddressType", list["AddressType"]]:
-        url = "delegates/"
+        url = "/delegates"
         delegates: dict[AddressType, list[AddressType]] = {}
 
         while url:
@@ -227,7 +243,7 @@ def add_delegate(self, delegate: "AddressType", label: str, delegator: "AccountA
             "label": label,
             "signature": sig.encode_rsv().hex(),
         }
-        self._post("delegates/", json=payload)
+        self._post("/delegates", json=payload)
 
     def remove_delegate(self, delegate: "AddressType", delegator: "AccountAPI"):
         msg_hash = self.create_delegate_message(delegate)
@@ -240,7 +256,7 @@ def remove_delegate(self, delegate: "AddressType", delegator: "AccountAPI"):
             "delegator": delegator.address,
             "signature": sig.encode_rsv().hex(),
         }
-        self._delete(f"delegates/{delegate}/", json=payload)
+        self._delete(f"/delegates/{delegate}", json=payload)
 
 
 __all__ = [

ape_safe/client/base.py

@@ -33,8 +33,8 @@
 
 
 class BaseSafeClient(ABC):
-    def __init__(self, transaction_service_url: str):
-        self.transaction_service_url = transaction_service_url
+    def __init__(self, base_url: str):
+        self.base_url = base_url
 
     """Abstract methods"""
 
@@ -160,13 +160,15 @@ def _http(self):
         return urllib3.PoolManager(ca_certs=certifi.where())
 
     def _request(self, method: str, url: str, json: Optional[dict] = None, **kwargs) -> "Response":
+        api_version = kwargs.pop("api_version", "v1")
+
         # NOTE: paged requests include full url already
-        if url.startswith(f"{self.transaction_service_url}/api/v1/"):
+        if url.startswith(self.base_url):
             api_url = url
+
         else:
-            # **WARNING**: The trailing slash in the URL is CRITICAL!
-            # If you remove it, things will not work as expected.
-            api_url = f"{self.transaction_service_url}/api/v1/{url}/"
+            api_url = f"{self.base_url}/{api_version}{url}"
+
         do_fail = not kwargs.pop("allow_failure", False)
 
         # Use `or 10` to handle when None is explicit.