Update EIP-6404 with basic EIP-7932 support

Author: SirSpudlington

EIPS/eip-6404.md

@@ -11,10 +11,6 @@ created: 2023-01-30
 requires: 155, 1559, 2718, 2930, 4844, 7495, 7702, 7916, 7932, 8016
 ---
 
-<!-- TODO -->
-<!-- Update ../assets/eip-6404 -->
-
-
 ## Abstract
 
 This EIP defines a migration process of [EIP-2718](./eip-2718.md) Recursive-Length Prefix (RLP) transactions to [Simple Serialize (SSZ)](https://github.com/ethereum/consensus-specs/blob/b5c3b619887c7850a8c1d3540b471092be73ad84/ssz/simple-serialize.md).
@@ -66,28 +62,45 @@ Transaction signatures are represented by their native, opaque representation, p
 | `ExecutionSignature` | `ProgressiveByteList` |
 | `ExecutionSignatureAlgorithm` | `uint8` |
 
+#### Secp256k1
+
+| Name | Value | Description |
+| - | - | - |
+| `SECP256K1_ALGORITHM` | `uint8(0xFF)` | Defined from [EIP-7932](./eip-7932.md) |
+
+#### Verification
+
+The `algorithm_registry` is defined from [EIP-7932](./eip-7932.md) along with the `calculate_penalty` and `pubkey_to_address` functions.
 
-The `algorithm_registry`, `calculate_penalty` and `pubkey_to_address` objects are imported from [EIP-7932](./eip-7932.md):
 
 ```python
-algorithm_registry: Dict[uint8, AlgorithmEntry]
-pubkey_to_address: Callable[[Bytes, uint8], ExecutionAddress]
+
+def get_signature_gas_cost(
+    signature: ExecutionSignature,
+    expected_algorithm: Optional[ExecutionSignatureAlgorithm]=None
+) -> uint:
+    assert len(signature) > 0
+
+    if expected_algorithm is not None:
+        assert signature[0] == expected_algorithm
+
+    return calculate_penalty(signature)
+
 
 def validate_execution_signature(
     signature: ExecutionSignature,
-    signing_hash: Hash32,
+    signature_hash: Hash32,
+    expected_algorithm: Optional[ExecutionSignatureAlgorithm]=None,
 ) -> ExecutionAddress:
     assert len(signature) > 0
-    algorithm = signature[0]
-    assert algorithm in algorithm_registry
 
-    public_key = algorithm_registry[algorithm].verify(signature, signing_hash)
-    return pubkey_to_address(public_key, algorithm)
-```
+    if expected_algorithm is not None:
+        assert signature[0] == expected_algorithm
 
-The gas calculation section in [EIP-7932](./eip-7932.md) does apply to verification of transaction signatures.
+    public_key = algorithm_registry[signature[0]].verify(signature, signature_hash)
+    return pubkey_to_address(public_key, signature[0])
 
-ECDSA signatures are handled via the `secp256k1` (0xFF) algorithm of [EIP-7932](./eip-7932.md). All restrictions applied to algorithmic transaction by the `secp256k1` algorithm do not apply to any object specified in this document.
+```
 
 ### Gas fees
 
@@ -353,10 +366,26 @@ class RlpTxType(IntEnum):
     SET_CODE = 0x04
     SET_CODE_MAGIC = 0x05
 
+def calculate_base_gas_usage(tx: Transaction) -> uint:
+    tx_data = tx.payload.data()
+    gas_cost = TX_BASE_COST # FIXME: Should this be defined from another EIP?
+
+    if hasattr(tx_data, "authorization_list"):
+        for auth in tx_data.authorization_list:
+            gas_cost += get_signature_gas_cost(auth.signature, expected_algorithm=expected_signature_algorithm)
+
+    gas_cost += get_signature_gas_cost(tx.signature, expected_algorithm=expected_signature_algorithm)
+
+    return gas_cost
+
 def validate_transaction(tx: Transaction):
     tx_data = tx.payload.data()
 
+    expected_signature_algorithm = None
+    assert tx_data.gas >= calculate_base_gas_usage(tx)
+
     if hasattr(tx_data, "type_"):
+        expected_signature_algorithm = SECP256K1_ALGORITHM
         match tx_data.type_:
             case RlpTxType.LEGACY:
                 assert isinstance(tx_data, RlpLegacyTransactionPayload)
@@ -380,11 +409,10 @@ def validate_transaction(tx: Transaction):
             if hasattr(auth_data, "chain_id"):
                 assert auth_data.chain_id != 0
 
-            # TODO: figure out how to get the auth_hash variable
-            validate_execution_signature(auth.signature, auth_hash)
+            validate_execution_signature(auth.signature, compute_auth_hash(auth), expected_algorithm=expected_signature_algorithm)
+    
+    validate_execution_signature(tx.signature, compute_sig_hash(tx), expected_algorithm=expected_signature_algorithm)
 
-    # TODO: figure out how to get the signature_hash variable
-    validate_execution_signature(tx.signature, signature_hash)
 ```
 
 ### Execution block header changes

assets/eip-6404/algorithm_registry/helpers.py

@@ -0,0 +1,38 @@
+from remerkleable.basic import uint8, uint256, uint
+from remerkleable.byte_arrays import ByteVector
+
+from eth_hash.auto import keccak
+from eth_typing import Hash32
+
+
+from .registry import algorithm_registry
+
+
+class ExecutionAddress(ByteVector[20]):
+    pass
+
+
+def pubkey_to_address(public_key: bytes, algorithm_id: uint8) -> ExecutionAddress:
+    if algorithm_id == 0xFF:  # Compatibility shim to ensure backwards compatibility
+        return ExecutionAddress(keccak(public_key[1:])[12:])
+
+    # || is binary concatenation
+    return ExecutionAddress(keccak(bytes(algorithm_id) + public_key)[12:])
+
+
+def calculate_penalty(signature_info: bytes) -> uint:
+    GAS_PER_ADDITIONAL_VERIFICATION_BYTE = 16
+    SECP256K1_SIGNATURE_SIZE = 65
+
+    assert len(signature_info) > 0
+    assert uint8(signature_info[0]) in algorithm_registry
+
+    gas_penalty_base = (
+        max(len(signature_info) - (SECP256K1_SIGNATURE_SIZE + 1), 0)
+        * GAS_PER_ADDITIONAL_VERIFICATION_BYTE
+    )
+    total_gas_penalty = (
+        gas_penalty_base + algorithm_registry[uint8(signature_info[0])].GAS_PENALTY
+    )
+
+    return uint256(total_gas_penalty)

assets/eip-6404/algorithm_registry/registry.py

@@ -0,0 +1,59 @@
+from typing import Callable, Dict
+from eth_typing import Hash32
+from remerkleable.byte_arrays import ByteVector
+from remerkleable.basic import uint8, uint256, uint
+
+from secp256k1 import PublicKey, ECDSA
+
+
+# Registry setup
+
+
+class AlgorithmEntry:
+    ALG_TYPE: uint8
+    GAS_PENALTY: uint
+    verify: Callable[[bytes, Hash32], bytes]
+
+
+algorithm_registry: Dict[uint8, AlgorithmEntry] = {}
+
+# Secp256k1
+
+SECP256K1_SIGNATURE_SIZE = 65
+
+
+def secp256k1_unpack(
+    signature: ByteVector[SECP256K1_SIGNATURE_SIZE],
+) -> tuple[uint256, uint256, uint8]:
+    r = uint256.from_bytes(signature[0:32], "big")
+    s = uint256.from_bytes(signature[32:64], "big")
+    y_parity = signature[64]
+    return (r, s, uint8(y_parity))
+
+
+def secp256k1_validate(signature: ByteVector[SECP256K1_SIGNATURE_SIZE]):
+    SECP256K1N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
+    r, s, y_parity = secp256k1_unpack(signature)
+    assert 0 < r < SECP256K1N
+    assert 0 < s <= SECP256K1N // 2
+    assert y_parity in (0, 1)
+
+
+class Secp256k1(AlgorithmEntry):
+    ALG_TYPE = uint8(0xFF)
+    GAS_PENALTY = uint256(0)
+
+    def verify(signature_info: bytes, payload_hash: Hash32) -> bytes:
+        assert len(signature_info) == (SECP256K1_SIGNATURE_SIZE + 1)
+        secp256k1_validate(signature_info[1:])
+
+        ecdsa = ECDSA()
+        recover_sig = ecdsa.ecdsa_recoverable_deserialize(
+            signature_info[1:65], signature_info[65]
+        )
+        public_key = PublicKey(ecdsa.ecdsa_recover(payload_hash, recover_sig, raw=True))
+        uncompressed = public_key.serialize(compressed=False)
+        return uncompressed
+
+
+algorithm_registry[Secp256k1.ALG_TYPE] = Secp256k1

assets/eip-6404/convert.py

@@ -1,6 +1,7 @@
 from rlp import decode
 from rlp_types import *
 from ssz_types import *
+from ssz_helpers import *
 
 def upgrade_rlp_transaction_to_ssz(tx_bytes: bytes):
     type_ = tx_bytes[0]

assets/eip-6404/convert_tests.py

@@ -13,6 +13,8 @@ class Test:
     from_: ExecutionAddress
     authorities: list[ExecutionAddress]
 
+# FIXME: Make the tests work. i.e. change the algorithm_id from `0x01` -> `0xFF`
+# Also the tx with a gas value of `5`, isn't that inherently invalid?
 tests = [
     Test(
         tx_payload_selector=0x01,

assets/eip-6404/ssz_helpers.py

@@ -0,0 +1,51 @@
+from tx_hashes import compute_auth_hash, compute_sig_hash
+from ssz_types import *
+
+def calculate_base_gas_usage(tx: Transaction) -> uint:
+    tx_data = tx.payload.data()
+
+    TX_BASE_COST = 21000 # FIXME
+    gas_cost = TX_BASE_COST
+
+    if hasattr(tx_data, "authorization_list"):
+        for auth in tx_data.authorization_list:
+            gas_cost += get_signature_gas_cost(auth.signature)
+
+    gas_cost += get_signature_gas_cost(tx.signature)
+
+    return uint256(gas_cost)
+
+def validate_transaction(tx: Transaction):
+    tx_data = tx.payload.data()
+
+    expected_signature_algorithm = None
+    assert tx_data.gas >= calculate_base_gas_usage(tx)
+
+    if hasattr(tx_data, "type_"):
+        expected_signature_algorithm = SECP256K1_ALGORITHM
+        match tx_data.type_:
+            case RlpTxType.LEGACY:
+                assert isinstance(tx_data, RlpLegacyTransactionPayload)
+            case RlpTxType.ACCESS_LIST:
+                assert isinstance(tx_data, RlpAccessListTransactionPayload)
+            case RlpTxType.FEE_MARKET:
+                assert isinstance(tx_data, RlpFeeMarketTransactionPayload)
+            case RlpTxType.BLOB:
+                assert isinstance(tx_data, RlpBlobTransactionPayload)
+            case RlpTxType.SET_CODE:
+                assert isinstance(tx_data, RlpSetCodeTransactionPayload)
+            case _:
+                assert False
+
+    if hasattr(tx_data, "authorization_list"):
+        for auth in tx_data.authorization_list:
+            auth_data = auth.payload.data()
+
+            if hasattr(auth_data, "magic"):
+                assert auth_data.magic == RlpTxType.SET_CODE_MAGIC
+            if hasattr(auth_data, "chain_id"):
+                assert auth_data.chain_id != 0
+
+            validate_execution_signature(auth.signature, compute_auth_hash(auth), expected_algorithm=expected_signature_algorithm)
+
+    validate_execution_signature(tx.signature, compute_sig_hash(tx), expected_algorithm=expected_signature_algorithm)

assets/eip-6404/ssz_types.py

@@ -1,13 +1,16 @@
-from typing import Callable, Dict, Optional, Type
-from dataclasses import dataclass
+from typing import Optional
 from enum import IntEnum
 from eth_hash.auto import keccak
-from remerkleable.basic import uint8, uint64, uint256
+from remerkleable.basic import uint8, uint64, uint256, uint
 from remerkleable.byte_arrays import ByteVector, Bytes32
 from remerkleable.complex import Container
 from remerkleable.progressive import CompatibleUnion, ProgressiveByteList, ProgressiveContainer, ProgressiveList
 from secp256k1 import ECDSA, PublicKey
 
+from algorithm_registry.helpers import pubkey_to_address, calculate_penalty
+from algorithm_registry.registry import algorithm_registry
+
+
 class Hash32(Bytes32):
     pass
 
@@ -23,29 +26,34 @@ class ExecutionSignature(ProgressiveByteList):
 class ExecutionSignatureAlgorithm(uint8):
     pass
 
-@dataclass
-class ExecutionSignatureAttributes(object):
-    validate: Callable[[ExecutionSignature], None]
-    recover_signer: Callable[[ExecutionSignature, Hash32], ExecutionAddress]
 
-execution_signature_registry: Dict[ExecutionSignatureAlgorithm, ExecutionSignatureAttributes] = {}
+def get_signature_gas_cost(
+    signature: ExecutionSignature,
+    expected_algorithm: Optional[ExecutionSignatureAlgorithm]=None
+) -> uint:
+    assert len(signature) > 0
+
+    if expected_algorithm is not None:
+        assert signature[0] == expected_algorithm
+
+    return calculate_penalty(signature)
+
 
 def validate_execution_signature(
     signature: ExecutionSignature,
+    signature_hash: Hash32,
     expected_algorithm: Optional[ExecutionSignatureAlgorithm]=None,
-):
+) -> ExecutionAddress:
     assert len(signature) > 0
-    algorithm = signature[0]
+
     if expected_algorithm is not None:
-        assert algorithm == expected_algorithm
-    assert algorithm in execution_signature_registry
-    execution_signature_registry[algorithm].validate(signature)
+        assert signature[0] == expected_algorithm
 
-def recover_execution_signer(signature: ExecutionSignature, sig_hash: Hash32) -> ExecutionAddress:
-    algorithm = signature[0]
-    return execution_signature_registry[algorithm].recover_signer(signature, sig_hash)
+    public_key = algorithm_registry[signature[0]].verify(signature, signature_hash)
+    return pubkey_to_address(public_key, signature[0])
 
-SECP256K1_ALGORITHM = ExecutionSignatureAlgorithm(0x01)
+
+SECP256K1_ALGORITHM = ExecutionSignatureAlgorithm(0xFF)
 SECP256K1_SIGNATURE_SIZE = 1 + 32 + 32 + 1
 
 def secp256k1_pack(r: uint256, s: uint256, y_parity: uint8) -> ExecutionSignature:
@@ -62,25 +70,6 @@ def secp256k1_unpack(signature: ExecutionSignature) -> tuple[uint256, uint256, u
     y_parity = signature[65]
     return (r, s, y_parity)
 
-def secp256k1_validate(signature: ExecutionSignature):
-    r, s, y_parity = secp256k1_unpack(signature)
-    SECP256K1N = 0xfffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141
-    assert 0 < r < SECP256K1N
-    assert 0 < s <= SECP256K1N // 2
-    assert y_parity in (0, 1)
-
-def secp256k1_recover_signer(signature: ExecutionSignature, sig_hash: Hash32) -> ExecutionAddress:
-    ecdsa = ECDSA()
-    recover_sig = ecdsa.ecdsa_recoverable_deserialize(signature[1:65], signature[65])
-    public_key = PublicKey(ecdsa.ecdsa_recover(sig_hash, recover_sig, raw=True))
-    uncompressed = public_key.serialize(compressed=False)
-    return ExecutionAddress(keccak(uncompressed[1:])[12:])
-
-execution_signature_registry[SECP256K1_ALGORITHM] = ExecutionSignatureAttributes(
-    validate=secp256k1_validate,
-    recover_signer=secp256k1_recover_signer,
-)
-
 class FeePerGas(uint256):
     pass
 
@@ -294,25 +283,3 @@ class RlpTxType(IntEnum):
     SET_CODE = 0x04
     SET_CODE_MAGIC = 0x05
 
-def validate_transaction(tx: Transaction):
-    tx_data = tx.payload.data()
-    match tx_data.type_:
-        case RlpTxType.LEGACY:
-            assert isinstance(tx_data, RlpLegacyTransactionPayload)
-        case RlpTxType.ACCESS_LIST:
-            assert isinstance(tx_data, RlpAccessListTransactionPayload)
-        case RlpTxType.FEE_MARKET:
-            assert isinstance(tx_data, RlpFeeMarketTransactionPayload)
-        case RlpTxType.BLOB:
-            assert isinstance(tx_data, RlpBlobTransactionPayload)
-        case RlpTxType.SET_CODE:
-            assert isinstance(tx_data, RlpSetCodeTransactionPayload)
-            for auth in tx_data.authorization_list:
-                auth_data = auth.payload.data()
-                assert auth_data.magic == RlpTxType.SET_CODE_MAGIC
-                if hasattr(auth_data, "chain_id"):
-                    assert auth_data.chain_id != 0
-                validate_execution_signature(auth.signature, expected_algorithm=SECP256K1_ALGORITHM)
-        case _:
-            assert False
-    validate_execution_signature(tx.signature, expected_algorithm=SECP256K1_ALGORITHM)