Stop using eip712_structs

- Copy library from https://github.com/jvinet/eip712

docs/source/gnosis.eth.eip712.rst

@@ -0,0 +1,10 @@
+gnosis.eth.eip712 package
+============================
+
+Module contents
+---------------
+
+.. automodule:: gnosis.eth.eip712
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/source/quickstart.rst

@@ -76,6 +76,46 @@ gnosis.eth.constants
 - ``SENTINEL_ADDRESS (0x000...1)``: Used for Gnosis Safe's linked lists (modules, owners...).
 - Maximum an minimum values for `R`, `S` and `V` in ethereum signatures.
 
+gnosis.eth.eip712
+~~~~~~~~~~~~~~~~~~~~
+.. code-block:: python
+    from gnosis.eth.eip712 import eip712_encode_hash
+
+    types = {'EIP712Domain': [{'name': 'name', 'type': 'string'},
+                              {'name': 'version', 'type': 'string'},
+                              {'name': 'chainId', 'type': 'uint256'},
+                              {'name': 'verifyingContract', 'type': 'address'}],
+             'Mailbox': [{'name': 'owner', 'type': 'address'},
+                         {'name': 'messages', 'type': 'Message[]'}],
+             'Message': [{'name': 'sender', 'type': 'address'},
+                         {'name': 'subject', 'type': 'string'},
+                         {'name': 'isSpam', 'type': 'bool'},
+                         {'name': 'body', 'type': 'string'}]}
+
+    msgs = [{'sender': ADDRESS,
+             'subject': 'Hello World',
+             'body': 'The sparrow flies at midnight.',
+             'isSpam': False},
+            {'sender': ADDRESS,
+             'subject': 'You may have already Won! :dumb-emoji:',
+             'body': 'Click here for sweepstakes!',
+             'isSpam': True}]
+
+    mailbox = {'owner': ADDRESS,
+               'messages': msgs}
+
+    payload = {'types': types,
+               'primaryType': 'Mailbox',
+               'domain': {'name': 'MyDApp',
+                          'version': '3.0',
+                          'chainId': 41,
+                          'verifyingContract': ADDRESS},
+               'message': mailbox}
+
+    eip712_hash = eip712_encode_hash(payload)
+
+
+
 gnosis.eth.oracles
 ~~~~~~~~~~~~~~~~~~
 Price oracles for Uniswap, UniswapV2, Kyber, SushiSwap, Aave, Balancer, Curve, Mooniswap, Yearn...

gnosis/eth/eip712/__init__.py

@@ -0,0 +1,178 @@
+"""
+Based on https://github.com/jvinet/eip712, adjustments by https://github.com/uxio0
+
+Routines for EIP712 encoding and signing.
+
+Copyright (C) 2022 Judd Vinet <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+of the Software, and to permit persons to whom the Software is furnished to do
+so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+"""
+import re
+from typing import Any, Dict, List, Union
+
+from eth_abi import encode_abi
+from eth_account import Account
+from eth_typing import Hash32, HexStr
+
+from ..utils import fast_keccak
+
+
+def encode_data(primary_type: str, data, types):
+    """
+    Encode structured data as per Ethereum's signTypeData_v4.
+
+    https://docs.metamask.io/guide/signing-data.html#sign-typed-data-v4
+
+    This code is ported from the Javascript "eth-sig-util" package.
+    """
+    encoded_types = ["bytes32"]
+    encoded_values = [hash_type(primary_type, types)]
+
+    def _encode_field(name, typ, value):
+        if typ in types:
+            if value is None:
+                return [
+                    "bytes32",
+                    "0x0000000000000000000000000000000000000000000000000000000000000000",
+                ]
+            else:
+                return ["bytes32", fast_keccak(encode_data(typ, value, types))]
+
+        if value is None:
+            raise Exception(f"Missing value for field {name} of type {type}")
+
+        if typ == "bytes":
+            return ["bytes32", fast_keccak(value)]
+
+        if typ == "string":
+            # Convert string to bytes.
+            value = value.encode("utf-8")
+            return ["bytes32", fast_keccak(value)]
+
+        if typ.endswith("]"):
+            parsed_type = typ[:-2]
+            type_value_pairs = dict(
+                [_encode_field(name, parsed_type, v) for v in value]
+            )
+            h = fast_keccak(
+                encode_abi(
+                    list(type_value_pairs.keys()), list(type_value_pairs.values())
+                )
+            )
+            return ["bytes32", h]
+
+        return [typ, value]
+
+    for field in types[primary_type]:
+        typ, val = _encode_field(field["name"], field["type"], data[field["name"]])
+        encoded_types.append(typ)
+        encoded_values.append(val)
+
+    return encode_abi(encoded_types, encoded_values)
+
+
+def encode_type(primary_type: str, types) -> str:
+    result = ""
+    deps = find_type_dependencies(primary_type, types)
+    deps = sorted([d for d in deps if d != primary_type])
+    deps = [primary_type] + deps
+    for typ in deps:
+        children = types[typ]
+        if not children:
+            raise Exception(f"No type definition specified: {type}")
+
+        defs = [f"{t['type']} {t['name']}" for t in types[typ]]
+        result += typ + "(" + ",".join(defs) + ")"
+    return result
+
+
+def find_type_dependencies(primary_type: str, types, results=None):
+    if results is None:
+        results = []
+
+    primary_type = re.split(r"\W", primary_type)[0]
+    if primary_type in results or not types.get(primary_type):
+        return results
+    results.append(primary_type)
+
+    for field in types[primary_type]:
+        deps = find_type_dependencies(field["type"], types, results)
+        for dep in deps:
+            if dep not in results:
+                results.append(dep)
+
+    return results
+
+
+def hash_type(primary_type: str, types) -> Hash32:
+    return fast_keccak(encode_type(primary_type, types).encode())
+
+
+def hash_struct(primary_type: str, data, types) -> Hash32:
+    return fast_keccak(encode_data(primary_type, data, types))
+
+
+def eip712_encode(typed_data: Dict[str, Any]) -> List[bytes]:
+    """
+    Given a dict of structured data and types, return a 3-element list of
+    the encoded, signable data.
+
+      0: The magic & version (0x1901)
+      1: The encoded types
+      2: The encoded data
+    """
+    parts = [
+        bytes.fromhex("1901"),
+        hash_struct("EIP712Domain", typed_data["domain"], typed_data["types"]),
+    ]
+    if typed_data["primaryType"] != "EIP712Domain":
+        parts.append(
+            hash_struct(
+                typed_data["primaryType"], typed_data["message"], typed_data["types"]
+            )
+        )
+    return parts
+
+
+def eip712_encode_hash(typed_data: Dict[str, Any]) -> Hash32:
+    """
+    :param typed_data: EIP712 structured data and types
+    :return: Keccak256 hash of encoded signable data
+    """
+    return fast_keccak(b"".join(eip712_encode(typed_data)))
+
+
+def eip712_signature(
+    payload: Dict[str, Any], private_key: Union[HexStr, bytes]
+) -> bytes:
+    """
+    Given a bytes object and a private key, return a signature suitable for
+    EIP712 and EIP191 messages.
+    """
+    if isinstance(payload, (list, tuple)):
+        payload = b"".join(payload)
+
+    if isinstance(private_key, str) and private_key.startswith("0x"):
+        private_key = private_key[2:]
+    elif isinstance(private_key, bytes):
+        private_key = bytes.hex()
+
+    account = Account.from_key(private_key)
+    hashed_payload = fast_keccak(payload)
+    return account.signHash(hashed_payload)["signature"]

gnosis/protocol/gnosis_protocol_api.py

@@ -2,14 +2,13 @@
 from typing import Any, Dict, List, Optional, TypedDict, Union, cast
 
 import requests
-from eip712_structs import make_domain
 from eth_account import Account
 from eth_account.messages import encode_defunct
 from eth_typing import AnyAddress, ChecksumAddress, HexStr
 from hexbytes import HexBytes
-from web3 import Web3
 
 from gnosis.eth import EthereumNetwork, EthereumNetworkNotSupported
+from gnosis.eth.eip712 import eip712_encode_hash
 
 from .order import Order, OrderKind
 
@@ -42,26 +41,28 @@ class GnosisProtocolAPI:
     Client for GnosisProtocol API. More info: https://docs.cowswap.exchange/
     """
 
-    settlement_contract_addresses = {
+    SETTLEMENT_CONTRACT_ADDRESSES = {
         EthereumNetwork.MAINNET: "0x9008D19f58AAbD9eD0D60971565AA8510560ab41",
         EthereumNetwork.GOERLI: "0x9008D19f58AAbD9eD0D60971565AA8510560ab41",
         EthereumNetwork.XDAI: "0x9008D19f58AAbD9eD0D60971565AA8510560ab41",
     }
 
-    api_base_urls = {
+    API_BASE_URLS = {
         EthereumNetwork.MAINNET: "https://api.cow.fi/mainnet/api/v1/",
         EthereumNetwork.GOERLI: "https://api.cow.fi/goerli/api/v1/",
         EthereumNetwork.XDAI: "https://api.cow.fi/xdai/api/v1/",
     }
 
     def __init__(self, ethereum_network: EthereumNetwork):
         self.network = ethereum_network
-        if self.network not in self.api_base_urls:
+        if self.network not in self.API_BASE_URLS:
             raise EthereumNetworkNotSupported(
                 f"{self.network.name} network not supported by Gnosis Protocol"
             )
-        self.domain_separator = self.build_domain_separator(self.network)
-        self.base_url = self.api_base_urls[self.network]
+        self.settlement_contract_address = self.SETTLEMENT_CONTRACT_ADDRESSES[
+            self.network
+        ]
+        self.base_url = self.API_BASE_URLS[self.network]
         self.http_session = requests.Session()
 
     @cached_property
@@ -76,24 +77,15 @@ def weth_address(self) -> ChecksumAddress:
         else:  # XDAI
             return ChecksumAddress("0x6A023CCd1ff6F2045C3309768eAd9E68F978f6e1")
 
-    @classmethod
-    def build_domain_separator(cls, ethereum_network: EthereumNetwork):
-        return make_domain(
-            name="Gnosis Protocol",
-            version="v2",
-            chainId=str(ethereum_network.value),
-            verifyingContract=cls.settlement_contract_addresses[ethereum_network],
-        )
-
     def get_fee(self, order: Order) -> int:
-        if order["kind"] == "sell":
-            amount = order["sellAmount"]
+        if order.is_sell_order():
+            amount = order.sellAmount
         else:
-            amount = order["buyAmount"]
+            amount = order.buyAmount
         url = (
             self.base_url
-            + f'fee/?sellToken={order["sellToken"]}&buyToken={order["buyToken"]}'
-            f'&amount={amount}&kind={order["kind"]}'
+            + f"fee/?sellToken={order.sellToken}&buyToken={order.buyToken}"
+            f"&amount={amount}&kind={order.kind}"
         )
         result = self.http_session.get(url).json()
         if "amount" in result:
@@ -110,28 +102,31 @@ def place_order(
         :return: UUID for the order as an hex hash
         """
         assert (
-            order["buyAmount"] and order["sellAmount"]
+            order.buyAmount and order.sellAmount
         ), "Order buyAmount and sellAmount cannot be empty"
 
         url = self.base_url + "orders/"
-        order["feeAmount"] = order["feeAmount"] or self.get_fee(order)
-        signable_bytes = order.signable_bytes(domain=self.domain_separator)
-        signable_hash = Web3.keccak(signable_bytes)
+        order.feeAmount = order.feeAmount or self.get_fee(order)
+        signable_hash = eip712_encode_hash(
+            order.get_eip712_structured_data(
+                self.network.value, self.settlement_contract_address
+            )
+        )
         message = encode_defunct(primitive=signable_hash)
         signed_message = Account.from_key(private_key).sign_message(message)
 
         data_json = {
-            "sellToken": order["sellToken"].lower(),
-            "buyToken": order["buyToken"].lower(),
-            "sellAmount": str(order["sellAmount"]),
-            "buyAmount": str(order["buyAmount"]),
-            "validTo": order["validTo"],
-            "appData": HexBytes(order["appData"]).hex()
-            if isinstance(order["appData"], bytes)
-            else order["appData"],
-            "feeAmount": str(order["feeAmount"]),
-            "kind": order["kind"],
-            "partiallyFillable": order["partiallyFillable"],
+            "sellToken": order.sellToken.lower(),
+            "buyToken": order.buyToken.lower(),
+            "sellAmount": str(order.sellAmount),
+            "buyAmount": str(order.buyAmount),
+            "validTo": order.validTo,
+            "appData": HexBytes(order.appData).hex()
+            if isinstance(order.appData, bytes)
+            else order.appData,
+            "feeAmount": str(order.feeAmount),
+            "kind": order.kind,
+            "partiallyFillable": order.partiallyFillable,
             "signature": signed_message.signature.hex(),
             "signingScheme": "ethsign",
             "from": Account.from_key(private_key).address,