add contracts module

cow_py/contracts/domain.py

@@ -0,0 +1,29 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from cow_py.common.chains import Chain
+
+
+@dataclass
+class TypedDataDomain:
+    name: str
+    version: str
+    chainId: int
+    verifyingContract: str
+    salt: Optional[str] = None
+
+
+def domain(chain: Chain, verifying_contract: str) -> TypedDataDomain:
+    """
+    Return the Gnosis Protocol v2 domain used for signing.
+
+    :param chain: The EIP-155 chain ID.
+    :param verifying_contract: The address of the contract that will verify the signature.
+    :return: An EIP-712 compatible typed domain data.
+    """
+    return TypedDataDomain(
+        name="Gnosis Protocol",
+        version="v2",
+        chainId=chain.chain_id,
+        verifyingContract=verifying_contract,
+    )

cow_py/contracts/order.py

@@ -0,0 +1,243 @@
+from dataclasses import dataclass
+from enum import Enum
+from typing import Literal, Optional, Union
+
+from eth_account.messages import _hash_eip191_message, encode_typed_data
+from eth_typing import Hash32, HexStr
+from eth_utils.conversions import to_bytes, to_hex
+
+
+@dataclass
+class Order:
+    # Sell token address.
+    sellToken: str
+    # Buy token address.
+    buyToken: str
+    # An optional address to receive the proceeds of the trade instead of the
+    # owner (i.e. the order signer).
+    receiver: str
+    # The order sell amount.
+    #
+    # For fill or kill sell orders, this amount represents the exact sell amount
+    # that will be executed in the trade. For fill or kill buy orders, this
+    # amount represents the maximum sell amount that can be executed. For partial
+    # fill orders, this represents a component of the limit price fraction.
+    #
+    sellAmount: int
+    # The order buy amount.
+    #
+    # For fill or kill sell orders, this amount represents the minimum buy amount
+    # that can be executed in the trade. For fill or kill buy orders, this amount
+    # represents the exact buy amount that will be executed. For partial fill
+    # orders, this represents a component of the limit price fraction.
+    #
+    buyAmount: int
+    # The timestamp this order is valid until
+    validTo: int
+    # Arbitrary application specific data that can be added to an order. This can
+    # also be used to ensure uniqueness between two orders with otherwise the
+    # exact same parameters.
+    appData: str
+    # Fee to give to the protocol.
+    feeAmount: int
+    # The order kind.
+    kind: str
+    # Specifies whether or not the order is partially fillable.
+    partiallyFillable: bool = False
+    # Specifies how the sell token balance will be withdrawn. It can either be
+    # taken using ERC20 token allowances made directly to the Vault relayer
+    # (default) or using Balancer Vault internal or external balances.
+    sellTokenBalance: Optional[str] = None
+    # Specifies how the buy token balance will be paid. It can either be paid
+    # directly in ERC20 tokens (default) in Balancer Vault internal balances.
+    buyTokenBalance: Optional[str] = None
+
+
+# Gnosis Protocol v2 order cancellation data.
+@dataclass
+class OrderCancellations:
+    orderUids: bytearray
+
+
+# Marker address to indicate that an order is buying Ether.
+#
+# Note that this address is only has special meaning in the `buyToken` and will
+# be treated as a ERC20 token address in the `sellToken` position, causing the
+# settlement to revert.
+BUY_ETH_ADDRESS = "0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE"
+
+BytesLike = Union[str, bytes, HexStr]
+
+HashLike = Union[BytesLike, int]
+
+
+class OrderKind(Enum):
+    SELL = "sell"
+    BUY = "buy"
+
+
+class OrderBalance(Enum):
+    # Use ERC20 token balances.
+    ERC20 = "erc20"
+    # Use Balancer Vault external balances.
+
+    # This can only be specified specified for the sell balance and allows orders
+    # to re-use Vault ERC20 allowances. When specified for the buy balance, it
+    # will be treated as {@link OrderBalance.ERC20}.
+    EXTERNAL = "external"
+    # Use Balancer Vault internal balances.
+    INTERNAL = "internal"
+
+
+# /**
+#  * The EIP-712 type fields definition for a Gnosis Protocol v2 order.
+#  */
+ORDER_TYPE_FIELDS = [
+    dict(name="sellToken", type="address"),
+    dict(name="buyToken", type="address"),
+    dict(name="receiver", type="address"),
+    dict(name="sellAmount", type="uint256"),
+    dict(name="buyAmount", type="uint256"),
+    dict(name="validTo", type="uint32"),
+    dict(name="appData", type="bytes32"),
+    dict(name="feeAmount", type="uint256"),
+    dict(name="kind", type="string"),
+    dict(name="partiallyFillable", type="bool"),
+    dict(name="sellTokenBalance", type="string"),
+    dict(name="buyTokenBalance", type="string"),
+]
+# The EIP-712 type fields definition for a Gnosis Protocol v2 order.
+CANCELLATIONS_TYPE_FIELDS = [
+    dict(name="orderUids", type="bytes[]"),
+]
+
+
+def hashify(h: Union[int, str, bytes]) -> str:
+    """
+    Normalizes an app data value to a 32-byte hash.
+    :param h: A hash-like value to normalize. Can be an integer, hexadecimal string, or bytes.
+    :return: A 32-byte hash encoded as a hex-string.
+    """
+    if isinstance(h, int):
+        # Convert the integer to a hexadecimal string and pad it to 64 characters
+        return f"0x{h:064x}"
+    elif isinstance(h, str):
+        # Convert string to bytes, then pad it to 32 bytes (64 hex characters)
+        return to_hex(to_bytes(hexstr=h).rjust(32, b"\0"))
+    elif isinstance(h, bytes):
+        # Pad the bytes to 32 bytes (64 hex characters)
+        return to_hex(h.rjust(32, b"\0"))
+    else:
+        raise ValueError("Input must be an integer, a hexadecimal string, or bytes.")
+
+
+def normalize_buy_token_balance(
+    balance: Optional[str],
+) -> Literal["erc20", "internal"]:
+    """
+    Normalizes the balance configuration for a buy token.
+
+    :param balance: The balance configuration.
+    :return: The normalized balance configuration.
+    """
+    if balance in [None, OrderBalance.ERC20.value, OrderBalance.EXTERNAL.value]:
+        return OrderBalance.ERC20.value
+    elif balance == OrderBalance.INTERNAL:
+        return OrderBalance.INTERNAL.value
+    else:
+        raise ValueError(f"Invalid order balance {balance}")
+
+
+ZERO_ADDRESS = "0x" + "00" * 20
+
+
+def normalize_order(order: Order):
+    if order.receiver == ZERO_ADDRESS:
+        raise ValueError("receiver cannot be address(0)")
+
+    return {
+        "sellToken": order.sellToken,
+        "buyToken": order.buyToken,
+        "receiver": order.receiver if order.receiver else ZERO_ADDRESS,
+        "sellAmount": order.sellAmount,
+        "buyAmount": order.buyAmount,
+        "validTo": order.validTo,
+        "appData": hashify(order.appData),
+        "feeAmount": order.feeAmount,
+        "kind": order.kind,
+        "partiallyFillable": order.partiallyFillable,
+        "sellTokenBalance": (
+            order.sellTokenBalance
+            if order.sellTokenBalance
+            else OrderBalance.ERC20.value
+        ),
+        "buyTokenBalance": normalize_buy_token_balance(order.buyTokenBalance),
+    }
+
+
+def hash_typed_data(domain, types, data) -> Hash32:
+    """
+    Compute the 32-byte signing hash for the specified order.
+
+    :param domain: The EIP-712 domain separator to compute the hash for.
+    :param types: The typed data types.
+    :param data: The data to compute the digest for.
+    :return: Hex-encoded 32-byte order digest.
+    """
+    encoded_data = encode_typed_data(
+        domain_data=domain, message_types=types, message_data=data
+    )
+    return _hash_eip191_message(encoded_data)
+
+
+def hash_order(domain, order):
+    """
+    Compute the 32-byte signing hash for the specified order.
+
+    :param domain: The EIP-712 domain separator to compute the hash for.
+    :param order: The order to compute the digest for.
+    :return: Hex-encoded 32-byte order digest.
+    """
+    return hash_typed_data(domain, ORDER_TYPE_FIELDS, normalize_order(order))
+
+
+def hash_order_cancellation(domain, order_uid) -> str:
+    """
+    Compute the 32-byte signing hash for the specified cancellation.
+
+    :param domain: The EIP-712 domain separator to compute the hash for.
+    :param order_uid: The unique identifier of the order to cancel.
+    :return: Hex-encoded 32-byte order digest.
+    """
+    return hash_order_cancellations(domain, [order_uid])
+
+
+def hash_order_cancellations(domain_data, order_uids) -> str:
+    """
+    Compute the 32-byte signing hash for the specified order cancellations.
+
+    :param domain_data: The EIP-712 domain separator to compute the hash for.
+    :param order_uids: The unique identifiers of the orders to cancel.
+    :return: Hex-encoded 32-byte order digest.
+    """
+    return _hash_eip191_message(
+        encode_typed_data(
+            domain_data,
+            message_types={"OrderCancellations": CANCELLATIONS_TYPE_FIELDS},
+            message_data={"orderUids": order_uids},
+        )
+    ).hex()
+
+
+# The byte length of an order UID.
+ORDER_UID_LENGTH = 56
+
+
+@dataclass
+class OrderUidParams:
+    # The EIP-712 order struct hash.
+    orderDigest: str
+    # The owner of the order.
+    owner: str
+    # The timestamp this order is valid until.
+    validTo: int

cow_py/contracts/sign.py

@@ -0,0 +1,104 @@
+from enum import IntEnum
+from typing import List, NamedTuple, Union
+
+from eth_account import Account
+from eth_account.datastructures import SignedMessage
+from eth_account.signers.local import LocalAccount
+from eth_utils.conversions import to_hex
+from eth_utils.crypto import keccak
+from web3 import Web3
+
+from cow_py.contracts.order import (
+    CANCELLATIONS_TYPE_FIELDS,
+    ORDER_TYPE_FIELDS,
+    Order,
+    hash_typed_data,
+    normalize_order,
+)
+
+EIP1271_MAGICVALUE = to_hex(keccak(text="isValidSignature(bytes32,bytes)"))[:10]
+PRE_SIGNED = to_hex(keccak(text="GPv2Signing.Scheme.PreSign"))
+
+
+class SigningScheme(IntEnum):
+    # The EIP-712 typed data signing scheme. This is the preferred scheme as it
+    # provides more infomation to wallets performing the signature on the data
+    # being signed.
+    #
+    # https://github.com/ethereum/EIPs/blob/master/EIPS/eip-712.md#definition-of-domainseparator
+    EIP712 = 0b00
+    # Message signed using eth_sign RPC call.
+    ETHSIGN = 0b01
+    # Smart contract signatures as defined in EIP-1271.
+    EIP1271 = 0b10
+    # Pre-signed order.
+    PRESIGN = 0b11
+
+
+class EcdsaSignature(NamedTuple):
+    scheme: SigningScheme
+    data: str
+
+
+class Eip1271SignatureData(NamedTuple):
+    verifier: str
+    signature: bytes
+
+
+class Eip1271Signature(NamedTuple):
+    scheme: SigningScheme
+    data: Eip1271SignatureData
+
+
+class PreSignSignature(NamedTuple):
+    scheme: SigningScheme
+    data: str
+
+
+Signature = Union[EcdsaSignature, Eip1271Signature, PreSignSignature]
+
+
+def ecdsa_sign_typed_data(
+    scheme, owner: LocalAccount, domain_data, message_types, message_data
+) -> SignedMessage:
+    return Account._sign_hash(
+        hash_typed_data(domain_data, message_types, message_data), owner.key
+    )
+
+
+def sign_order(
+    domain, order: Order, owner: LocalAccount, scheme: SigningScheme
+) -> EcdsaSignature:
+    normalized_order = normalize_order(order)
+    signed_data = ecdsa_sign_typed_data(
+        scheme, owner, domain, {"Order": ORDER_TYPE_FIELDS}, normalized_order
+    )
+    return EcdsaSignature(
+        scheme=scheme,
+        data=signed_data.signature.hex(),
+    )
+
+
+def sign_order_cancellation(domain, order_uid: Union[str, bytes], owner, scheme):
+    return sign_order_cancellations(domain, [order_uid], owner, scheme)
+
+
+def sign_order_cancellations(
+    domain, order_uids: List[Union[str, bytes]], owner, scheme
+):
+    data = {"orderUids": order_uids}
+    types = {"OrderCancellations": CANCELLATIONS_TYPE_FIELDS}
+
+    signed_data = ecdsa_sign_typed_data(scheme, owner, domain, types, data)
+
+    return EcdsaSignature(scheme=scheme, data=signed_data.signature.hex())
+
+
+def encode_eip1271_signature_data(verifier, signature):
+    return Web3.solidity_keccak(["address", "bytes"], [verifier, signature])
+
+
+def decode_eip1271_signature_data(signature):
+    arrayified_signature = bytes.fromhex(signature[2:])  # Removing '0x'
+    verifier = Web3.to_checksum_address(arrayified_signature[:20].hex())
+    return Eip1271SignatureData(verifier, arrayified_signature[20:])

cow_py/web3/provider.py

@@ -1,6 +1,7 @@
 from typing import Dict
 
 import web3
+
 from cow_py.common.chains import Chain
 
 DEFAULT_PROVIDER_NETWORK_MAPPING = {