Implement extended blob rlp logic for 4844 definition:

- When a blob transaction is signed, it needs to contain the blob information
  along with commitments and proofs for the consensus layer. We need to be
  able to serialize and deserialize this information.
- Use different serializers for execution layer style `TransactionPayloadBody`
  blob transactions and for `PooledTransaction` blob transactions.

eth_account/_utils/legacy_transactions.py

@@ -35,11 +35,15 @@
 )
 
 
-def serializable_unsigned_transaction_from_dict(transaction_dict):
+def serializable_unsigned_transaction_from_dict(transaction_dict, blobs=None):
     transaction_dict = set_transaction_type_if_needed(transaction_dict)
     if "type" in transaction_dict:
         # We delegate to TypedTransaction, which will carry out validation & formatting.
-        return TypedTransaction.from_dict(transaction_dict)
+        return TypedTransaction.from_dict(transaction_dict, blobs=blobs)
+
+    if blobs is not None:
+        # sanity check, blobs should never get past typed transactions check above
+        raise TypeError("Blob data is not supported for legacy transactions.")
 
     assert_valid_fields(transaction_dict)
     filled_transaction = pipe(
@@ -65,7 +69,11 @@ def encode_transaction(unsigned_transaction, vrs):
         chain_naive_transaction["v"] = v
         chain_naive_transaction["r"] = r
         chain_naive_transaction["s"] = s
-        signed_typed_transaction = TypedTransaction.from_dict(chain_naive_transaction)
+        blob_data = unsigned_transaction.blob_data
+        signed_typed_transaction = TypedTransaction.from_dict(
+            chain_naive_transaction,
+            blobs=[blob.as_bytes() for blob in blob_data.blobs] if blob_data else None,
+        )
         return signed_typed_transaction.encode()
     signed_transaction = Transaction(v=v, r=r, s=s, **chain_naive_transaction)
     return rlp.encode(signed_transaction)

eth_account/_utils/signing.py

@@ -27,9 +27,11 @@
 STRUCTURED_DATA_SIGN_VERSION = b"\x01"  # Hex value 0x01
 
 
-def sign_transaction_dict(eth_key, transaction_dict):
+def sign_transaction_dict(eth_key, transaction_dict, blobs=None):
     # generate RLP-serializable transaction, with defaults filled
-    unsigned_transaction = serializable_unsigned_transaction_from_dict(transaction_dict)
+    unsigned_transaction = serializable_unsigned_transaction_from_dict(
+        transaction_dict, blobs=blobs
+    )
 
     transaction_hash = unsigned_transaction.hash()
 

eth_account/_utils/typed_transactions/__init__.py

@@ -1,7 +1,7 @@
 from .access_list_transaction import (
     AccessListTransaction,
 )
-from .blob_transaction import (
+from .blob_transactions.blob_transaction import (
     BlobTransaction,
 )
 from .dynamic_fee_transaction import (

eth_account/_utils/typed_transactions/access_list_transaction.py

@@ -1,6 +1,7 @@
 from typing import (
     Any,
     Dict,
+    List,
     Tuple,
     cast,
 )
@@ -28,7 +29,7 @@
     BigEndianInt,
     Binary,
     CountableList,
-    List,
+    List as ListSedesClass,
     big_endian_int,
     binary,
 )
@@ -51,7 +52,7 @@
 # [[{20 bytes}, [{32 bytes}...]]...], where ... means
 # “zero or more of the thing to the left”.
 access_list_sede_type = CountableList(
-    List(
+    ListSedesClass(
         [
             Binary.fixed_length(20, allow_empty=False),
             CountableList(BigEndianInt(32)),
@@ -140,11 +141,16 @@ def assert_valid_fields(cls, dictionary: Dict[str, Any]) -> None:
             raise TypeError(f"Transaction had invalid fields: {repr(invalid)}")
 
     @classmethod
-    def from_dict(cls, dictionary: Dict[str, Any]) -> "AccessListTransaction":
+    def from_dict(
+        cls, dictionary: Dict[str, Any], blobs: List[bytes] = None
+    ) -> "AccessListTransaction":
         """
         Builds an AccessListTransaction from a dictionary.
         Verifies that the dictionary is well formed.
         """
+        if blobs is not None:
+            raise ValueError("Blob data is not supported for `AccessListTransaction`.")
+
         # Validate fields.
         cls.assert_valid_fields(dictionary)
         sanitized_dictionary = pipe(

eth_account/_utils/typed_transactions/base.py

@@ -2,12 +2,25 @@
     ABC,
     abstractmethod,
 )
+import hashlib
+import os
 from typing import (
     Any,
     Dict,
+    List,
+    Optional,
     Tuple,
+    Union,
 )
 
+from ckzg import (
+    blob_to_kzg_commitment,
+    compute_blob_kzg_proof,
+    load_trusted_setup,
+)
+from eth_typing import (
+    HexStr,
+)
 from eth_utils import (
     is_bytes,
     is_string,
@@ -24,6 +37,15 @@
     identity,
     merge,
 )
+from hexbytes import (
+    HexBytes,
+)
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    computed_field,
+    field_validator,
+)
 
 from eth_account._utils.validation import (
     LEGACY_TRANSACTION_FORMATTERS,
@@ -54,13 +76,171 @@
     },
 )
 
+# import TRUSTED_SETUP from ./kzg_trusted_setup.txt
+TRUSTED_SETUP = os.path.join(
+    os.path.dirname(__file__), "blob_transactions", "kzg_trusted_setup.txt"
+)
+VERSIONED_HASH_VERSION_KZG = b"\x01"
+
+
+class _BlobDataElement(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    data: HexBytes
+
+    def as_hexbytes(self) -> HexBytes:
+        return self.data
+
+    def as_bytes(self) -> bytes:
+        return bytes(self.data)
+
+    def as_hexstr(self) -> HexStr:
+        return HexStr(f"0x{self.as_bytes().hex()}")
+
+
+class Blob(_BlobDataElement):
+    """
+    Represents a Blob.
+    """
+
+    @field_validator("data")
+    def validate_data(cls, v: Union[HexBytes, bytes]) -> Union[HexBytes, bytes]:
+        if len(v) != 4096 * 32:
+            raise ValueError(
+                "Invalid Blob size. Blob data must be comprised of 4096 32-byte "
+                "field elements."
+            )
+        return v
+
+
+class BlobKZGCommitment(_BlobDataElement):
+    """
+    Represents a Blob KZG Commitment.
+    """
+
+    @field_validator("data")
+    def validate_commitment(cls, v: Union[HexBytes, bytes]) -> Union[HexBytes, bytes]:
+        if len(v) != 48:
+            raise ValueError("Blob KZG Commitment must be 48 bytes long.")
+        return v
+
+
+class BlobProof(_BlobDataElement):
+    """
+    Represents a Blob Proof.
+    """
+
+    @field_validator("data")
+    def validate_proof(cls, v: Union[HexBytes, bytes]) -> Union[HexBytes, bytes]:
+        if len(v) != 48:
+            raise ValueError("Blob Proof must be 48 bytes long.")
+        return v
+
+
+class BlobVersionedHash(_BlobDataElement):
+    """
+    Represents a Blob Versioned Hash.
+    """
+
+    @field_validator("data")
+    def validate_versioned_hash(
+        cls, v: Union[HexBytes, bytes]
+    ) -> Union[HexBytes, bytes]:
+        if len(v) != 32:
+            raise ValueError("Blob Versioned Hash must be 32 bytes long.")
+        if v[:1] != VERSIONED_HASH_VERSION_KZG:
+            raise ValueError(
+                "Blob Versioned Hash must start with the KZG version byte."
+            )
+        return v
+
+
+class BlobPooledTransactionData(BaseModel):
+    """
+    Represents the blob data for a type 3 `PooledTransaction` as defined by
+    EIP-4844. This class takes blobs as bytes and computes the corresponding
+    commitments, proofs, and versioned hashes.
+    """
+
+    _versioned_hash_version_kzg: bytes = VERSIONED_HASH_VERSION_KZG
+    _versioned_hashes: Optional[List[BlobVersionedHash]] = None
+    _commitments: Optional[List[BlobKZGCommitment]] = None
+    _proofs: Optional[List[BlobProof]] = None
+
+    blobs: List[Blob]
+
+    def _kzg_to_versioned_hash(self, kzg_commitment: BlobKZGCommitment) -> bytes:
+        return (
+            self._versioned_hash_version_kzg
+            + hashlib.sha256(kzg_commitment.data).digest()[1:]
+        )
+
+    @field_validator("blobs")
+    def validate_blobs(cls, v: List[Blob]) -> List[Blob]:
+        if len(v) == 0:
+            raise ValueError("Blob transactions must contain at least one blob.")
+        return v
+
+    # type ignored bc mypy does not support decorated properties
+    # https://github.com/python/mypy/issues/1362
+    @computed_field  # type: ignore
+    @property
+    def versioned_hashes(self) -> List[BlobVersionedHash]:
+        if self._versioned_hashes is None:
+            self._versioned_hashes = [
+                BlobVersionedHash(
+                    data=HexBytes(self._kzg_to_versioned_hash(commitment))
+                )
+                for commitment in self.commitments
+            ]
+        return self._versioned_hashes
+
+    # type ignored bc mypy does not support decorated properties
+    # https://github.com/python/mypy/issues/1362
+    @computed_field  # type: ignore
+    @property
+    def commitments(self) -> List[BlobKZGCommitment]:
+        if self._commitments is None:
+            self._commitments = [
+                BlobKZGCommitment(
+                    data=HexBytes(
+                        blob_to_kzg_commitment(
+                            blob.data, load_trusted_setup(TRUSTED_SETUP)
+                        )
+                    )
+                )
+                for blob in self.blobs
+            ]
+        return self._commitments
+
+    # type ignored bc mypy does not support decorated properties
+    # https://github.com/python/mypy/issues/1362
+    @computed_field  # type: ignore
+    @property
+    def proofs(self) -> List[BlobProof]:
+        if self._proofs is None:
+            self._proofs = [
+                BlobProof(
+                    data=HexBytes(
+                        compute_blob_kzg_proof(
+                            blob.data,
+                            commitment.data,
+                            load_trusted_setup(TRUSTED_SETUP),
+                        )
+                    )
+                )
+                for blob, commitment in zip(self.blobs, self.commitments)
+            ]
+        return self._proofs
+
 
 class _TypedTransactionImplementation(ABC):
     """
     Abstract class that every typed transaction must implement.
     Should not be imported or used by clients of the library.
     """
 
+    blob_data: Optional[BlobPooledTransactionData] = None
+
     @abstractmethod
     def hash(self) -> bytes:
         pass

eth_account/_utils/typed_transactions/blob_transactions/__init__.py

@@ -0,0 +1,3 @@
+from .blob_transaction import (
+    BlobTransaction,
+)