Merge pull request #3283 from ethereum/deneb-crypto-coverage

Deneb crypto helpers test coverage

specs/deneb/p2p-interface.md

@@ -15,6 +15,8 @@ The specification of these changes continues in the same format as the network s
   - [`BlobSidecar`](#blobsidecar)
   - [`SignedBlobSidecar`](#signedblobsidecar)
   - [`BlobIdentifier`](#blobidentifier)
+  - [Helpers](#helpers)
+    - [`verify_sidecar_signature`](#verify_sidecar_signature)
 - [The gossip domain: gossipsub](#the-gossip-domain-gossipsub)
   - [Topics and messages](#topics-and-messages)
     - [Global topics](#global-topics)
@@ -73,6 +75,17 @@ class BlobIdentifier(Container):
     index: BlobIndex
 ```
 
+### Helpers
+
+#### `verify_sidecar_signature`
+
+```python
+def verify_blob_sidecar_signature(state: BeaconState, signed_blob_sidecar: SignedBlobSidecar) -> bool:
+    proposer = state.validators[signed_blob_sidecar.message.proposer_index]
+    signing_root = compute_signing_root(signed_blob_sidecar.message, get_domain(state, DOMAIN_BLOB_SIDECAR))
+    return bls.Verify(proposer.pubkey, signing_root, signed_blob_sidecar.signature)
+```
+
 ## The gossip domain: gossipsub
 
 Some gossip meshes are upgraded in the fork of Deneb to support upgraded types.
@@ -113,7 +126,7 @@ The following validations MUST pass before forwarding the `sidecar` on the netwo
 - _[IGNORE]_ The sidecar's block's parent (defined by `sidecar.block_parent_root`) has been seen (via both gossip and non-gossip sources) (a client MAY queue sidecars for processing once the parent block is retrieved).
 - _[REJECT]_ The sidecar's block's parent (defined by `sidecar.block_parent_root`) passes validation.
 - _[REJECT]_ The sidecar is from a higher slot than the sidecar's block's parent (defined by `sidecar.block_parent_root`).
-- _[REJECT]_ The proposer signature, `signed_blob_sidecar.signature`, is valid with respect to the `sidecar.proposer_index` pubkey.
+- _[REJECT]_ The proposer signature, `signed_blob_sidecar.signature`, is valid as verified by `verify_sidecar_signature`.
 - _[IGNORE]_ The sidecar is the only sidecar with valid signature received for the tuple `(sidecar.block_root, sidecar.index)`.
 - _[REJECT]_ The sidecar is proposed by the expected `proposer_index` for the block's slot in the context of the current shuffling (defined by `block_parent_root`/`slot`).
   If the `proposer_index` cannot immediately be verified against the expected shuffling, the sidecar MAY be queued for later processing while proposers for the block's branch are calculated -- in such a case _do not_ `REJECT`, instead `IGNORE` this message.

specs/deneb/validator.md

@@ -44,7 +44,9 @@ Note: This API is *unstable*. `get_blobs_and_kzg_commitments` and `get_payload`
 Implementers may also retrieve blobs individually per transaction.
 
 ```python
-def get_blobs_and_kzg_commitments(payload_id: PayloadId) -> Tuple[Sequence[BLSFieldElement], Sequence[KZGCommitment]]:
+def get_blobs_and_kzg_commitments(
+    payload_id: PayloadId
+) -> Tuple[Sequence[Blob], Sequence[KZGCommitment], Sequence[KZGProof]]:
     # pylint: disable=unused-argument
     ...
 ```
@@ -66,13 +68,14 @@ use the `payload_id` to retrieve `blobs` and `blob_kzg_commitments` via `get_blo
 ```python
 def validate_blobs_and_kzg_commitments(execution_payload: ExecutionPayload,
                                        blobs: Sequence[Blob],
-                                       blob_kzg_commitments: Sequence[KZGCommitment]) -> None:
+                                       blob_kzg_commitments: Sequence[KZGCommitment],
+                                       blob_kzg_proofs: Sequence[KZGProof]) -> None:
     # Optionally sanity-check that the KZG commitments match the versioned hashes in the transactions
     assert verify_kzg_commitments_against_transactions(execution_payload.transactions, blob_kzg_commitments)
 
     # Optionally sanity-check that the KZG commitments match the blobs (as produced by the execution engine)
-    assert len(blob_kzg_commitments) == len(blobs)
-    assert [blob_to_kzg_commitment(blob) == commitment for blob, commitment in zip(blobs, blob_kzg_commitments)]
+    assert len(blob_kzg_commitments) == len(blobs) == len(blob_kzg_proofs)
+    assert verify_blob_kzg_proof_batch(blobs, blob_kzg_commitments, blob_kzg_proofs)
 ```
 
 3. If valid, set `block.body.blob_kzg_commitments = blob_kzg_commitments`.
@@ -87,7 +90,9 @@ Blobs associated with a block are packaged into sidecar objects for distribution
 
 Each `sidecar` is obtained from:
 ```python
-def get_blob_sidecars(block: BeaconBlock, blobs: Sequence[Blob]) -> Sequence[BlobSidecar]:
+def get_blob_sidecars(block: BeaconBlock,
+                      blobs: Sequence[Blob],
+                      blob_kzg_proofs: Sequence[KZGProof]) -> Sequence[BlobSidecar]:
     return [
         BlobSidecar(
             block_root=hash_tree_root(block),
@@ -96,7 +101,7 @@ def get_blob_sidecars(block: BeaconBlock, blobs: Sequence[Blob]) -> Sequence[Blo
             block_parent_root=block.parent_root,
             blob=blob,
             kzg_commitment=block.body.blob_kzg_commitments[index],
-            kzg_proof=compute_blob_kzg_proof(blob, block.body.blob_kzg_commitments[index]),
+            kzg_proof=blob_kzg_proofs[index],
         )
         for index, blob in enumerate(blobs)
     ]

tests/core/pyspec/eth2spec/test/deneb/sanity/test_blocks.py

@@ -22,7 +22,7 @@ def test_one_blob(spec, state):
     yield 'pre', state
 
     block = build_empty_block_for_next_slot(spec, state)
-    opaque_tx, _, blob_kzg_commitments = get_sample_opaque_tx(spec)
+    opaque_tx, _, blob_kzg_commitments, _ = get_sample_opaque_tx(spec)
     block.body.blob_kzg_commitments = blob_kzg_commitments
     block.body.execution_payload.transactions = [opaque_tx]
     block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
@@ -38,7 +38,7 @@ def test_max_blobs(spec, state):
     yield 'pre', state
 
     block = build_empty_block_for_next_slot(spec, state)
-    opaque_tx, _, blob_kzg_commitments = get_sample_opaque_tx(spec, blob_count=spec.MAX_BLOBS_PER_BLOCK)
+    opaque_tx, _, blob_kzg_commitments, _ = get_sample_opaque_tx(spec, blob_count=spec.MAX_BLOBS_PER_BLOCK)
     block.body.blob_kzg_commitments = blob_kzg_commitments
     block.body.execution_payload.transactions = [opaque_tx]
     block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)

tests/core/pyspec/eth2spec/test/deneb/unittests/fork_choice/test_validate_blobs.py

@@ -18,14 +18,13 @@ from eth2spec.test.helpers.sharding import (
 
 def _run_validate_blobs(spec, state, blob_count):
     block = build_empty_block_for_next_slot(spec, state)
-    opaque_tx, blobs, blob_kzg_commitments = get_sample_opaque_tx(spec, blob_count=blob_count)
+    opaque_tx, blobs, blob_kzg_commitments, kzg_proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
     block.body.blob_kzg_commitments = blob_kzg_commitments
     block.body.execution_payload.transactions = [opaque_tx]
     block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
     state_transition_and_sign_block(spec, state, block)
 
-    # Also test the proof generation in `get_blob_sidecars`
-    blob_sidecars = spec.get_blob_sidecars(block, blobs)
+    blob_sidecars = spec.get_blob_sidecars(block, blobs, kzg_proofs)
     blobs = [sidecar.blob for sidecar in blob_sidecars]
     kzg_proofs = [sidecar.kzg_proof for sidecar in blob_sidecars]
     spec.validate_blobs(blob_kzg_commitments, blobs, kzg_proofs)

tests/core/pyspec/eth2spec/test/deneb/unittests/polynomial_commitments/test_polynomial_commitments.py

@@ -1,20 +1,79 @@
 import random
 
 from eth2spec.test.context import (
-    spec_state_test,
+    spec_test,
+    single_phase,
     with_deneb_and_later,
+    expect_assertion_error
 )
 from eth2spec.test.helpers.sharding import (
     get_sample_blob,
     get_poly_in_both_forms,
     eval_poly_in_coeff_form,
 )
+from eth2spec.utils import bls
+from eth2spec.utils.bls import BLS_MODULUS
+
+G1 = bls.G1_to_bytes48(bls.G1())
+P1_NOT_IN_G1 = bytes.fromhex("8123456789abcdef0123456789abcdef0123456789abcdef" +
+                             "0123456789abcdef0123456789abcdef0123456789abcdef")
+P1_NOT_ON_CURVE = bytes.fromhex("8123456789abcdef0123456789abcdef0123456789abcdef" +
+                                "0123456789abcdef0123456789abcdef0123456789abcde0")
+
+
+def bls_add_one(x):
+    """
+    Adds "one" (actually bls.G1()) to a compressed group element.
+    Useful to compute definitely incorrect proofs.
+    """
+    return bls.G1_to_bytes48(
+        bls.add(bls.bytes48_to_G1(x), bls.G1())
+    )
+
+
+def field_element_bytes(x):
+    return int.to_bytes(x % BLS_MODULUS, 32, "little")
 
 
 @with_deneb_and_later
-@spec_state_test
-def test_verify_kzg_proof(spec, state):
-    x = 3
+@spec_test
+@single_phase
+def test_verify_kzg_proof(spec):
+    """
+    Test the wrapper functions (taking bytes arguments) for computing and verifying KZG proofs.
+    """
+    x = field_element_bytes(3)
+    blob = get_sample_blob(spec)
+    commitment = spec.blob_to_kzg_commitment(blob)
+    proof, y = spec.compute_kzg_proof(blob, x)
+
+    assert spec.verify_kzg_proof(commitment, x, y, proof)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_verify_kzg_proof_incorrect_proof(spec):
+    """
+    Test the wrapper function `verify_kzg_proof` fails on an incorrect proof.
+    """
+    x = field_element_bytes(3465)
+    blob = get_sample_blob(spec)
+    commitment = spec.blob_to_kzg_commitment(blob)
+    proof, y = spec.compute_kzg_proof(blob, x)
+    proof = bls_add_one(proof)
+
+    assert not spec.verify_kzg_proof(commitment, x, y, proof)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_verify_kzg_proof_impl(spec):
+    """
+    Test the implementation functions (taking field element arguments) for computing and verifying KZG proofs.
+    """
+    x = BLS_MODULUS - 1
     blob = get_sample_blob(spec)
     commitment = spec.blob_to_kzg_commitment(blob)
     polynomial = spec.blob_to_polynomial(blob)
@@ -24,8 +83,26 @@ def test_verify_kzg_proof(spec, state):
 
 
 @with_deneb_and_later
-@spec_state_test
-def test_barycentric_outside_domain(spec, state):
+@spec_test
+@single_phase
+def test_verify_kzg_proof_impl_incorrect_proof(spec):
+    """
+    Test the implementation function `verify_kzg_proof` fails on an incorrect proof.
+    """
+    x = 324561
+    blob = get_sample_blob(spec)
+    commitment = spec.blob_to_kzg_commitment(blob)
+    polynomial = spec.blob_to_polynomial(blob)
+    proof, y = spec.compute_kzg_proof_impl(polynomial, x)
+    proof = bls_add_one(proof)
+
+    assert not spec.verify_kzg_proof_impl(commitment, x, y, proof)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_barycentric_outside_domain(spec):
     """
     Test barycentric formula correctness by using it to evaluate a polynomial at a bunch of points outside its domain
     (the roots of unity).
@@ -42,9 +119,9 @@ def test_barycentric_outside_domain(spec, state):
 
     for _ in range(n_samples):
         # Get a random evaluation point and make sure it's not a root of unity
-        z = rng.randint(0, spec.BLS_MODULUS - 1)
+        z = rng.randint(0, BLS_MODULUS - 1)
         while z in roots_of_unity_brp:
-            z = rng.randint(0, spec.BLS_MODULUS - 1)
+            z = rng.randint(0, BLS_MODULUS - 1)
 
         # Get p(z) by evaluating poly in coefficient form
         p_z_coeff = eval_poly_in_coeff_form(spec, poly_coeff, z)
@@ -57,8 +134,9 @@ def test_barycentric_outside_domain(spec, state):
 
 
 @with_deneb_and_later
-@spec_state_test
-def test_barycentric_within_domain(spec, state):
+@spec_test
+@single_phase
+def test_barycentric_within_domain(spec):
     """
     Test barycentric formula correctness by using it to evaluate a polynomial at all the points of its domain
     (the roots of unity).
@@ -89,8 +167,9 @@ def test_barycentric_within_domain(spec, state):
 
 
 @with_deneb_and_later
-@spec_state_test
-def test_compute_kzg_proof_within_domain(spec, state):
+@spec_test
+@single_phase
+def test_compute_kzg_proof_within_domain(spec):
     """
     Create and verify KZG proof that p(z) == y
     where z is in the domain of our KZG scheme (i.e. a relevant root of unity).
@@ -105,3 +184,122 @@ def test_compute_kzg_proof_within_domain(spec, state):
         proof, y = spec.compute_kzg_proof_impl(polynomial, z)
 
         assert spec.verify_kzg_proof_impl(commitment, z, y, proof)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_verify_blob_kzg_proof(spec):
+    """
+    Test the functions to compute and verify a blob KZG proof
+    """
+    blob = get_sample_blob(spec)
+    commitment = spec.blob_to_kzg_commitment(blob)
+    proof = spec.compute_blob_kzg_proof(blob, commitment)
+
+    assert spec.verify_blob_kzg_proof(blob, commitment, proof)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_verify_blob_kzg_proof_incorrect_proof(spec):
+    """
+    Check that `verify_blob_kzg_proof` fails on an incorrect proof
+    """
+    blob = get_sample_blob(spec)
+    commitment = spec.blob_to_kzg_commitment(blob)
+    proof = spec.compute_blob_kzg_proof(blob, commitment)
+    proof = bls_add_one(proof)
+
+    assert not spec.verify_blob_kzg_proof(blob, commitment, proof)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_validate_kzg_g1_generator(spec):
+    """
+    Verify that `validate_kzg_g1` allows the generator G1
+    """
+
+    spec.validate_kzg_g1(bls.G1_to_bytes48(bls.G1()))
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_validate_kzg_g1_neutral_element(spec):
+    """
+    Verify that `validate_kzg_g1` allows the neutral element in G1
+    """
+
+    spec.validate_kzg_g1(bls.G1_to_bytes48(bls.Z1()))
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_validate_kzg_g1_not_in_g1(spec):
+    """
+    Verify that `validate_kzg_g1` fails on point not in G1
+    """
+
+    expect_assertion_error(lambda: spec.validate_kzg_g1(P1_NOT_IN_G1))
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_validate_kzg_g1_not_on_curve(spec):
+    """
+    Verify that `validate_kzg_g1` fails on point not in G1
+    """
+
+    expect_assertion_error(lambda: spec.validate_kzg_g1(P1_NOT_ON_CURVE))
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_bytes_to_bls_field_zero(spec):
+    """
+    Verify that `bytes_to_bls_field` handles zero
+    """
+
+    spec.bytes_to_bls_field(b"\0" * 32)
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_bytes_to_bls_field_modulus_minus_one(spec):
+    """
+    Verify that `bytes_to_bls_field` handles modulus minus one
+    """
+
+    spec.bytes_to_bls_field((BLS_MODULUS - 1).to_bytes(spec.BYTES_PER_FIELD_ELEMENT, spec.ENDIANNESS))
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_bytes_to_bls_field_modulus(spec):
+    """
+    Verify that `bytes_to_bls_field` fails on BLS modulus
+    """
+
+    expect_assertion_error(lambda: spec.bytes_to_bls_field(
+        BLS_MODULUS.to_bytes(spec.BYTES_PER_FIELD_ELEMENT, spec.ENDIANNESS)
+    ))
+
+
+@with_deneb_and_later
+@spec_test
+@single_phase
+def test_bytes_to_bls_field_max(spec):
+    """
+    Verify that `bytes_to_bls_field` fails on 2**256 - 1
+    """
+
+    expect_assertion_error(lambda: spec.bytes_to_bls_field(b"\xFF" * 32))

tests/core/pyspec/eth2spec/test/deneb/unittests/test_kzg.py

@@ -1,21 +0,0 @@
-
-from eth2spec.test.helpers.constants import (
-    DENEB,
-    MINIMAL,
-)
-from eth2spec.test.helpers.sharding import (
-    get_sample_blob,
-)
-from eth2spec.test.context import (
-    with_phases,
-    spec_state_test,
-    with_presets,
-)
-
-
-@with_phases([DENEB])
-@spec_state_test
-@with_presets([MINIMAL])
-def test_blob_to_kzg_commitment(spec, state):
-    blob = get_sample_blob(spec)
-    spec.blob_to_kzg_commitment(blob)

tests/core/pyspec/eth2spec/test/deneb/unittests/test_offset.py

@@ -17,7 +17,7 @@ from eth2spec.test.context import (
 @spec_state_test
 @with_presets([MINIMAL])
 def test_tx_peek_blob_versioned_hashes(spec, state):
-    otx, blobs, commitments = get_sample_opaque_tx(spec)
+    otx, _, commitments, _ = get_sample_opaque_tx(spec)
     data_hashes = spec.tx_peek_blob_versioned_hashes(otx)
     expected = [spec.kzg_commitment_to_versioned_hash(blob_commitment) for blob_commitment in commitments]
     assert expected == data_hashes

tests/core/pyspec/eth2spec/test/deneb/unittests/validator/test_validator.py

@@ -0,0 +1,158 @@
+from eth2spec.test.context import (
+    always_bls,
+    spec_state_test,
+    with_deneb_and_later,
+    expect_assertion_error
+)
+from eth2spec.test.helpers.execution_payload import (
+    compute_el_block_hash,
+)
+from eth2spec.test.helpers.sharding import (
+    get_sample_opaque_tx,
+)
+from eth2spec.test.helpers.block import (
+    build_empty_block_for_next_slot
+)
+from eth2spec.test.helpers.keys import (
+    pubkey_to_privkey
+)
+
+
+@with_deneb_and_later
+@spec_state_test
+def test_validate_blobs_and_kzg_commitments(spec, state):
+    """
+    Test `validate_blobs_and_kzg_commitments`
+    """
+    blob_count = 4
+    block = build_empty_block_for_next_slot(spec, state)
+    opaque_tx, blobs, blob_kzg_commitments, proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
+    block.body.blob_kzg_commitments = blob_kzg_commitments
+    block.body.execution_payload.transactions = [opaque_tx]
+    block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
+
+    spec.validate_blobs_and_kzg_commitments(block.body.execution_payload,
+                                            blobs,
+                                            blob_kzg_commitments,
+                                            proofs)
+
+
+@with_deneb_and_later
+@spec_state_test
+def test_validate_blobs_and_kzg_commitments_missing_blob(spec, state):
+    """
+    Test `validate_blobs_and_kzg_commitments`
+    """
+    blob_count = 4
+    block = build_empty_block_for_next_slot(spec, state)
+    opaque_tx, blobs, blob_kzg_commitments, proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
+    block.body.blob_kzg_commitments = blob_kzg_commitments
+    block.body.execution_payload.transactions = [opaque_tx]
+    block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
+
+    expect_assertion_error(
+        lambda: spec.validate_blobs_and_kzg_commitments(
+            block.body.execution_payload,
+            blobs[:-1],
+            blob_kzg_commitments,
+            proofs
+        )
+    )
+
+
+@with_deneb_and_later
+@spec_state_test
+def test_validate_blobs_and_kzg_commitments_missing_proof(spec, state):
+    """
+    Test `validate_blobs_and_kzg_commitments`
+    """
+    blob_count = 4
+    block = build_empty_block_for_next_slot(spec, state)
+    opaque_tx, blobs, blob_kzg_commitments, proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
+    block.body.blob_kzg_commitments = blob_kzg_commitments
+    block.body.execution_payload.transactions = [opaque_tx]
+    block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
+
+    expect_assertion_error(
+        lambda: spec.validate_blobs_and_kzg_commitments(
+            block.body.execution_payload,
+            blobs,
+            blob_kzg_commitments,
+            proofs[:-1]
+        )
+    )
+
+
+@with_deneb_and_later
+@spec_state_test
+def test_validate_blobs_and_kzg_commitments_incorrect_blob(spec, state):
+    """
+    Test `validate_blobs_and_kzg_commitments`
+    """
+    blob_count = 4
+    block = build_empty_block_for_next_slot(spec, state)
+    opaque_tx, blobs, blob_kzg_commitments, proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
+    block.body.blob_kzg_commitments = blob_kzg_commitments
+    block.body.execution_payload.transactions = [opaque_tx]
+    block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
+
+    blobs[1] = spec.Blob(blobs[1][:13] + bytes([(blobs[1][13] + 1) % 256]) + blobs[1][14:])
+
+    expect_assertion_error(
+        lambda: spec.validate_blobs_and_kzg_commitments(
+            block.body.execution_payload,
+            blobs,
+            blob_kzg_commitments,
+            proofs
+        )
+    )
+
+
+@with_deneb_and_later
+@spec_state_test
+def test_blob_sidecar_signature(spec, state):
+    """
+    Test `get_blob_sidecar_signature`
+    """
+    blob_count = 4
+    block = build_empty_block_for_next_slot(spec, state)
+    opaque_tx, blobs, blob_kzg_commitments, proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
+    block.body.blob_kzg_commitments = blob_kzg_commitments
+    block.body.execution_payload.transactions = [opaque_tx]
+    block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
+
+    blob_sidecars = spec.get_blob_sidecars(block, blobs, proofs)
+    proposer = state.validators[blob_sidecars[1].proposer_index]
+    privkey = pubkey_to_privkey[proposer.pubkey]
+    sidecar_signature = spec.get_blob_sidecar_signature(state,
+                                                        blob_sidecars[1],
+                                                        privkey)
+
+    signed_blob_sidecar = spec.SignedBlobSidecar(message=blob_sidecars[1], signature=sidecar_signature)
+
+    assert spec.verify_blob_sidecar_signature(state, signed_blob_sidecar)
+
+
+@with_deneb_and_later
+@spec_state_test
+@always_bls
+def test_blob_sidecar_signature_incorrect(spec, state):
+    """
+    Test `get_blob_sidecar_signature`
+    """
+    blob_count = 4
+    block = build_empty_block_for_next_slot(spec, state)
+    opaque_tx, blobs, blob_kzg_commitments, proofs = get_sample_opaque_tx(spec, blob_count=blob_count)
+    block.body.blob_kzg_commitments = blob_kzg_commitments
+    block.body.execution_payload.transactions = [opaque_tx]
+    block.body.execution_payload.block_hash = compute_el_block_hash(spec, block.body.execution_payload)
+
+    blob_sidecars = spec.get_blob_sidecars(block, blobs, proofs)
+
+    sidecar_signature = spec.get_blob_sidecar_signature(state,
+                                                        blob_sidecars[1],
+                                                        123)
+
+    signed_blob_sidecar = spec.SignedBlobSidecar(message=blob_sidecars[1], signature=sidecar_signature)
+
+    assert not spec.verify_blob_sidecar_signature(state, signed_blob_sidecar)

tests/core/pyspec/eth2spec/test/helpers/sharding.py

@@ -101,13 +101,16 @@ def get_poly_in_both_forms(spec, rng=None):
 def get_sample_opaque_tx(spec, blob_count=1, rng=None):
     blobs = []
     blob_kzg_commitments = []
+    blob_kzg_proofs = []
     blob_versioned_hashes = []
     for _ in range(blob_count):
         blob = get_sample_blob(spec, rng)
         blob_commitment = spec.KZGCommitment(spec.blob_to_kzg_commitment(blob))
+        blob_kzg_proof = spec.compute_blob_kzg_proof(blob, blob_commitment)
         blob_versioned_hash = spec.kzg_commitment_to_versioned_hash(blob_commitment)
         blobs.append(blob)
         blob_kzg_commitments.append(blob_commitment)
+        blob_kzg_proofs.append(blob_kzg_proof)
         blob_versioned_hashes.append(blob_versioned_hash)
 
     signed_blob_tx = SignedBlobTransaction(
@@ -117,4 +120,4 @@ def get_sample_opaque_tx(spec, blob_count=1, rng=None):
     )
     serialized_tx = serialize(signed_blob_tx)
     opaque_tx = spec.uint_to_bytes(spec.BLOB_TX_TYPE) + serialized_tx
-    return opaque_tx, blobs, blob_kzg_commitments
+    return opaque_tx, blobs, blob_kzg_commitments, blob_kzg_proofs

tests/core/pyspec/eth2spec/utils/bls.py

@@ -12,6 +12,7 @@ from py_ecc.optimized_bls12_381 import (  # noqa: F401
     FQ12 as py_ecc_GT,
 )
 from py_ecc.bls.g2_primitives import (  # noqa: F401
+    curve_order as BLS_MODULUS,
     G1_to_pubkey as py_ecc_G1_to_bytes48,
     pubkey_to_G1 as py_ecc_bytes48_to_G1,
     G2_to_signature as py_ecc_G2_to_bytes96,