diff --git a/cardano_node_tests/tests/bls.py b/cardano_node_tests/tests/bls.py index 02a4a38f4..3b0a3dc41 100644 --- a/cardano_node_tests/tests/bls.py +++ b/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 diff --git a/cardano_node_tests/tests/test_bls_keys.py b/cardano_node_tests/tests/test_bls_keys.py index f93d83865..f8ac0a81b 100644 --- a/cardano_node_tests/tests/test_bls_keys.py +++ b/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}" + ) diff --git a/cardano_node_tests/tests/test_bls_rotation.py b/cardano_node_tests/tests/test_bls_rotation.py index c5b4ba323..7c9ad869d 100644 --- a/cardano_node_tests/tests/test_bls_rotation.py +++ b/cardano_node_tests/tests/test_bls_rotation.py @@ -52,7 +52,6 @@ import dataclasses import json import logging -import math import pathlib as pl import shutil import time @@ -92,12 +91,6 @@ ), ] -# 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 - # Number of pool owner addresses created for a test pool POOL_OWNERS_NUM = 2 # Funds for the pool owners, all of it given to the first address, which is the one that @@ -254,39 +247,6 @@ def get_future_bls_key(*, cluster_obj: clusterlib.ClusterLib, pool_id: str) -> d return future_params.get("blsKey") 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 - - def write_bls_key_bundle(*, key_files: list[pl.Path], out_file: pl.Path) -> pl.Path: """Write several BLS signing keys into one file, as a JSON array of key envelopes. @@ -457,8 +417,10 @@ def create_test_pool( amount=POOL_OWNERS_FUNDS, ) + # 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_{temp_template}", + pool_name=f"pool_{clusterlib.get_rand_str(4)}", pool_pledge=1_000, pool_cost=cluster_obj.g_query.get_protocol_params().get("minPoolCost", 0), pool_margin=0.01, @@ -569,7 +531,7 @@ def check_seat_voting( expected_vkey: The hex encoded BLS key the seat is expected to hold (optional). context: Added to the reported message, to say why the pool should be voting. """ - seat = get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) if not seat: errors.append( @@ -608,7 +570,7 @@ def check_seat_keyless( epoch: The epoch the seat is checked in, for the reported message. errors: Collects the failures, appended to in place. """ - seat = get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) if not seat: errors.append( @@ -680,6 +642,7 @@ def rotate_bls_key( pool_creation_out: clusterlib.PoolCreationOutput, bls_skey_file: pl.Path | None, tx_name: str, + cert_suffix: str, ) -> None: """Re-register a pool with a different BLS signing key, leaving everything else as is. @@ -688,12 +651,15 @@ def rotate_bls_key( pool_creation_out: The output of the original pool registration. bls_skey_file: A path to the BLS signing key file to register, or `None` to submit a certificate that carries no BLS key at all. - tx_name: A name of the transaction, also used for naming the certificate. + tx_name: A name of the transaction. + cert_suffix: A short name of this rotation, added to the pool name. The + certificate file is named after the pool, so a rotation needs a name of its + own to avoid overwriting the certificate of the original registration. It is + kept short because a pool name is limited to 50 characters. """ - # The certificate file is named after the pool, so a rotation needs a name of its - # own to avoid overwriting the certificate of the original registration pool_data = dataclasses.replace( - pool_creation_out.pool_data, pool_name=f"{pool_creation_out.pool_data.pool_name}_{tx_name}" + pool_creation_out.pool_data, + pool_name=f"{pool_creation_out.pool_data.pool_name}_{cert_suffix}", ) cluster_obj.g_stake_pool.register_stake_pool( @@ -794,6 +760,7 @@ def test_rotate_bls_key( pool_creation_out=pool_creation_out, bls_skey_file=new_bls_key_pair.skey_file, tx_name=f"{temp_template}_rotate", + cert_suffix="rotate", ) assert cluster_obj.g_query.get_epoch() == rotate_epoch, ( @@ -830,7 +797,7 @@ def test_rotate_bls_key( # The committee of this epoch was seated before the update, so it still holds # the old key - seat = get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) assert seat, f"The pool holds no Leios committee seat in epoch {this_epoch}" assert bls.get_bls_pub_key(bls_key_state=seat.get("key")) == orig_vkey, ( f"The Leios committee of epoch {this_epoch} doesn't hold the original BLS key: {seat}" @@ -842,9 +809,9 @@ def test_rotate_bls_key( # One more boundary and the committee is seated from the snapshot that saw the # update this_epoch = cluster_obj.wait_for_epoch( - epoch_no=rotate_epoch + BLS_ACTIVATION_EPOCHS, padding_seconds=5 + epoch_no=rotate_epoch + bls.BLS_ACTIVATION_EPOCHS, padding_seconds=5 ) - seat = get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) assert seat, f"The pool holds no Leios committee seat in epoch {this_epoch}" assert bls.get_bls_pub_key(bls_key_state=seat.get("key")) == new_vkey, ( f"The Leios committee of epoch {this_epoch} doesn't hold the rotated BLS key: {seat}" @@ -913,6 +880,7 @@ def test_reregister_same_bls_key( pool_creation_out=pool_creation_out, bls_skey_file=get_pool_bls_key_pair(pool_creation_out=pool_creation_out).skey_file, tx_name=f"{temp_template}_renew", + cert_suffix="renew", ) assert cluster_obj.g_query.get_epoch() == renew_epoch, ( @@ -1020,9 +988,9 @@ def test_drop_and_restore_bls_key( # A pool with no key is still seated, it just cannot vote with its weight this_epoch = cluster_obj.wait_for_epoch( - epoch_no=reg_epoch + BLS_ACTIVATION_EPOCHS, padding_seconds=5 + epoch_no=reg_epoch + bls.BLS_ACTIVATION_EPOCHS, padding_seconds=5 ) - seat = get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster_obj, pool_id=pool_id) assert seat, f"The pool holds no Leios committee seat in epoch {this_epoch}" assert seat.get("key") is None, ( f"The keyless pool got a key on the Leios committee of epoch {this_epoch}: {seat}" @@ -1046,6 +1014,7 @@ def test_drop_and_restore_bls_key( pool_creation_out=pool_creation_out, bls_skey_file=new_bls_key_pair.skey_file, tx_name=f"{temp_template}_restore", + cert_suffix="restore", ) assert cluster_obj.g_query.get_epoch() == restore_epoch, ( @@ -1100,7 +1069,7 @@ def test_expired_bls_key( cluster = cluster_short_bls_keyage temp_template = common.get_test_id(cluster) - max_key_age = get_max_key_age(cluster_obj=cluster) + max_key_age = bls.get_max_key_age(cluster_obj=cluster) assert max_key_age == SHORT_MAX_KEY_AGE, ( f"The cluster instance gives BLS keys a lifetime of {max_key_age} epochs, " f"expected {SHORT_MAX_KEY_AGE}" @@ -1129,8 +1098,8 @@ def test_expired_bls_key( expire_epoch = orig_bls_key["bksRegisteredIn"] + max_key_age # The rotated key has to be seated before `expire_epoch`, and the rotation needs - # `BLS_ACTIVATION_EPOCHS` epoch boundaries to get there - rotate_epoch = expire_epoch - BLS_ACTIVATION_EPOCHS - 1 + # `bls.BLS_ACTIVATION_EPOCHS` epoch boundaries to get there + rotate_epoch = expire_epoch - bls.BLS_ACTIVATION_EPOCHS - 1 # The operational certificates are refreshed before the rotation, while the KES # keys of the pools are still valid refresh_epoch = rotate_epoch - 1 @@ -1394,7 +1363,7 @@ def test_rotation_pair_keeps_voting( this_epoch = cluster.wait_for_epoch( epoch_no=rotate_epoch + 1, padding_seconds=5, future_is_ok=False ) - seat = get_committee_seat(cluster_obj=cluster, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster, pool_id=pool_id) assert bls.get_bls_pub_key(bls_key_state=seat.get("key")) == orig_vkey, ( f"The Leios committee of epoch {this_epoch} doesn't hold the original BLS key " f"of '{pool_name}': {seat}" @@ -1410,9 +1379,9 @@ def test_rotation_pair_keeps_voting( # One more boundary and the rotated key is the seated one this_epoch = cluster.wait_for_epoch( - epoch_no=rotate_epoch + BLS_ACTIVATION_EPOCHS, padding_seconds=5 + epoch_no=rotate_epoch + bls.BLS_ACTIVATION_EPOCHS, padding_seconds=5 ) - seat = get_committee_seat(cluster_obj=cluster, pool_id=pool_id) + seat = bls.get_committee_seat(cluster_obj=cluster, pool_id=pool_id) assert bls.get_bls_pub_key(bls_key_state=seat.get("key")) == new_vkey, ( f"The Leios committee of epoch {this_epoch} doesn't hold the rotated BLS key " f"of '{pool_name}': {seat}"