test: check the BLS proof of possession

cardano_node_tests/tests/bls.py

@@ -1,4 +1,8 @@
-"""Properties of the node BLS (Leios voting) keys, shared by the BLS tests.
+"""The node BLS (Leios voting) keys, shared by the BLS tests.
+
+Holds what the key files look like, what the ledger reports for a registered key, and
+the two facts that follow from the key being registered on chain - how long it stays
+honoured, and when the Leios committee starts holding it.
 
 The scheme is BLS12-381 in its minimal signature size variant, so the verification key is
 96 bytes in G2 and a signature is 48 bytes in G1. CIP-0164 fixes the on-chain encodings
@@ -13,6 +17,7 @@
 """
 
 import dataclasses
+import math
 import pathlib as pl
 
 from cardano_clusterlib import clusterlib
@@ -25,6 +30,12 @@
 # attacks, so only a key with a valid proof of possession may occupy a committee seat.
 POP_LEN = 48
 
+# Number of epoch boundaries between the transaction that registers a BLS key and the
+# epoch in which the Leios committee holds it. The first boundary applies the pool
+# update, the second seats the committee from the snapshot that saw the update. That is
+# the VRF key schedule, which CIP-0164 aligns voting keys with.
+BLS_ACTIVATION_EPOCHS = 2
+
 
 @dataclasses.dataclass(frozen=True)
 class KeySpec:
@@ -135,3 +146,36 @@ def get_registered_bls_key(*, cluster_obj: clusterlib.ClusterLib, pool_id: str)
     """
     pool_params = cluster_obj.g_query.get_pool_state(stake_pool_id=pool_id).pool_params
     return helpers.get_pool_param("spsBlsKey", pool_params=pool_params) or {}
+
+
+def get_committee_seat(*, cluster_obj: clusterlib.ClusterLib, pool_id: str) -> dict:
+    """Return the Leios committee seat of a pool in the current epoch.
+
+    The committee is reported as a whole regardless of the queried pool, so a single
+    pool ID is enough to get it.
+
+    Args:
+        cluster_obj: An instance of `clusterlib.ClusterLib`.
+        pool_id: An ID of the stake pool (Bech32-encoded or hex-encoded).
+
+    Returns:
+        dict: The seat of the pool, or an empty dict when the pool holds no seat.
+    """
+    pool_id_dec = helpers.decode_bech32(pool_id) if pool_id.startswith("pool") else pool_id
+    snapshot = cluster_obj.g_query.get_stake_snapshot(stake_pool_ids=[pool_id_dec])
+    committee: list[dict] = snapshot.get("leiosCommittee") or []
+    return next((s for s in committee if s["poolId"] == pool_id_dec), {})
+
+
+def get_max_key_age(*, cluster_obj: clusterlib.ClusterLib) -> int:
+    """Return the BLS key lifetime in epochs, as the ledger derives it from genesis.
+
+    Args:
+        cluster_obj: An instance of `clusterlib.ClusterLib`.
+
+    Returns:
+        int: The number of epochs a registered BLS key is honoured for.
+    """
+    genesis = cluster_obj.genesis
+    kes_lifetime = int(genesis["maxKESEvolutions"]) * int(genesis["slotsPerKESPeriod"])
+    return math.ceil(kes_lifetime / int(genesis["epochLength"])) + 2

cardano_node_tests/tests/test_bls_keys.py

@@ -1,16 +1,23 @@
-"""Tests for node BLS key generation."""
+"""Tests for generating and registering the node BLS (Leios voting) keys."""
 
+import dataclasses
 import logging
 import pathlib as pl
 import re
+import typing as tp
 
 import allure
+import cbor2
 import pytest
+from _pytest.fixtures import FixtureRequest
 from cardano_clusterlib import clusterlib
 
+from cardano_node_tests.cluster_management import cluster_management
+from cardano_node_tests.tests import addrs_common
 from cardano_node_tests.tests import bls
 from cardano_node_tests.tests import common
 from cardano_node_tests.utils import clusterlib_utils
+from cardano_node_tests.utils import dbsync_utils
 from cardano_node_tests.utils import helpers
 from cardano_node_tests.utils.versions import VERSIONS
 
@@ -149,7 +156,7 @@ def test_gen_bls_key_pair_matching_keys(
         owner_stake = cluster.g_stake_address.gen_stake_key_pair(key_name=f"{temp_template}_owner")
 
         pool_data = clusterlib.PoolData(
-            pool_name=f"pool_{temp_template}",
+            pool_name=f"pool_{clusterlib.get_rand_str(4)}",
             pool_pledge=5,
             pool_cost=500_000_000,
             pool_margin=0.01,
@@ -304,7 +311,7 @@ def test_bls_vkey_not_accepted_as_signing_key(
         with pytest.raises(clusterlib.CLIError) as excinfo:
             cluster.g_stake_pool.gen_pool_registration_cert(
                 pool_data=clusterlib.PoolData(
-                    pool_name=f"pool_{temp_template}",
+                    pool_name=f"pool_{clusterlib.get_rand_str(4)}",
                     pool_pledge=5,
                     pool_cost=500_000_000,
                     pool_margin=0.01,
@@ -352,3 +359,273 @@ def test_bls_skey_not_accepted_as_ed25519(
         with common.allow_unstable_error_messages():
             assert "TextEnvelope type error" in exc_value, exc_value
             assert bls.SKEY_SPEC.envelope_type in exc_value, exc_value
+
+
+@dataclasses.dataclass(frozen=True)
+class MismatchedPopPool:
+    """A stake pool registered with a BLS key that carries another key's proof."""
+
+    pool_id: str
+    # Hex encoded, the way the ledger reports them
+    vkey: str
+    pop: str
+    tx_output: clusterlib.TxRawOutput
+
+
+def register_pool_with_mismatched_pop(
+    *,
+    cluster_obj: clusterlib.ClusterLib,
+    pool_user: clusterlib.PoolUser,
+    temp_template: str,
+    testfile_temp_dir: pl.Path,
+    request: FixtureRequest,
+) -> MismatchedPopPool:
+    """Register a stake pool whose BLS key carries the proof of possession of another key.
+
+    The proof cannot be produced on its own - the CLI derives it from the signing key and
+    emits it only inside a registration certificate - so the mismatch is built by taking
+    the proof out of a second pool's certificate and splicing it into the first one.
+
+    The deregistration of the pool is scheduled as a finalizer.
+
+    Args:
+        cluster_obj: An instance of `clusterlib.ClusterLib`.
+        pool_user: A pool user that owns the pool and pays for the registration.
+        temp_template: A test identifier used for naming the created files.
+        testfile_temp_dir: A directory to run the deregistration in.
+        request: A pytest fixture request, used to register the deregistration.
+
+    Returns:
+        MismatchedPopPool: The pool ID, the registered key and proof, and the
+            registration transaction.
+    """
+    node_vrf = cluster_obj.g_node.gen_vrf_key_pair(node_name=f"{temp_template}_vrf")
+    node_cold = cluster_obj.g_node.gen_cold_key_pair_and_counter(node_name=f"{temp_template}_cold")
+
+    # A pool name is limited to 50 characters, which a test ID can eat on its own, so the
+    # pool is named after a random string instead
+    pool_data = clusterlib.PoolData(
+        pool_name=f"pool_{clusterlib.get_rand_str(4)}",
+        pool_pledge=5,
+        pool_cost=cluster_obj.g_query.get_protocol_params().get("minPoolCost", 0),
+        pool_margin=0.01,
+    )
+
+    # The proof of possession exists only inside a registration certificate, so a
+    # second certificate is generated just to take its proof
+    certs = {}
+    for name in ("own", "other"):
+        key_pair = cluster_obj.g_node.gen_bls_key_pair(node_name=f"{temp_template}_{name}")
+        certs[name] = cluster_obj.g_stake_pool.gen_pool_registration_cert(
+            pool_data=dataclasses.replace(pool_data, pool_name=f"{pool_data.pool_name}_{name}"),
+            vrf_vkey_file=node_vrf.vkey_file,
+            cold_vkey_file=node_cold.vkey_file,
+            owner_stake_vkey_files=[pool_user.stake.vkey_file],
+            bls_signing_key_file=key_pair.skey_file,
+        )
+
+    own_envelope = clusterlib_utils.load_envelope(envelope_file=certs["own"])
+    own_cert = cbor2.loads(bytes.fromhex(own_envelope["cborHex"]))
+    other_cert = clusterlib_utils.load_envelope_cbor(envelope_file=certs["other"])
+
+    assert len(own_cert) == common.POOL_REG_CERT_DIJKSTRA_ITEMS, (
+        f"Unexpected pool registration certificate: {own_cert}"
+    )
+
+    own_vkey, own_pop = own_cert[common.POOL_REG_CERT_BLS_IX]
+    __, other_pop = other_cert[common.POOL_REG_CERT_BLS_IX]
+    for name, pop in (("own", own_pop), ("other", other_pop)):
+        assert len(pop) == bls.POP_LEN, f"Unexpected {name} proof length: {len(pop)}"
+    assert other_pop != own_pop, "The two certificates carry the same proof of possession"
+
+    # Re-encoding has to be faithful, otherwise the submitted certificate would differ
+    # from the generated one in more than the proof and a rejection would say nothing
+    # about the proof
+    mismatched_cert: list[tp.Any] = list(own_cert)
+    assert cbor2.dumps(mismatched_cert).hex() == own_envelope["cborHex"], (
+        "Re-encoding the pool registration certificate is not faithful"
+    )
+
+    mismatched_cert[common.POOL_REG_CERT_BLS_IX] = [own_vkey, other_pop]
+    mismatched_cert_file = pl.Path(f"{temp_template}_mismatched_pop.cert")
+    helpers.write_json(
+        out_file=mismatched_cert_file,
+        content={**own_envelope, "cborHex": cbor2.dumps(mismatched_cert).hex()},
+    )
+
+    tx_files = clusterlib.TxFiles(
+        certificate_files=[mismatched_cert_file],
+        signing_key_files=[
+            pool_user.payment.skey_file,
+            pool_user.stake.skey_file,
+            node_cold.skey_file,
+        ],
+    )
+    tx_output = cluster_obj.g_transaction.send_tx(
+        src_address=pool_user.payment.address,
+        tx_name=f"{temp_template}_reg_pool",
+        tx_files=tx_files,
+    )
+
+    def _deregister() -> None:
+        with helpers.change_cwd(testfile_temp_dir):
+            cluster_obj.g_stake_pool.deregister_stake_pool(
+                pool_owners=[pool_user],
+                cold_key_pair=node_cold,
+                epoch=cluster_obj.g_query.get_epoch() + 2,
+                pool_name=pool_data.pool_name,
+                tx_name=f"{temp_template}_cleanup",
+            )
+
+    request.addfinalizer(_deregister)
+
+    return MismatchedPopPool(
+        pool_id=cluster_obj.g_stake_pool.get_stake_pool_id(node_cold.vkey_file),
+        vkey=own_vkey.hex(),
+        pop=other_pop.hex(),
+        tx_output=tx_output,
+    )
+
+
+class TestBlsProofOfPossession:
+    """Tests for the proof of possession that accompanies a registered BLS key."""
+
+    @pytest.fixture
+    def pool_user(
+        self,
+        cluster_manager: cluster_management.ClusterManager,
+        cluster: clusterlib.ClusterLib,
+    ) -> clusterlib.PoolUser:
+        """Create a pool user with a registered stake address."""
+        return addrs_common.get_registered_pool_user(
+            name_template=common.get_test_id(cluster),
+            cluster_manager=cluster_manager,
+            cluster_obj=cluster,
+            caching_key=helpers.get_current_line_str(),
+            amount=900_000_000,
+            min_amount=600_000_000,
+        )
+
+    @allure.link(helpers.get_vcs_link())
+    @pytest.mark.testnets
+    @pytest.mark.smoke
+    @pytest.mark.dbsync
+    def test_mismatched_pop_is_not_verified(
+        self,
+        cluster: clusterlib.ClusterLib,
+        pool_user: clusterlib.PoolUser,
+        testfile_temp_dir: pl.Path,
+        request: FixtureRequest,
+    ):
+        """Register a pool with a BLS key whose proof of possession belongs to another key.
+
+        CIP-0164 makes two statements about the proof, because BLS aggregate signatures
+        are otherwise open to rogue-key attacks. It "is mandatory and verified at
+        registration", and "Only a key with a valid proof of possession may occupy a
+        committee seat or contribute to a certificate".
+
+        Only the second one is implemented. The Dijkstra `POOL` rule reuses
+        `ShelleyPoolPredFailure` unchanged and has no BLS predicate failure at all, so a
+        registration carrying a proof that does not belong to the key is accepted and the
+        pair is stored verbatim. The proof is verified later, when the committee is
+        seated - see `test_mismatched_pop_is_not_seated`.
+
+        This test pins the registration side down, so that the day the ledger starts
+        rejecting such a registration is a day this test fails and says so, rather than a
+        silent change of behaviour.
+
+        * Register a pool with a BLS key and the proof of possession of another key
+        * Check that the registration is accepted
+        * Check that the ledger stored the mismatched key and proof exactly as submitted
+        """
+        temp_template = common.get_test_id(cluster)
+
+        pool = register_pool_with_mismatched_pop(
+            cluster_obj=cluster,
+            pool_user=pool_user,
+            temp_template=temp_template,
+            testfile_temp_dir=testfile_temp_dir,
+            request=request,
+        )
+
+        # The key and the proof are stored the way they were submitted, unverified
+        bls_key = bls.get_registered_bls_key(cluster_obj=cluster, pool_id=pool.pool_id)
+        assert bls.get_bls_pub_key(bls_key_state=bls_key) == pool.vkey, (
+            f"The pool didn't register the expected BLS key: {bls_key}"
+        )
+        assert bls_key["bksKey"]["blsPossessionProof"] == pool.pop, (
+            f"The pool didn't register the mismatched proof of possession: {bls_key}"
+        )
+
+        dbsync_utils.check_tx(cluster_obj=cluster, tx_raw_output=pool.tx_output)
+
+    @allure.link(helpers.get_vcs_link())
+    @pytest.mark.leios
+    @pytest.mark.long
+    def test_mismatched_pop_is_not_seated(
+        self,
+        cluster: clusterlib.ClusterLib,
+        pool_user: clusterlib.PoolUser,
+        testfile_temp_dir: pl.Path,
+        request: FixtureRequest,
+    ):
+        """Check that a key with a mismatched proof of possession cannot vote.
+
+        This is the half of the CIP-0164 requirement that is implemented, and it is the
+        half that matters: the registration is accepted with the proof unverified, but
+        `mkLeiosCommittee` verifies it when it builds the committee and admits the seat
+        keyless, so the key can neither vote nor contribute to a certificate.
+
+        The pool keeps its seat and the ledger keeps reporting the key it registered -
+        what it doesn't get is the ability to vote with the weight the seat carries. The
+        key is well inside its lifetime here, so age cannot be the reason.
+
+        * Register a pool with a BLS key and the proof of possession of another key
+        * Wait until the committee is seated from a snapshot that holds the key
+        * Check that the pool is on the committee and that the key is reported
+        * Check that the key is not past its lifetime, so only the proof can disqualify it
+        * Check that the seat is not voting
+        """
+        temp_template = common.get_test_id(cluster)
+
+        pool = register_pool_with_mismatched_pop(
+            cluster_obj=cluster,
+            pool_user=pool_user,
+            temp_template=temp_template,
+            testfile_temp_dir=testfile_temp_dir,
+            request=request,
+        )
+        bls_key = bls.get_registered_bls_key(cluster_obj=cluster, pool_id=pool.pool_id)
+        assert bls.get_bls_pub_key(bls_key_state=bls_key) == pool.vkey, (
+            f"The pool didn't register the expected BLS key: {bls_key}"
+        )
+        registered_in = bls_key["bksRegisteredIn"]
+
+        this_epoch = cluster.wait_for_epoch(
+            epoch_no=registered_in + bls.BLS_ACTIVATION_EPOCHS, padding_seconds=5
+        )
+        seat = bls.get_committee_seat(cluster_obj=cluster, pool_id=pool.pool_id)
+        assert seat, (
+            f"The pool has no Leios committee seat in epoch {this_epoch}, so there is "
+            "nothing to say about its key"
+        )
+        assert bls.get_bls_pub_key(bls_key_state=seat.get("key")) == pool.vkey, (
+            f"The seat doesn't report the registered BLS key: {seat}"
+        )
+        assert seat["key"]["bksRegisteredIn"] == registered_in, (
+            f"The seat reports the key as registered in epoch "
+            f"{seat['key']['bksRegisteredIn']} instead of {registered_in}: {seat}"
+        )
+
+        # An aged-out key cannot vote either, so rule that out
+        max_key_age = bls.get_max_key_age(cluster_obj=cluster)
+        assert this_epoch < registered_in + max_key_age, (
+            f"The BLS key registered in epoch {registered_in} is already past its "
+            f"{max_key_age} epoch lifetime in epoch {this_epoch}, so the proof of "
+            "possession is not what keeps it from voting"
+        )
+
+        assert seat["voting"] is False, (
+            f"The pool is voting in epoch {this_epoch} with a BLS key whose proof of "
+            f"possession belongs to another key: {seat}"
+        )