feat: add delegates (#41)

* feat(API): add delegate support to base client

* feat(API): add delegate implementation to mock client

* feat(API): add delegates implementation to Safe API client

* feat(CLI): add support for viewing and modifying delegates

* fix: API uses `.signHash` raw hash signing (dangerous)

* fix(API): client accepts empty signatures

* feat(account): add method to propose a SafeTx directly

* test: add tests for delegate feature (using MockClient)

* refactor: apply suggestions from code review

Co-authored-by: antazoey <admin@antazoey.me>

* refactor: use asserts rather than excepiton for mock

* refactor: don't add extra conversion logic in mock

* refactor(CLI): change `ERROR` to `INFO` for no delegates found

* refactor: use new `AccountAPI.sign_raw_msghash` method

* fix: forgot to update removal too

* refactor: use `account_option` instead of `callback_factory`

* fix: unused imports

* fix: asserts were wrong

* fix: use SafeClientException to mimic real environment

* chore: ignore not helpful logs to reduce display overload for testing

* chore: run ape tests and integration tests with different commands

* feat: add `SafeAccount.propose` as a shortcut to propose to Safe API

* fix: flake8 errors

* fix(API): use trailing slashes everywhere

---------

Co-authored-by: antazoey <admin@antazoey.me>

Author: fubuloubu

.github/workflows/test.yaml

@@ -88,5 +88,8 @@ jobs:
             python -m pip install --upgrade pip
             pip install .[test]
 
-        - name: Run Tests
-          run: ape test -n 0 -s --cov
+        - name: Run Functional Tests
+          run: ape test -n 0 tests/functional/ -s --cov -v WARNING
+
+        - name: Run Integration Tests
+          run: ape test -n 0 tests/integration/ -s --cov -v WARNING

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

@@ -406,6 +406,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(
@@ -653,7 +707,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``.
@@ -758,12 +812,11 @@ 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.

ape_safe/client/__init__.py

@@ -2,7 +2,7 @@
 from collections.abc import Iterator
 from datetime import datetime
 from functools import reduce
-from typing import Optional, Union, cast
+from typing import TYPE_CHECKING, Optional, Union, cast
 
 from ape.types import AddressType, HexBytes, MessageSignature
 from ape.utils import USER_AGENT, get_package_version
@@ -12,6 +12,7 @@
 from ape_safe.client.base import BaseSafeClient
 from ape_safe.client.mock import MockSafeClient
 from ape_safe.client.types import (
+    DelegateInfo,
     ExecutedTxData,
     OperationType,
     SafeApiTxData,
@@ -23,12 +24,16 @@
     UnexecutedTxData,
 )
 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
+
 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
@@ -80,7 +85,7 @@ def __init__(
 
     @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:
@@ -91,7 +96,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}/all-transactions/"
         while url:
             response = self._get(url)
             data = response.json()
@@ -110,7 +115,7 @@ 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"
+        url = f"multisig-transactions/{str(safe_tx_hash)}/confirmations/"
         while url:
             response = self._get(url)
             data = response.json()
@@ -129,7 +134,7 @@ def post_transaction(
                 b"",
             )
         )
-        post_dict: dict = {"signature": to_hex(signature), "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):
@@ -148,7 +153,7 @@ def post_transaction(
             # Signature handled above.
             post_dict.pop("signatures")
 
-        url = f"safes/{tx_data.safe}/multisig-transactions"
+        url = f"safes/{tx_data.safe}/multisig-transactions/"
         response = self._post(url, json=post_dict)
         return response
 
@@ -164,7 +169,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"multisig-transactions/{safe_tx_hash}/confirmations/"
         signature = to_hex(
             HexBytes(b"".join([x.encode_rsv() for x in order_by_signer(signatures)]))
         )
@@ -179,7 +184,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,
@@ -190,6 +195,53 @@ def estimate_gas_cost(
         gas = result.get("safeTxGas")
         return int(to_hex(HexBytes(gas)), 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",