start plugin

README.md

@@ -13,16 +13,16 @@ TODO: Description
 You can install the latest release via [`pip`](https://pypi.org/project/pip/):
 
 ```bash
-pip install <PYPI_NAME>
+pip install ape-uniswap
 ```
 
 ### via `setuptools`
 
 You can clone the repository and use [`setuptools`](https://github.com/pypa/setuptools) for the most up-to-date version:
 
 ```bash
-git clone https://github.com/ApeWorX/<PYPI_NAME>.git
-cd <PYPI_NAME>
+git clone https://github.com/ApeWorX/ape-uniswap.git
+cd ape-uniswap
 python3 setup.py install
 ```
 

ape_uniswap/__init__.py

@@ -0,0 +1,6 @@
+from .router_codec import RouterCodec
+
+
+__all__ = [
+    "RouterCodec",
+]

ape_uniswap/_abi_builder.py

@@ -0,0 +1,170 @@
+"""
+Factory that builds the UR function ABIs used by the Uniswap Universal Router Codec
+
+* Author: Elnaril (https://www.fiverr.com/elnaril, https://github.com/Elnaril).
+* License: MIT.
+* Doc: https://github.com/Elnaril/uniswap-universal-router-decoder
+"""
+from __future__ import annotations
+
+from dataclasses import (
+    asdict,
+    dataclass,
+)
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+)
+
+from eth_utils import function_abi_to_4byte_selector
+
+from ._enums import _RouterFunction
+
+
+@dataclass(frozen=True)
+class _FunctionABI:
+    inputs: List[Any]
+    name: str
+    type: str
+
+    def get_abi(self) -> Dict[str, Any]:
+        result = asdict(self)
+        if self.type == "tuple":
+            result["components"] = result.pop("inputs")
+        return result
+
+    def get_full_abi(self) -> List[Dict[str, Any]]:
+        return [self.get_abi()]
+
+
+@dataclass(frozen=True)
+class _FunctionDesc:
+    fct_abi: _FunctionABI
+    selector: bytes
+
+
+_ABIMap = Dict[_RouterFunction, _FunctionDesc]
+
+
+class _FunctionABIBuilder:
+    def __init__(self, fct_name: str, _type: str = "function") -> None:
+        self.abi = _FunctionABI(inputs=[], name=fct_name, type=_type)
+
+    def add_address(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "address"})
+        return self
+
+    def add_uint256(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "uint256"})
+        return self
+
+    add_int = add_uint256
+
+    def add_uint160(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "uint160"})
+        return self
+
+    def add_uint48(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "uint48"})
+        return self
+
+    def add_address_array(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "address[]"})
+        return self
+
+    def add_bool(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "bool"})
+        return self
+
+    def build(self) -> _FunctionABI:
+        return self.abi
+
+    @staticmethod
+    def create_struct(arg_name: str) -> _FunctionABIBuilder:
+        return _FunctionABIBuilder(arg_name, "tuple")
+
+    def add_struct(self, struct: _FunctionABIBuilder) -> _FunctionABIBuilder:
+        self.abi.inputs.append(struct.abi.get_abi())
+        return self
+
+    def add_bytes(self, arg_name: str) -> _FunctionABIBuilder:
+        self.abi.inputs.append({"name": arg_name, "type": "bytes"})
+        return self
+
+
+class _ABIBuilder:
+    def build_abi_map(self) -> _ABIMap:
+        abi_map: _ABIMap = {
+            # mapping between command identifier and fct descriptor (fct abi + selector)
+            _RouterFunction.V3_SWAP_EXACT_IN: self._add_mapping(self._build_v3_swap_exact_in),
+            _RouterFunction.V3_SWAP_EXACT_OUT: self._add_mapping(self._build_v3_swap_exact_out),
+            _RouterFunction.V2_SWAP_EXACT_IN: self._add_mapping(self._build_v2_swap_exact_in),
+            _RouterFunction.V2_SWAP_EXACT_OUT: self._add_mapping(self._build_v2_swap_exact_out),
+            _RouterFunction.PERMIT2_PERMIT: self._add_mapping(self._build_permit2_permit),
+            _RouterFunction.WRAP_ETH: self._add_mapping(self._build_wrap_eth),
+            _RouterFunction.UNWRAP_WETH: self._add_mapping(self._build_unwrap_weth),
+            _RouterFunction.SWEEP: self._add_mapping(self._build_sweep),
+            _RouterFunction.PAY_PORTION: self._add_mapping(self._build_pay_portion),
+        }
+        return abi_map
+
+    @staticmethod
+    def _add_mapping(build_abi_method: Callable[[], _FunctionABI]) -> _FunctionDesc:
+        fct_abi = build_abi_method()
+        selector = function_abi_to_4byte_selector(fct_abi.get_abi())
+        return _FunctionDesc(fct_abi=fct_abi, selector=selector)
+
+    @staticmethod
+    def _build_v2_swap_exact_in() -> _FunctionABI:
+        builder = _FunctionABIBuilder("V2_SWAP_EXACT_IN")
+        builder.add_address("recipient").add_int("amountIn").add_int("amountOutMin").add_address_array("path")
+        return builder.add_bool("payerIsSender").build()
+
+    @staticmethod
+    def _build_permit2_permit() -> _FunctionABI:
+        builder = _FunctionABIBuilder("PERMIT2_PERMIT")
+        inner_struct = builder.create_struct("details")
+        inner_struct.add_address("token").add_uint160("amount").add_uint48("expiration").add_uint48("nonce")
+        outer_struct = builder.create_struct("struct")
+        outer_struct.add_struct(inner_struct).add_address("spender").add_int("sigDeadline")
+        return builder.add_struct(outer_struct).add_bytes("data").build()
+
+    @staticmethod
+    def _build_unwrap_weth() -> _FunctionABI:
+        builder = _FunctionABIBuilder("UNWRAP_WETH")
+        return builder.add_address("recipient").add_int("amountMin").build()
+
+    @staticmethod
+    def _build_v3_swap_exact_in() -> _FunctionABI:
+        builder = _FunctionABIBuilder("V3_SWAP_EXACT_IN")
+        builder.add_address("recipient").add_int("amountIn").add_int("amountOutMin").add_bytes("path")
+        return builder.add_bool("payerIsSender").build()
+
+    @staticmethod
+    def _build_wrap_eth() -> _FunctionABI:
+        builder = _FunctionABIBuilder("WRAP_ETH")
+        return builder.add_address("recipient").add_int("amountMin").build()
+
+    @staticmethod
+    def _build_v2_swap_exact_out() -> _FunctionABI:
+        builder = _FunctionABIBuilder("V2_SWAP_EXACT_OUT")
+        builder.add_address("recipient").add_int("amountOut").add_int("amountInMax").add_address_array("path")
+        return builder.add_bool("payerIsSender").build()
+
+    @staticmethod
+    def _build_v3_swap_exact_out() -> _FunctionABI:
+        builder = _FunctionABIBuilder("V3_SWAP_EXACT_OUT")
+        builder.add_address("recipient").add_int("amountOut").add_int("amountInMax").add_bytes("path")
+        return builder.add_bool("payerIsSender").build()
+
+    @staticmethod
+    def _build_sweep() -> _FunctionABI:
+        builder = _FunctionABIBuilder("SWEEP")
+        return builder.add_address("token").add_address("recipient").add_int("amountMin").build()
+
+    @staticmethod
+    def _build_pay_portion() -> _FunctionABI:
+        builder = _FunctionABIBuilder("PAY_PORTION")
+        return builder.add_address("token").add_address("recipient").add_int("bips").build()

ape_uniswap/_constants.py

@@ -0,0 +1,45 @@
+"""
+Constants used by the Uniswap Universal Router Codec
+
+* Author: Elnaril (https://www.fiverr.com/elnaril, https://github.com/Elnaril).
+* License: MIT.
+* Doc: https://github.com/Elnaril/uniswap-universal-router-decoder
+"""
+from typing import (
+    Any,
+    Dict,
+)
+
+from eth_typing import HexStr
+
+
+_execution_function_input_types = ["bytes", "bytes[]", "int"]
+_execution_function_selector = HexStr("0x3593564c")
+_router_abi = '[{"inputs":[{"components":[{"internalType":"address","name":"permit2","type":"address"},{"internalType":"address","name":"weth9","type":"address"},{"internalType":"address","name":"seaportV1_5","type":"address"},{"internalType":"address","name":"seaportV1_4","type":"address"},{"internalType":"address","name":"openseaConduit","type":"address"},{"internalType":"address","name":"nftxZap","type":"address"},{"internalType":"address","name":"x2y2","type":"address"},{"internalType":"address","name":"foundation","type":"address"},{"internalType":"address","name":"sudoswap","type":"address"},{"internalType":"address","name":"elementMarket","type":"address"},{"internalType":"address","name":"nft20Zap","type":"address"},{"internalType":"address","name":"cryptopunks","type":"address"},{"internalType":"address","name":"looksRareV2","type":"address"},{"internalType":"address","name":"routerRewardsDistributor","type":"address"},{"internalType":"address","name":"looksRareRewardsDistributor","type":"address"},{"internalType":"address","name":"looksRareToken","type":"address"},{"internalType":"address","name":"v2Factory","type":"address"},{"internalType":"address","name":"v3Factory","type":"address"},{"internalType":"bytes32","name":"pairInitCodeHash","type":"bytes32"},{"internalType":"bytes32","name":"poolInitCodeHash","type":"bytes32"}],"internalType":"struct RouterParameters","name":"params","type":"tuple"}],"stateMutability":"nonpayable","type":"constructor"},{"inputs":[],"name":"BalanceTooLow","type":"error"},{"inputs":[],"name":"BuyPunkFailed","type":"error"},{"inputs":[],"name":"ContractLocked","type":"error"},{"inputs":[],"name":"ETHNotAccepted","type":"error"},{"inputs":[{"internalType":"uint256","name":"commandIndex","type":"uint256"},{"internalType":"bytes","name":"message","type":"bytes"}],"name":"ExecutionFailed","type":"error"},{"inputs":[],"name":"FromAddressIsNotOwner","type":"error"},{"inputs":[],"name":"InsufficientETH","type":"error"},{"inputs":[],"name":"InsufficientToken","type":"error"},{"inputs":[],"name":"InvalidBips","type":"error"},{"inputs":[{"internalType":"uint256","name":"commandType","type":"uint256"}],"name":"InvalidCommandType","type":"error"},{"inputs":[],"name":"InvalidOwnerERC1155","type":"error"},{"inputs":[],"name":"InvalidOwnerERC721","type":"error"},{"inputs":[],"name":"InvalidPath","type":"error"},{"inputs":[],"name":"InvalidReserves","type":"error"},{"inputs":[],"name":"InvalidSpender","type":"error"},{"inputs":[],"name":"LengthMismatch","type":"error"},{"inputs":[],"name":"SliceOutOfBounds","type":"error"},{"inputs":[],"name":"TransactionDeadlinePassed","type":"error"},{"inputs":[],"name":"UnableToClaim","type":"error"},{"inputs":[],"name":"UnsafeCast","type":"error"},{"inputs":[],"name":"V2InvalidPath","type":"error"},{"inputs":[],"name":"V2TooLittleReceived","type":"error"},{"inputs":[],"name":"V2TooMuchRequested","type":"error"},{"inputs":[],"name":"V3InvalidAmountOut","type":"error"},{"inputs":[],"name":"V3InvalidCaller","type":"error"},{"inputs":[],"name":"V3InvalidSwap","type":"error"},{"inputs":[],"name":"V3TooLittleReceived","type":"error"},{"inputs":[],"name":"V3TooMuchRequested","type":"error"},{"anonymous":false,"inputs":[{"indexed":false,"internalType":"uint256","name":"amount","type":"uint256"}],"name":"RewardsSent","type":"event"},{"inputs":[{"internalType":"bytes","name":"looksRareClaim","type":"bytes"}],"name":"collectRewards","outputs":[],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"bytes","name":"commands","type":"bytes"},{"internalType":"bytes[]","name":"inputs","type":"bytes[]"}],"name":"execute","outputs":[],"stateMutability":"payable","type":"function"},{"inputs":[{"internalType":"bytes","name":"commands","type":"bytes"},{"internalType":"bytes[]","name":"inputs","type":"bytes[]"},{"internalType":"uint256","name":"deadline","type":"uint256"}],"name":"execute","outputs":[],"stateMutability":"payable","type":"function"},{"inputs":[{"internalType":"address","name":"","type":"address"},{"internalType":"address","name":"","type":"address"},{"internalType":"uint256[]","name":"","type":"uint256[]"},{"internalType":"uint256[]","name":"","type":"uint256[]"},{"internalType":"bytes","name":"","type":"bytes"}],"name":"onERC1155BatchReceived","outputs":[{"internalType":"bytes4","name":"","type":"bytes4"}],"stateMutability":"pure","type":"function"},{"inputs":[{"internalType":"address","name":"","type":"address"},{"internalType":"address","name":"","type":"address"},{"internalType":"uint256","name":"","type":"uint256"},{"internalType":"uint256","name":"","type":"uint256"},{"internalType":"bytes","name":"","type":"bytes"}],"name":"onERC1155Received","outputs":[{"internalType":"bytes4","name":"","type":"bytes4"}],"stateMutability":"pure","type":"function"},{"inputs":[{"internalType":"address","name":"","type":"address"},{"internalType":"address","name":"","type":"address"},{"internalType":"uint256","name":"","type":"uint256"},{"internalType":"bytes","name":"","type":"bytes"}],"name":"onERC721Received","outputs":[{"internalType":"bytes4","name":"","type":"bytes4"}],"stateMutability":"pure","type":"function"},{"inputs":[{"internalType":"bytes4","name":"interfaceId","type":"bytes4"}],"name":"supportsInterface","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"pure","type":"function"},{"inputs":[{"internalType":"int256","name":"amount0Delta","type":"int256"},{"internalType":"int256","name":"amount1Delta","type":"int256"},{"internalType":"bytes","name":"data","type":"bytes"}],"name":"uniswapV3SwapCallback","outputs":[],"stateMutability":"nonpayable","type":"function"},{"stateMutability":"payable","type":"receive"}]'  # noqa
+
+_structured_data_permit: Dict[str, Any] = {
+    'types': {
+        'EIP712Domain': [
+            {'name': 'name', 'type': 'string'},
+            {'name': 'chainId', 'type': 'uint256'},
+            {'name': 'verifyingContract', 'type': 'address'}
+        ],
+        'PermitDetails': [
+            {'name': 'token', 'type': 'address'},
+            {'name': 'amount', 'type': 'uint160'},
+            {'name': 'expiration', 'type': 'uint48'},
+            {'name': 'nonce', 'type': 'uint48'},
+        ],
+        'PermitSingle': [
+            {'name': 'details', 'type': 'PermitDetails'},
+            {'name': 'spender', 'type': 'address'},
+            {'name': 'sigDeadline', 'type': 'uint256'},
+        ],
+    },
+    'primaryType': 'PermitSingle',
+    'domain': {
+        'name': 'Permit2',
+        'chainId': 1,
+        'verifyingContract': '0x000000000022D473030F116dDEE9F6B43aC78BA3',
+    },
+}

ape_uniswap/_decoder.py

@@ -0,0 +1,108 @@
+"""
+Decoding part of the Uniswap Universal Router Codec
+
+* Author: Elnaril (https://www.fiverr.com/elnaril, https://github.com/Elnaril).
+* License: MIT.
+* Doc: https://github.com/Elnaril/uniswap-universal-router-decoder
+"""
+from itertools import chain
+from typing import (
+    Any,
+    Dict,
+    List,
+    Tuple,
+    Union,
+)
+
+from ethpm_types import HexBytes
+
+from ._abi_builder import _ABIMap
+from ._constants import _router_abi
+from ._enums import _RouterFunction
+
+from ape.utils import ManagerAccessMixin
+from eth_utils import to_checksum_address
+from eth_typing import AnyAddress, ChecksumAddress, HexStr
+from ape.api import (
+    ReceiptAPI
+)
+
+
+class _Decoder(ManagerAccessMixin):
+    def __init__(self, abi_map: _ABIMap, router_address: Union[AnyAddress, str, bytes]) -> None:
+        # self._router_contract = self._w3.eth.contract(abi=_router_abi)
+        self._router_contract = self.chain_manager.contracts.instance_at(
+            to_checksum_address(router_address), abi=_router_abi)
+        self._abi_map = abi_map
+
+    def function_input(self, input_data: Union[HexStr, HexBytes]) -> Tuple[str, Dict[str, Any]]:
+        """
+        Decode the data sent to an UR function
+
+        :param input_data: the transaction 'input' data
+        :return: The decoded data if the function has been implemented.
+        """
+        fct_name, decoded_input = self._router_contract.decode_input(input_data)
+        command = decoded_input["commands"]
+        command_input = decoded_input["inputs"]
+        decoded_command_input = []
+        for i, b in enumerate(command[-7:]):
+            # iterating over bytes produces integers
+            try:
+                abi_mapping = self._abi_map[_RouterFunction(b)]
+                data = abi_mapping.selector + command_input[i]
+                # sub_contract = self._w3.eth.contract(abi=abi_mapping.fct_abi.get_full_abi())
+                decoded_command_input.append(self._router_contract.decode_input(data))
+            except (ValueError, KeyError):
+                decoded_command_input.append(command_input[i].hex())
+        decoded_input["inputs"] = decoded_command_input
+        return fct_name, decoded_input
+
+    def transaction(self, trx_hash: Union[HexBytes, HexStr]) -> Dict[str, Any]:
+        """
+        Get transaction details and decode the data used to call a UR function.
+
+        ⚠ To use this method, the decoder must be built with a Web3 instance or a rpc endpoint address.
+
+        :param trx_hash: the hash of the transaction sent to the UR
+        :return: the transaction as a dict with the additional 'decoded_input' field
+        """
+        trx = self._get_transaction(trx_hash)
+        fct_name, decoded_input = self.function_input(trx.transaction.data)
+        result_trx = dict(trx.transaction)
+        result_trx["decoded_input"] = decoded_input
+        return result_trx
+
+    def _get_transaction(self, trx_hash: Union[HexBytes, HexStr]) -> ReceiptAPI:
+        return self.chain_manager.get_receipt(trx_hash)
+
+    @staticmethod
+    def v3_path(v3_fn_name: str, path: Union[bytes, str]) -> Tuple[Union[int, ChecksumAddress], ...]:
+        """
+        Decode a V3 router path
+
+        :param v3_fn_name: V3_SWAP_EXACT_IN or V3_SWAP_EXACT_OUT only
+        :param path: the V3 path as returned by decode_function_input() or decode_transaction()
+        :return: a tuple of token addresses separated by the corresponding pool fees, first token being the 'in-token',
+        last token being the 'out-token'
+        """
+        valid_fn_names = ("V3_SWAP_EXACT_IN", "V3_SWAP_EXACT_OUT")
+        if v3_fn_name.upper() not in valid_fn_names:
+            raise ValueError(f"v3_fn_name must be in {valid_fn_names}")
+        path_str = path.hex() if isinstance(path, bytes) else str(path)
+        path_str = path_str[2:] if path_str.startswith("0x") else path_str
+        path_list: List[Union[int, ChecksumAddress]] = [to_checksum_address(path_str[0:40]), ]
+        parsed_remaining_path: List[List[Union[int, ChecksumAddress]]] = [
+            [
+                int(path_str[40:][i:i + 6], 16),
+                to_checksum_address(path_str[40:][i + 6:i + 46]),
+            ]
+            for i in range(0, len(path_str[40:]), 46)
+        ]
+        path_list.extend(list(chain.from_iterable(parsed_remaining_path)))
+
+        if v3_fn_name.upper() == "V3_SWAP_EXACT_OUT":
+            path_list.reverse()
+
+        return tuple(path_list)
+