eip712 signature requests for pydantic messages

Author: devinaconley

framelib/__init__.py

@@ -4,10 +4,10 @@
 
 from .frame import frame, message, error
 from .hub import validate_message, validate_message_or_mock
-from .models import FrameMessage, ValidatedMessage, User
+from .models import FrameMessage, ValidatedMessage, User, Address, Bytes, Bytes32
 from .warpcast import get_user
 from .neynar import (
     validate_message as validate_message_neynar,
     validate_message_or_mock as validate_message_or_mock_neynar
 )
-from .transaction import transaction, mint
+from .transaction import transaction, mint, signature

framelib/models.py

@@ -4,7 +4,8 @@
 
 import datetime
 from typing import Optional, Literal
-from pydantic import BaseModel
+from pydantic import BaseModel, SerializeAsAny
+from eth_utils import is_address
 
 
 # ---- frame message ----
@@ -46,8 +47,8 @@ class FrameMessage(BaseModel):
 class EthTransactionParams(BaseModel):
     abi: list[dict]
     to: str
-    value: Optional[str]
-    data: Optional[str]
+    value: Optional[str] = None
+    data: Optional[str] = None
 
 
 class Transaction(BaseModel):
@@ -56,6 +57,54 @@ class Transaction(BaseModel):
     params: EthTransactionParams
 
 
+# ---- signature ----
+
+class Address(str):
+    @classmethod
+    def __get_validators__(cls):
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, value, info):
+        if not is_address(value):
+            raise ValueError('invalid ethereum address')
+        return value
+
+
+class Bytes32(str):
+    pass
+
+
+class Bytes(str):
+    pass
+
+
+class Eip712Domain(BaseModel):
+    name: Optional[str] = None
+    version: Optional[str] = None
+    chainId: Optional[int] = None
+    verifyingContract: Optional[Address] = None
+    salt: Optional[str] = None
+
+
+class Eip712TypeField(BaseModel):
+    name: str
+    type: str
+
+
+class Eip712Params(BaseModel):
+    domain: Eip712Domain
+    types: dict[str, list[Eip712TypeField]]
+    primaryType: str
+    message: SerializeAsAny[BaseModel]
+
+
+class Signature(BaseModel):
+    chainId: str
+    method: Literal['eth_signTypedData_v4']
+    params: Eip712Params
+
+
 # ---- frame error ----
 
 class FrameError(BaseModel):

framelib/transaction.py

@@ -3,21 +3,24 @@
 """
 
 # lib
+from typing import Type
 from eth_abi import encode
 from eth_utils import is_address, function_signature_to_4byte_selector, function_abi_to_4byte_selector
 from flask import jsonify, Response
+from pydantic import BaseModel
 
 # src
-from .models import Transaction, EthTransactionParams
+from .models import Transaction, EthTransactionParams, Address, Bytes, Bytes32, Eip712TypeField, Signature, \
+    Eip712Domain, Eip712Params
 
 
 def transaction(
-        chain_id: int,
-        contract: str,
-        abi: list[dict],
-        value: str = None,
-        function_signature: str = None,
-        function_arguments: list = None
+    chain_id: int,
+    contract: str,
+    abi: list[dict],
+    value: str = None,
+    function_signature: str = None,
+    function_arguments: list = None
 ) -> Response:
     if not is_address(contract):
         raise ValueError(f'invalid contract address {contract}')
@@ -62,3 +65,70 @@ def mint(chain_id: int, contract: str, token_id: int = None) -> str:
         target += f':{token_id}'
 
     return target
+
+
+def signature(
+    chain_id: int,
+    message: BaseModel,
+    domain: str = None,
+    version: str = None,
+    contract: str = None,
+    salt: str = None
+) -> Response:
+    # collect custom types
+    def recurse_model_types(model: Type[BaseModel]):
+        types_ = {}
+        for name_, field_ in model.__annotations__.items():
+            if not issubclass(field_, BaseModel):
+                continue
+            types_ = recurse_model_types(field_)
+            types_[field_.__name__] = field_
+        types_[model.__name__] = model
+        return types_
+
+    types = recurse_model_types(message.__class__)
+
+    primitives = {
+        'int': 'uint256',
+        'str': 'string',
+        'bool': 'bool',
+        'Address': 'address',
+        'Bytes': 'bytes',
+        'Bytes32': 'bytes32'
+    }
+
+    # format eip712 type definitions
+    eip712_types = {}
+    for name, cls in types.items():
+        fields = []
+        for n, f in cls.__annotations__.items():
+            t = f.__name__
+            if t in primitives:
+                fields.append(Eip712TypeField(name=n, type=primitives[t]))
+            elif t in types:
+                fields.append(Eip712TypeField(name=n, type=t))
+            else:
+                raise ValueError(f'unsupported field type {n} {t}')
+        eip712_types[name] = fields
+
+    sig = Signature(
+        chainId=f'eip155:{chain_id}',
+        method='eth_signTypedData_v4',
+        params=Eip712Params(
+            domain=Eip712Domain(
+                name=domain,
+                version=version,
+                chainId=chain_id,
+                verifyingContract=contract,
+                salt=salt
+            ),
+            types=eip712_types,
+            primaryType=message.__class__.__name__,
+            message=message
+        )
+    )
+
+    # response
+    res = jsonify(sig.model_dump(mode='json', exclude_none=True))
+    res.status_code = 200
+    return res

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name='framelib',
-    version='0.0.6',
+    version='0.0.7b0',
     author='Devin A. Conley',
     author_email='[email protected]',
     description='lightweight library for building farcaster frames using python and flask',

test/test_mint.py

@@ -2,11 +2,6 @@
 test cases for frame minting logic
 """
 
-# lib
-import json
-import pytest
-from flask import Flask
-
 # src
 from framelib import mint
 

test/test_signature.py

@@ -0,0 +1,140 @@
+"""
+test cases for frame signature requests and verification
+"""
+
+# lib
+from pydantic import BaseModel
+from flask import Flask
+
+# src
+from framelib import signature, Address
+
+app = Flask(__name__)
+
+
+class TestSignature(object):
+
+    def test_signature_request(self):
+        class Message(BaseModel):
+            message: str
+            timestamp: int
+
+        msg = Message(message='hello world', timestamp=1234567890)
+
+        with app.app_context():
+            res = signature(1, msg, domain='myprotocol')
+
+            assert res.status_code == 200
+            assert res.json['chainId'] == 'eip155:1'
+            assert res.json['method'] == 'eth_signTypedData_v4'
+            assert res.json['params']['domain']['name'] == 'myprotocol'
+            assert res.json['params']['domain']['chainId'] == 1
+            assert 'salt' not in res.json['params']['domain']
+            assert 'verifyingContract' not in res.json['params']['domain']
+
+            assert len(res.json['params']['types']) == 1
+            assert res.json['params']['types']['Message'] == [
+                {'name': 'message', 'type': 'string'},
+                {'name': 'timestamp', 'type': 'uint256'}
+            ]
+            assert res.json['params']['primaryType'] == 'Message'
+            assert res.json['params']['message'] == {'message': 'hello world', 'timestamp': 1234567890}
+
+    def test_signature_request_nested(self):
+        class User(BaseModel):
+            name: str
+
+        class Message(BaseModel):
+            message: str
+            timestamp: int
+            sender: User
+            recipient: User
+
+        msg = Message(
+            message='hello bob',
+            timestamp=1234567890,
+            sender=User(name='alice'),
+            recipient=User(name='bob')
+        )
+
+        with app.app_context():
+            res = signature(
+                8453,
+                msg,
+                domain='another_app',
+                version='v3',
+                contract='0x1234567890abcdef1234567890abcdef12345678',
+                salt='0xabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdef',
+            )
+
+            assert res.status_code == 200
+            assert res.json['chainId'] == 'eip155:8453'
+            assert res.json['method'] == 'eth_signTypedData_v4'
+            assert res.json['params']['domain']['name'] == 'another_app'
+            assert res.json['params']['domain']['version'] == 'v3'
+            assert res.json['params']['domain']['chainId'] == 8453
+            assert res.json['params']['domain']['verifyingContract'] \
+                   == '0x1234567890abcdef1234567890abcdef12345678'
+            assert res.json['params']['domain']['salt'] \
+                   == '0xabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdef'
+
+            assert len(res.json['params']['types']) == 2
+            assert res.json['params']['types']['Message'] == [
+                {'name': 'message', 'type': 'string'},
+                {'name': 'timestamp', 'type': 'uint256'},
+                {'name': 'sender', 'type': 'User'},
+                {'name': 'recipient', 'type': 'User'}
+            ]
+            assert res.json['params']['types']['User'] == [
+                {'name': 'name', 'type': 'string'}
+            ]
+            assert res.json['params']['primaryType'] == 'Message'
+            assert res.json['params']['message'] == {
+                'message': 'hello bob',
+                'timestamp': 1234567890,
+                'sender': {'name': 'alice'},
+                'recipient': {'name': 'bob'}
+            }
+
+    def test_signature_request_eth(self):
+        class Approval(BaseModel):
+            token: Address
+            limit: int
+            expiry: int
+
+        msg = Approval(
+            token='0x4200000000000000000000000000000000000006',
+            limit=int(500e18),
+            expiry=1234567890
+        )
+
+        with app.app_context():
+            res = signature(
+                8453,
+                msg,
+                domain='gasless_exchange',
+                contract='0x1234567890abcdef1234567890abcdef12345678',
+            )
+
+            assert res.status_code == 200
+            assert res.json['chainId'] == 'eip155:8453'
+            assert res.json['method'] == 'eth_signTypedData_v4'
+            assert res.json['params']['domain']['name'] == 'gasless_exchange'
+            assert res.json['params']['domain']['chainId'] == 8453
+            assert res.json['params']['domain']['verifyingContract'] \
+                   == '0x1234567890abcdef1234567890abcdef12345678'
+            assert 'salt' not in res.json['params']['domain']
+            assert 'version' not in res.json['params']['domain']
+
+            assert len(res.json['params']['types']) == 1
+            assert res.json['params']['types']['Approval'] == [
+                {'name': 'token', 'type': 'address'},
+                {'name': 'limit', 'type': 'uint256'},
+                {'name': 'expiry', 'type': 'uint256'}
+            ]
+            assert res.json['params']['primaryType'] == 'Approval'
+            assert res.json['params']['message'] == {
+                'token': '0x4200000000000000000000000000000000000006',
+                'limit': 500e18,
+                'expiry': 1234567890
+            }