feat: add delegates

.github/workflows/test.yaml

@@ -95,5 +95,8 @@ jobs:
         - name: Install gnosis safe
           run: npm install "@gnosis.pm/safe-contracts@${{ matrix.safe-version }}"
 
-        - name: Run Tests
-          run: pytest -n 0 -s --cov
+        - name: Run Functional Tests
+          run: ape test tests/functional/ -s --cov -v WARNING
+
+        - name: Run Integration Tests
+          run: pytest tests/integration/ -s --cov

ape_safe/_cli/__init__.py

@@ -1,5 +1,6 @@
 import click
 
+from ape_safe._cli.delegates import delegates
 from ape_safe._cli.pending import pending
 from ape_safe._cli.safe_mgmt import _list, add, all_txns, remove
 
@@ -17,3 +18,4 @@ def cli():
 cli.add_command(remove)
 cli.add_command(all_txns)
 cli.add_command(pending)
+cli.add_command(delegates)

ape_safe/_cli/delegates.py

@@ -0,0 +1,64 @@
+import click
+from ape.cli import ConnectedProviderCommand, account_option
+from ape.types import AddressType
+from eth_typing import ChecksumAddress
+
+from ape_safe._cli.click_ext import safe_argument, safe_cli_ctx, safe_option
+
+
+@click.group()
+def delegates():
+    """
+    View and configure delegates
+    """
+
+
+@delegates.command("list", cls=ConnectedProviderCommand)
+@safe_cli_ctx()
+@safe_argument
+def _list(cli_ctx, safe):
+    """
+    Show delegates for signers in a Safe
+    """
+    if delegates := safe.client.get_delegates():
+        cli_ctx.logger.success(f"Found delegates for {safe.address} ({safe.alias})")
+        for delegator in delegates:
+            click.echo(f"\nSigner {delegator}:")
+            click.echo("- " + "\n- ".join(delegates[delegator]))
+
+    else:
+        cli_ctx.logger.info(f"No delegates for {safe.address} ({safe.alias})")
+
+
+@delegates.command(cls=ConnectedProviderCommand)
+@safe_cli_ctx()
+@safe_option
+@click.argument("delegate", type=ChecksumAddress)
+@click.argument("label")
+@account_option()
+def add(cli_ctx, safe, delegate, label, account):
+    """
+    Add a delegate for a signer in a Safe
+    """
+    delegate = cli_ctx.conversion_manager.convert(delegate, AddressType)
+    safe.client.add_delegate(delegate, label, account)
+    cli_ctx.logger.success(
+        f"Added delegate {delegate} ({label}) for {account.address} "
+        f"in {safe.address} ({safe.alias})"
+    )
+
+
+@delegates.command(cls=ConnectedProviderCommand)
+@safe_cli_ctx()
+@safe_option
+@click.argument("delegate", type=ChecksumAddress)
+@account_option()
+def remove(cli_ctx, safe, delegate, account):
+    """
+    Remove a delegate for a specific signer in a Safe
+    """
+    delegate = cli_ctx.conversion_manager.convert(delegate, AddressType)
+    safe.client.remove_delegate(delegate, account)
+    cli_ctx.logger.success(
+        f"Removed delegate {delegate} for {account.address} in {safe.address} ({safe.alias})"
+    )

ape_safe/accounts.py

@@ -2,7 +2,7 @@
 import os
 from collections.abc import Iterable, Iterator, Mapping
 from pathlib import Path
-from typing import Any, Dict, Optional, Union, cast
+from typing import Any, Optional, Union, cast
 
 from ape.api import AccountAPI, AccountContainerAPI, ReceiptAPI, TransactionAPI
 from ape.api.address import BaseAddress
@@ -39,7 +39,7 @@
 
 
 class SafeContainer(AccountContainerAPI):
-    _accounts: Dict[str, "SafeAccount"] = {}
+    _accounts: dict[str, "SafeAccount"] = {}
 
     @property
     def _account_files(self) -> Iterator[Path]:
@@ -180,8 +180,8 @@ def _get_path(self, alias: str) -> Path:
 def get_signatures(
     safe_tx: SafeTx,
     signers: Iterable[AccountAPI],
-) -> Dict[AddressType, MessageSignature]:
-    signatures: Dict[AddressType, MessageSignature] = {}
+) -> dict[AddressType, MessageSignature]:
+    signatures: dict[AddressType, MessageSignature] = {}
     for signer in signers:
         signature = signer.sign_message(safe_tx)
         if signature:
@@ -362,6 +362,60 @@ def create_safe_tx(self, txn: Optional[TransactionAPI] = None, **safe_tx_kwargs)
         }
         return self.safe_tx_def(**safe_tx)
 
+    def all_delegates(self) -> Iterator[AddressType]:
+        for delegates in self.client.get_delegates().values():
+            yield from delegates
+
+    def propose_safe_tx(
+        self,
+        safe_tx: SafeTx,
+        submitter: Union[AccountAPI, AddressType, str, None] = None,
+        sigs_by_signer: Optional[dict[AddressType, MessageSignature]] = None,
+        contractTransactionHash: Optional[SafeTxID] = None,
+    ) -> SafeTxID:
+        """
+        Propose a safe_tx to the Safe API client
+        """
+        if not contractTransactionHash:
+            contractTransactionHash = get_safe_tx_hash(safe_tx)
+
+        if not sigs_by_signer:
+            sigs_by_signer = {}
+
+        if submitter is not None and not isinstance(submitter, AccountAPI):
+            submitter = self.load_submitter(submitter)
+
+        if (
+            submitter is not None
+            and submitter.address not in sigs_by_signer
+            and len(sigs_by_signer) < self.confirmations_required
+            and (submitter.address in self.signers or submitter.address in self.all_delegates())
+        ):
+            if sig := submitter.sign_message(safe_tx):
+                sigs_by_signer[submitter.address] = sig
+
+        # NOTE: Signatures don't have to be in order for Safe API post
+        self.client.post_transaction(
+            safe_tx,
+            sigs_by_signer,
+            sender=submitter.address if submitter else None,
+            contractTransactionHash=contractTransactionHash,
+        )
+
+        return contractTransactionHash
+
+    def propose(
+        self,
+        txn: Optional[TransactionAPI] = None,
+        submitter: Union[AccountAPI, AddressType, str, None] = None,
+        **safe_tx_kwargs,
+    ) -> SafeTxID:
+        """
+        Propose a transaction to the Safe API client
+        """
+        safe_tx = self.create_safe_tx(txn=txn, **safe_tx_kwargs)
+        return self.propose_safe_tx(safe_tx, submitter=submitter)
+
     def pending_transactions(self) -> Iterator[tuple[SafeTx, list[SafeTxConfirmation]]]:
         for executed_tx in self.client.get_transactions(confirmed=False):
             yield self.create_safe_tx(
@@ -533,7 +587,7 @@ def call(  # type: ignore[override]
 
         return super().call(txn, **call_kwargs)
 
-    def get_api_confirmations(self, safe_tx: SafeTx) -> Dict[AddressType, MessageSignature]:
+    def get_api_confirmations(self, safe_tx: SafeTx) -> dict[AddressType, MessageSignature]:
         safe_tx_id = get_safe_tx_hash(safe_tx)
         try:
             client_confirmations = self.client.get_confirmations(safe_tx_id)
@@ -558,7 +612,7 @@ def _contract_approvals(self, safe_tx: SafeTx) -> Mapping[AddressType, MessageSi
             if self.contract.approvedHashes(signer, safe_tx_hash) > 0
         }
 
-    def _all_approvals(self, safe_tx: SafeTx) -> Dict[AddressType, MessageSignature]:
+    def _all_approvals(self, safe_tx: SafeTx) -> dict[AddressType, MessageSignature]:
         approvals = self.get_api_confirmations(safe_tx)
 
         # NOTE: Do this last because it should take precedence
@@ -609,7 +663,7 @@ def sign_transaction(
             submit (bool): The option to submit the transaction. Defaults to ``True``.
             submitter (Union[``AccountAPI``, ``AddressType``, str, None]):
                 Determine who is submitting the transaction. Defaults to ``None``.
-            skip (Optional[List[Union[``AccountAPI, `AddressType``, str]]]):
+            skip (Optional[list[Union[``AccountAPI, `AddressType``, str]]]):
                 Allow bypassing any specified signer. Defaults to ``None``.
             signatures_required (Optional[int]):
                 The amount of signers required to confirm the transaction. Defaults to ``None``.
@@ -709,20 +763,19 @@ def skip_signer(signer: AccountAPI):
             f"for Safe {self.address}#{safe_tx.nonce}"  # TODO: put URI
         )
 
-        # NOTE: Signatures don't have to be in order for Safe API post
-        self.client.post_transaction(
+        self.propose_safe_tx(
             safe_tx,
-            sigs_by_signer,
+            submitter=submitter_account,
+            sigs_by_signer=sigs_by_signer,
             contractTransactionHash=safe_tx_hash,
-            sender=submitter_account.address,
         )
 
         # Return None so that Ape does not try to submit the transaction.
         return None
 
     def add_signatures(
         self, safe_tx: SafeTx, confirmations: Optional[list[SafeTxConfirmation]] = None
-    ) -> Dict[AddressType, MessageSignature]:
+    ) -> dict[AddressType, MessageSignature]:
         confirmations = confirmations or []
         if not self.local_signers:
             raise ApeSafeError("Cannot sign without local signers.")

ape_safe/client/__init__.py

@@ -1,15 +1,17 @@
 import json
 from datetime import datetime
 from functools import reduce
-from typing import Dict, Iterator, Optional, Union, cast
+from typing import Dict, Iterator, List, Optional, Union, cast
 
+from ape.api import AccountAPI
 from ape.types import AddressType, HexBytes, MessageSignature
 from ape.utils.misc import USER_AGENT, get_package_version
 from eip712.common import SafeTxV1, SafeTxV2
 
 from ape_safe.client.base import BaseSafeClient
 from ape_safe.client.mock import MockSafeClient
 from ape_safe.client.types import (
+    DelegateInfo,
     ExecutedTxData,
     OperationType,
     SafeApiTxData,
@@ -21,6 +23,7 @@
     UnexecutedTxData,
 )
 from ape_safe.exceptions import (
+    ActionNotPerformedError,
     ClientResponseError,
     ClientUnsupportedChainError,
     MultisigTransactionNotFoundError,
@@ -127,7 +130,7 @@ def post_transaction(
                 b"",
             )
         )
-        post_dict: Dict = {"signature": signature.hex(), "origin": ORIGIN}
+        post_dict: Dict = {"signature": signature.hex() if signature else None, "origin": ORIGIN}
 
         for key, value in tx_data.model_dump(by_alias=True, mode="json").items():
             if isinstance(value, HexBytes):
@@ -186,6 +189,53 @@ def estimate_gas_cost(
         gas = result.get("safeTxGas")
         return int(HexBytes(gas).hex(), 16)
 
+    def get_delegates(self) -> Dict[AddressType, List[AddressType]]:
+        url = "delegates"
+        delegates: Dict[AddressType, List[AddressType]] = {}
+
+        while url:
+            response = self._get(url, params={"safe": self.address})
+            data = response.json()
+
+            for delegate_info in map(DelegateInfo.model_validate, data.get("results", [])):
+                if delegate_info.delegator not in delegates:
+                    delegates[delegate_info.delegator] = [delegate_info.delegate]
+                else:
+                    delegates[delegate_info.delegator].append(delegate_info.delegate)
+
+            url = data.get("next")
+
+        return delegates
+
+    def add_delegate(self, delegate: AddressType, label: str, delegator: AccountAPI):
+        msg_hash = self.create_delegate_message(delegate)
+
+        # NOTE: This is required as Safe API uses an antiquated .signHash method
+        if not (sig := delegator.sign_raw_msghash(msg_hash)):
+            raise ActionNotPerformedError("Did not sign delegate approval")
+
+        payload = {
+            "safe": self.address,
+            "delegate": delegate,
+            "delegator": delegator.address,
+            "label": label,
+            "signature": sig.encode_rsv().hex(),
+        }
+        self._post("delegates", json=payload)
+
+    def remove_delegate(self, delegate: AddressType, delegator: AccountAPI):
+        msg_hash = self.create_delegate_message(delegate)
+
+        # NOTE: This is required as Safe API uses an antiquated .signHash method
+        if not (sig := delegator.sign_raw_msghash(msg_hash)):
+            raise ActionNotPerformedError("Did not sign delegate removal")
+
+        payload = {
+            "delegator": delegator.address,
+            "signature": sig.encode_rsv().hex(),
+        }
+        self._delete(f"delegates/{delegate}", json=payload)
+
 
 __all__ = [
     "ExecutedTxData",

ape_safe/client/base.py

@@ -1,12 +1,14 @@
+import time
 from abc import ABC, abstractmethod
-from collections.abc import Iterator
 from functools import cached_property
-from typing import Optional, Union
+from typing import Iterator, Optional, Union
 
 import certifi
 import requests
 import urllib3
-from ape.types import AddressType, MessageSignature
+from ape.api import AccountAPI
+from ape.types import AddressType, HexBytes, MessageSignature
+from eth_utils import keccak
 from requests import Response
 from requests.adapters import HTTPAdapter
 
@@ -112,6 +114,21 @@ def get_transactions(
 
             yield txn
 
+    def create_delegate_message(self, delegate: AddressType) -> HexBytes:
+        # NOTE: referencing https://github.com/safe-global/safe-eth-py/blob/
+        # a0a5771622f143ee6301cfc381c5ed50832ff482/gnosis/safe/api/transaction_service_api.py#L34
+        totp = int(time.time()) // 3600
+        return HexBytes(keccak(text=(delegate + str(totp))))
+
+    @abstractmethod
+    def get_delegates(self) -> dict[AddressType, list[AddressType]]: ...
+
+    @abstractmethod
+    def add_delegate(self, delegate: AddressType, label: str, delegator: AccountAPI): ...
+
+    @abstractmethod
+    def remove_delegate(self, delegate: AddressType, delegator: AccountAPI): ...
+
     """Request methods"""
 
     @cached_property
@@ -126,12 +143,15 @@ def session(self) -> requests.Session:
         session.mount("https://", adapter)
         return session
 
-    def _get(self, url: str) -> Response:
-        return self._request("GET", url)
+    def _get(self, url: str, params: Optional[dict] = None) -> Response:
+        return self._request("GET", url, params=params)
 
     def _post(self, url: str, json: Optional[dict] = None, **kwargs) -> Response:
         return self._request("POST", url, json=json, **kwargs)
 
+    def _delete(self, url: str, json: Optional[dict] = None, **kwargs) -> Response:
+        return self._request("DELETE", url, json=json, **kwargs)
+
     @cached_property
     def _http(self):
         return urllib3.PoolManager(ca_certs=certifi.where())