diff --git a/tests/frontier/validation/test_transaction_rlp.py b/tests/frontier/validation/test_transaction_rlp.py new file mode 100644 index 00000000000..fb3fbbb53a0 --- /dev/null +++ b/tests/frontier/validation/test_transaction_rlp.py @@ -0,0 +1,383 @@ +"""Tests for RLP-level validity of type-0 transaction encodings.""" + +from typing import Dict, List, Mapping + +import pytest +from execution_testing import ( + Alloc, + Bytes, + Fork, + Transaction, + TransactionException, + TransactionTestFiller, +) + +pytestmark = pytest.mark.valid_from("Frontier") + +LEGACY_TX_TESTS = ( + "https://github.com/ethereum/tests/blob/" + "c67e485ff8b5be9abc8ad15345ec21aa22e290d9/src/TransactionTestsFiller" +) + + +def encode_header(length: int, offset: int) -> bytes: + """Encode an RLP header for a payload of the given length.""" + if length < 56: + return bytes([offset + length]) + size = length.to_bytes((length.bit_length() + 7) // 8, "big") + return bytes([offset + 55 + len(size)]) + size + + +def rlp_bytes(payload: bytes) -> bytes: + """Encode a byte string.""" + if len(payload) == 1 and payload[0] < 0x80: + return payload + return encode_header(len(payload), 0x80) + payload + + +def rlp_list(items: List[bytes]) -> bytes: + """Encode a list of already-encoded items.""" + payload = b"".join(items) + return encode_header(len(payload), 0xC0) + payload + + +def int_payload(value: int) -> bytes: + """Return the canonical RLP payload of an integer.""" + if value == 0: + return b"" + return value.to_bytes((value.bit_length() + 7) // 8, "big") + + +def tx_fields(tx: Transaction) -> Dict[str, bytes]: + """ + Decompose a signed transaction into its canonical RLP payload per + field, using the framework's own field order and values. + """ + return { + name: bytes(el) if isinstance(el, bytes) else int_payload(int(el)) + for name, el in zip( + tx.get_rlp_fields(), + tx.to_list(signing=False), + strict=True, + ) + } + + +def signed_tx(pre: Alloc, fork: Fork) -> Transaction: + """Build and sign the base type-0 transaction.""" + return Transaction( + sender=pre.fund_eoa(), + to=pre.fund_eoa(amount=0), + gas_price=10, + gas_limit=30_000, + value=1, + protected=fork.supports_protected_txs(), + ).with_signature_and_sender() + + +def signed_tx_fields(pre: Alloc, fork: Fork) -> Dict[str, bytes]: + """Build a signed type-0 transaction and decompose it.""" + return tx_fields(signed_tx(pre, fork)) + + +def encode_tx( + fields: Mapping[str, bytes], override: Mapping[str, bytes] | None = None +) -> bytes: + """ + Encode the transaction fields, allowing per-field pre-encoded + replacements. + """ + override = override or {} + return rlp_list( + [override.get(name, rlp_bytes(fields[name])) for name in fields] + ) + + +def invalid_tx( + pre: Alloc, + rlp: bytes, + error: TransactionException | list[TransactionException], +) -> Transaction: + """Return a transaction whose serialization is the given raw bytes.""" + tx = Transaction( + sender=pre.fund_eoa(), + to=0, + gas_price=10, + gas_limit=30_000, + error=error, + ) + tx.rlp_override = Bytes(rlp) + return tx + + +def test_valid_reencoded_transaction( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, +) -> None: + """ + Verify the local field encoder reproduces the framework encoding + byte for byte, anchoring the malformation tests to a valid baseline. + """ + tx = signed_tx(pre, fork) + fields = tx_fields(tx) + assert encode_tx(fields) == bytes(tx.rlp()) + transaction_test(pre=pre, tx=tx) + + +@pytest.mark.ported_from( + [ + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPNonceWithFirstZerosCopier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPgasPriceWithFirstZerosCopier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPgasLimitWithFirstZerosCopier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPValueWithFirstZerosCopier.json", + f"{LEGACY_TX_TESTS}/ttRSValue/TransactionWithRvaluePrefixed00Filler.json", + f"{LEGACY_TX_TESTS}/ttRSValue/TransactionWithSvaluePrefixed00Filler.json", + f"{LEGACY_TX_TESTS}/ttRSValue/RightVRSTestVPrefixedBy0Filler.json", + ], +) +@pytest.mark.exception_test +@pytest.mark.parametrize( + "field, error", + [ + ("nonce", TransactionException.RLP_LEADING_ZEROS_NONCE), + ("gas_price", TransactionException.RLP_LEADING_ZEROS_GASPRICE), + ("gas_limit", TransactionException.RLP_LEADING_ZEROS_GASLIMIT), + ("value", TransactionException.RLP_LEADING_ZEROS_VALUE), + ("v", TransactionException.RLP_LEADING_ZEROS_V), + ("r", TransactionException.RLP_LEADING_ZEROS_R), + ("s", TransactionException.RLP_LEADING_ZEROS_S), + ], +) +def test_field_leading_zeros( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, + field: str, + error: TransactionException, +) -> None: + """ + Prefix one integer field's payload with a zero byte; the + non-canonical encoding must be rejected. The base nonce is zero, so + its variant is the classic zero-encoded-as-0x00 case. + """ + fields = signed_tx_fields(pre, fork) + corrupted = rlp_bytes(b"\x00" + fields[field]) + rlp = encode_tx(fields, {field: corrupted}) + transaction_test(pre=pre, tx=invalid_tx(pre, rlp, error)) + + +@pytest.mark.ported_from( + [ + f"{LEGACY_TX_TESTS}/ttNonce/TransactionWithNonceOverflowFiller.json", + f"{LEGACY_TX_TESTS}/ttValue/TransactionWithHighValueOverflowFiller.json", + f"{LEGACY_TX_TESTS}/ttRSValue/TransactionWithRvalueOverflowFiller.json", + f"{LEGACY_TX_TESTS}/ttRSValue/TransactionWithSvalueOverflowFiller.json", + ], +) +@pytest.mark.exception_test +@pytest.mark.parametrize( + "field, error", + [ + ("nonce", TransactionException.RLP_INVALID_NONCE), + ("value", TransactionException.VALUE_OVERFLOW), + ("r", TransactionException.RLP_INVALID_SIGNATURE_R), + ("s", TransactionException.RLP_INVALID_SIGNATURE_S), + ], +) +def test_field_overflow( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, + field: str, + error: TransactionException, +) -> None: + """ + Encode one integer field as a 33-byte value (2**256), exceeding the + field's width (64 bits for the nonce, 256 bits for the others). + + The gas limit and gas price are unbounded scalars in the spec, so + oversized values there are not a decoding error and are rejected + only in block context. + """ + fields = signed_tx_fields(pre, fork) + corrupted = rlp_bytes(int_payload(2**256)) + rlp = encode_tx(fields, {field: corrupted}) + transaction_test(pre=pre, tx=invalid_tx(pre, rlp, error)) + + +@pytest.mark.ported_from( + [ + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPAddressWrongSizeCopier.json", + f"{LEGACY_TX_TESTS}/ttAddress/AddressLessThan20Filler.json", + f"{LEGACY_TX_TESTS}/ttAddress/AddressMoreThan20PrefixedBy0Filler.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPAddressWithFirstZerosCopier.json", + ], +) +@pytest.mark.exception_test +@pytest.mark.parametrize( + "size, error", + [ + (19, TransactionException.ADDRESS_TOO_SHORT), + (21, TransactionException.ADDRESS_TOO_LONG), + ], +) +def test_to_address_size( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, + size: int, + error: TransactionException, +) -> None: + """ + Encode the to field with a truncated 19-byte address or a 21-byte + address made of a zero byte prefixing a valid address. + """ + fields = signed_tx_fields(pre, fork) + if size < 20: + fields["to"] = fields["to"][:size] + else: + fields["to"] = fields["to"].rjust(size, b"\x00") + rlp = encode_tx(fields) + transaction_test(pre=pre, tx=invalid_tx(pre, rlp, error)) + + +@pytest.mark.ported_from( + [ + f"{LEGACY_TX_TESTS}/ttWrongRLP/" + "RLPElementIsListWhenItShouldntBeCopier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/" + "RLPElementIsListWhenItShouldntBe2Copier.json", + ], +) +@pytest.mark.exception_test +@pytest.mark.parametrize( + "field, error", + [ + ("nonce", TransactionException.RLP_INVALID_NONCE), + ("gas_limit", TransactionException.RLP_INVALID_GASLIMIT), + ("to", TransactionException.RLP_INVALID_TO), + ("value", TransactionException.RLP_INVALID_VALUE), + ("data", TransactionException.RLP_INVALID_DATA), + ], +) +def test_field_as_list( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, + field: str, + error: TransactionException, +) -> None: + """Encode one field as an RLP list instead of a byte string.""" + fields = signed_tx_fields(pre, fork) + corrupted = rlp_list([rlp_bytes(fields[field])]) + rlp = encode_tx(fields, {field: corrupted}) + transaction_test(pre=pre, tx=invalid_tx(pre, rlp, error)) + + +@pytest.mark.ported_from( + [ + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPExtraRandomByteAtTheEndCopier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/TRANSCT_HeaderLargerThanRLP_0Copier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/TRANSCT_HeaderGivenAsArray_0Copier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/RLPListLengthWithFirstZerosCopier.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/aMaliciousRLPCopier.json", + ], +) +@pytest.mark.exception_test +@pytest.mark.parametrize( + "mutation, error", + [ + ("truncated", TransactionException.RLP_ERROR_EOF), + ("extra_byte", TransactionException.RLP_ERROR_SIZE), + ("too_few_elements", TransactionException.RLP_TOO_FEW_ELEMENTS), + ("too_many_elements", TransactionException.RLP_TOO_MANY_ELEMENTS), + ("header_declares_more", TransactionException.RLP_ERROR_EOF), + ("header_declares_less", TransactionException.RLP_ERROR_EOF), + ("tx_as_byte_string", TransactionException.RLP_INVALID_HEADER), + ( + "list_size_leading_zeros", + TransactionException.RLP_ERROR_SIZE_LEADING_ZEROS, + ), + ], +) +def test_invalid_structure( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, + mutation: str, + error: TransactionException, +) -> None: + """Corrupt the RLP structure of the whole transaction.""" + fields = signed_tx_fields(pre, fork) + items = [rlp_bytes(fields[name]) for name in fields] + payload = b"".join(items) + # The header mutations assume the long-form list header. + assert len(payload) >= 56 + good = encode_tx(fields) + if mutation == "truncated": + rlp = good[:-1] + elif mutation == "extra_byte": + rlp = good + b"\x00" + elif mutation == "too_few_elements": + rlp = rlp_list(items[:-1]) + elif mutation == "too_many_elements": + rlp = rlp_list(items + [rlp_bytes(b"")]) + elif mutation == "header_declares_more": + rlp = encode_header(len(payload) + 1, 0xC0) + payload + elif mutation == "header_declares_less": + rlp = encode_header(len(payload) - 1, 0xC0) + payload + elif mutation == "tx_as_byte_string": + rlp = encode_header(len(payload), 0x80) + payload + elif mutation == "list_size_leading_zeros": + size = len(payload).to_bytes(2, "big") + assert size[0] == 0 + rlp = bytes([0xC0 + 55 + len(size)]) + size + payload + transaction_test(pre=pre, tx=invalid_tx(pre, rlp, error)) + + +@pytest.mark.ported_from( + [ + f"{LEGACY_TX_TESTS}/ttRSValue/TransactionWithRvalue0Filler.json", + f"{LEGACY_TX_TESTS}/ttRSValue/TransactionWithSvalue0Filler.json", + f"{LEGACY_TX_TESTS}/ttVValue/V_wrongvalue_ffFiller.json", + f"{LEGACY_TX_TESTS}/ttWrongRLP/tr201506052141PYTHONCopier.json", + ], +) +@pytest.mark.exception_test +@pytest.mark.parametrize( + "field, payload, error", + [ + ("r", b"", TransactionException.INVALID_SIGNATURE_VRS), + ("s", b"", TransactionException.INVALID_SIGNATURE_VRS), + ("v", b"\x1d", TransactionException.INVALID_SIGNATURE_VRS), + ( + "v", + b"\xff", + [ + TransactionException.INVALID_SIGNATURE_VRS, + TransactionException.INVALID_CHAINID, + ], + ), + ], + ids=["r_zero", "s_zero", "v_29", "v_255"], +) +def test_invalid_signature_values( + transaction_test: TransactionTestFiller, + pre: Alloc, + fork: Fork, + field: str, + payload: bytes, + error: TransactionException | list[TransactionException], +) -> None: + """ + Replace a signature field with a well-encoded but invalid value: + zero r or s, or a v that is neither 27, 28 nor an EIP-155 value. + + Before EIP-155 a v of 255 is a plain invalid v; afterwards it is + interpreted as a chain id that does not match the chain. + """ + fields = signed_tx_fields(pre, fork) + fields[field] = payload + rlp = encode_tx(fields) + transaction_test(pre=pre, tx=invalid_tx(pre, rlp, error))