Source code for eth_defi.enzyme.generic_adapter

"""Enzyme generic adapter helpers.

- GenericAdapter is a vault adapter that allow the vault to perform any transaction

- GenericAdapter is not yet part of standard Enzyme distribution (not audited)

- Transactions are send in ABI encoded bundles

- Bundle can contain one or more smart contract calls like `approve()` and then `swapExactTokensForTokens()`

See :py:func:`eth_defi.enzyme.uniswap_v2.prepare_swap` how to use the generic adapter to make trades.

- Tokens are hold in `Vault` smart contract by default. They are transferred to GenericAdapter with
  `postActionIncomingAssetsTransferHandler` and `postActionSpendAssetsTransferHandler` function
  modifiers on `executeCalls()`

- Any `allowance()` is set on GenericAdapter address

.. warning::

    GenericAdapter is unaudited and dangerous. Do not use in production yet.

    For example, anyone can steal approve()'ed tokens in the example implementation if not handled correctly.

`See the GenericAdapter source code on Github <>`__.

import logging
from typing import TypeAlias, Collection, Tuple, Final

from eth_abi import encode
from eth_abi.exceptions import EncodingError
from eth_typing import HexAddress
from hexbytes import HexBytes
from web3 import Web3
from web3.contract.contract import Contract, ContractFunction

from eth_defi.enzyme.integration_manager import IntegrationManagerActionId

ExternalCall: TypeAlias = Tuple[Contract, bytes]

Asset: TypeAlias = Contract | HexAddress

Signer: TypeAlias = HexAddress

#: See IntegrationManager.sol
EXECUTE_CALLS_SELECTOR: Final[str] = Web3.keccak(b"executeCalls(address,bytes,bytes)")[0:4]

logger = logging.getLogger(__name__)

def _addressify(asset: Contract | HexAddress | str):
    assert isinstance(asset, Contract) or type(asset) == HexAddress or type(asset) == str, f"Got bad asset: {asset}"
    if isinstance(asset, Contract):
        return asset.address
    return asset

def _addressify_collection(assets: Collection[Contract | HexAddress]):
    return [_addressify(a) for a in assets]

[docs]def encode_generic_adapter_execute_calls_args(incoming_assets: Collection[Asset], min_incoming_asset_amounts: Collection[int], spend_assets: Collection[Asset], spend_asset_amounts: Collection[int], external_calls: Collection[ExternalCall]): """Encode arguments for a generic adapter call.""" # const encodedExternalCallsData = encodeArgs( # ['address[]', 'bytes[]'], # [externalCallsData.contracts, externalCallsData.callsData], # ); addresses = [_addressify(t[0]) for t in external_calls] datas = [t[1] for t in external_calls] try: encoded_external_calls_data = encode(["address[]", "bytes[]"], [addresses, datas]) except EncodingError as e: raise EncodingError(f"Could not encode: {addresses} {datas}") from e # return encodeArgs( # ['address[]', 'uint256[]', 'address[]', 'uint256[]', 'bytes'], # [incomingAssets, minIncomingAssetAmounts, spendAssets, spendAssetAmounts, encodedExternalCallsData], # ); all_args_encoded = encode( ["address[]", "uint256[]", "address[]", "uint256[]", "bytes"], [_addressify_collection(incoming_assets), min_incoming_asset_amounts, _addressify_collection(spend_assets), spend_asset_amounts, encoded_external_calls_data], ) return all_args_encoded
# export function callOnIntegrationArgs({ # adapter, # selector, # encodedCallArgs, # }: { # adapter: AddressLike; # selector: BytesLike; # encodedCallArgs: BytesLike; # }) { # return encodeArgs(['address', 'bytes4', 'bytes'], [adapter, selector, encodedCallArgs]); # }
[docs]def encode_call_on_integration_args( adapter: Contract, selector: bytes, encoded_call_args: bytes | HexBytes, ): """No idea yet.""" assert type(selector) in (bytes, HexBytes) assert type(encoded_call_args) in (bytes, HexBytes), f"encoded_call_args is {encoded_call_args} {type(encoded_call_args)}" assert len(selector) == 4, f"Selector is {selector} {type(selector)}" assert len(encoded_call_args) > 0 return encode(["address", "bytes4", "bytes"], [_addressify(adapter), selector, encoded_call_args])
[docs]def execute_calls_for_generic_adapter( comptroller: Contract, external_calls: Collection[ExternalCall], generic_adapter: Contract, integration_manager: Contract, incoming_assets: Collection[Asset], min_incoming_asset_amounts: Collection[int], spend_assets: Collection[Asset], spend_asset_amounts: Collection[int], ) -> ContractFunction: """Create a vault buy/sell transaction using a generic adapter. :return: A contract function object with bound arguments """"execute_calls_for_generic_adapter(): %s %s %s %s %s %s %s %s", comptroller, external_calls, generic_adapter, integration_manager, incoming_assets, min_incoming_asset_amounts, spend_assets, spend_asset_amounts) # Sanity checks assert isinstance(comptroller, Contract) assert len(external_calls) > 0 assert isinstance(generic_adapter, Contract) # assert len(incoming_assets) > 0 assert isinstance(integration_manager, Contract) # assert len(min_incoming_asset_amounts) > 0 # assert len(spend_asset_amounts) > 0 # assert len(spend_assets) > 0 execute_call_args = encode_generic_adapter_execute_calls_args( incoming_assets=incoming_assets, min_incoming_asset_amounts=min_incoming_asset_amounts, spend_assets=spend_assets, spend_asset_amounts=spend_asset_amounts, external_calls=external_calls, ) call_args = encode_call_on_integration_args( generic_adapter, EXECUTE_CALLS_SELECTOR, execute_call_args, ) # See ComptrollerLib.sol call = comptroller.functions.callOnExtension(integration_manager.address, IntegrationManagerActionId.CallOnIntegration.value, call_args) return call