feat(staking): support the two-step operator rotation

genlayer_py/client/genlayer_client.py

@@ -52,6 +52,11 @@
     validator_claim,
     validator_prime,
     set_operator,
+    get_operator_transfer_context,
+    initiate_operator_transfer,
+    complete_operator_transfer,
+    cancel_operator_transfer,
+    get_pending_operator,
     set_identity,
     delegator_join,
     delegator_exit,
@@ -490,6 +495,38 @@ def set_operator(
             self=self, validator=validator, operator=operator, account=account
         )
 
+    def get_operator_transfer_context(self, validator):
+        """Wallet-bound context for building a rotation proof."""
+        return get_operator_transfer_context(self=self, validator=validator)
+
+    def initiate_operator_transfer(
+        self, validator, registration, account: Optional[LocalAccount] = None
+    ) -> HexBytes:
+        """Starts the two-step operator rotation (CON-715)."""
+        return initiate_operator_transfer(
+            self=self, validator=validator, registration=registration, account=account
+        )
+
+    def complete_operator_transfer(
+        self, validator, account: Optional[LocalAccount] = None
+    ) -> HexBytes:
+        """Finalises a pending operator rotation."""
+        return complete_operator_transfer(
+            self=self, validator=validator, account=account
+        )
+
+    def cancel_operator_transfer(
+        self, validator, account: Optional[LocalAccount] = None
+    ) -> HexBytes:
+        """Abandons a pending operator rotation."""
+        return cancel_operator_transfer(
+            self=self, validator=validator, account=account
+        )
+
+    def get_pending_operator(self, validator) -> dict:
+        """Pending operator and when its transfer was initiated."""
+        return get_pending_operator(self=self, validator=validator)
+
     def set_identity(
         self, validator, moniker: str, account: Optional[LocalAccount] = None
     ) -> HexBytes:

genlayer_py/staking/__init__.py

@@ -5,6 +5,11 @@
     validator_claim,
     validator_prime,
     set_operator,
+    get_operator_transfer_context,
+    initiate_operator_transfer,
+    complete_operator_transfer,
+    cancel_operator_transfer,
+    get_pending_operator,
     set_identity,
     delegator_join,
     delegator_exit,
@@ -27,6 +32,11 @@
     "validator_claim",
     "validator_prime",
     "set_operator",
+    "get_operator_transfer_context",
+    "initiate_operator_transfer",
+    "complete_operator_transfer",
+    "cancel_operator_transfer",
+    "get_pending_operator",
     "set_identity",
     "delegator_join",
     "delegator_exit",

genlayer_py/staking/abi/validator_wallet_abi.json

@@ -998,5 +998,55 @@
   {
     "stateMutability": "payable",
     "type": "receive"
+  },
+  {
+    "name": "initiateOperatorTransfer",
+    "type": "function",
+    "stateMutability": "nonpayable",
+    "inputs": [
+      {
+        "name": "_newOperatorPubKey",
+        "type": "uint256[2]",
+        "internalType": "uint256[2]"
+      },
+      {
+        "name": "_possessionProof",
+        "type": "bytes",
+        "internalType": "bytes"
+      }
+    ],
+    "outputs": []
+  },
+  {
+    "name": "completeOperatorTransfer",
+    "type": "function",
+    "stateMutability": "nonpayable",
+    "inputs": [],
+    "outputs": []
+  },
+  {
+    "name": "cancelOperatorTransfer",
+    "type": "function",
+    "stateMutability": "nonpayable",
+    "inputs": [],
+    "outputs": []
+  },
+  {
+    "name": "getPendingOperator",
+    "type": "function",
+    "stateMutability": "view",
+    "inputs": [],
+    "outputs": [
+      {
+        "name": "",
+        "type": "address",
+        "internalType": "address"
+      },
+      {
+        "name": "",
+        "type": "uint256",
+        "internalType": "uint256"
+      }
+    ]
   }
 ]

genlayer_py/staking/actions.py

@@ -17,6 +17,11 @@
 
 from genlayer_py.exceptions import GenLayerError
 from genlayer_py.staking.abi import STAKING_ABI, VALIDATOR_WALLET_ABI
+from genlayer_py.staking.operator_registration import (
+    OperatorRegistrationContext,
+    OperatorRegistrationProof,
+    verify_operator_registration,
+)
 
 if TYPE_CHECKING:
     from genlayer_py.client import GenLayerClient
@@ -237,7 +242,12 @@ def set_operator(
     account: Optional[LocalAccount] = None,
 ) -> HexBytes:
     """Rotates the operator for an existing ValidatorWallet. Only the
-    wallet owner (the EOA that called validator_join) may do this."""
+    wallet owner (the EOA that called validator_join) may do this.
+
+    CON-715 removed this single call in favour of the two-step transfer
+    below; against a deployment carrying that change it reverts with no
+    decodable reason, because the selector no longer exists. Prefer
+    initiate_operator_transfer + complete_operator_transfer."""
     sender = _sender(self, account)
     wallet = _wallet(self, validator)
     data = wallet.encode_abi(
@@ -247,6 +257,89 @@ def set_operator(
     return _send(self, sender, tx)
 
 
+def get_operator_transfer_context(
+    self: "GenLayerClient", validator: AddressLike
+) -> OperatorRegistrationContext:
+    """Context for a rotation proof.
+
+    Rotation is verified by the wallet rather than the factory, so the
+    registrar is the wallet's own address. The owner is read from the wallet
+    instead of assumed to be the caller: the proof is bound to whatever
+    owner() returns, and a mismatch is easier to diagnose here than as an
+    onlyOwner revert."""
+    wallet = _wallet(self, validator)
+    return OperatorRegistrationContext(
+        registrar=self.w3.to_checksum_address(validator),
+        owner=self.w3.to_checksum_address(wallet.functions.owner().call()),
+        chain_id=self.w3.eth.chain_id,
+    )
+
+
+def initiate_operator_transfer(
+    self: "GenLayerClient",
+    validator: AddressLike,
+    registration: OperatorRegistrationProof,
+    account: Optional[LocalAccount] = None,
+) -> HexBytes:
+    """Starts the two-step operator rotation. Owner only.
+
+    `registration` must be built against get_operator_transfer_context — a
+    join proof is bound to the factory and will not verify here."""
+    context = get_operator_transfer_context(self, validator)
+    if not verify_operator_registration(registration, context):
+        raise GenLayerError(
+            "Operator registration proof does not match the wallet, owner, chain, "
+            "or public key. Rotation proofs must use the validator wallet as "
+            "their registrar."
+        )
+
+    sender = _sender(self, account)
+    wallet = _wallet(self, validator)
+    data = wallet.encode_abi(
+        "initiateOperatorTransfer",
+        args=[list(registration.operator_pub_key), registration.possession_proof],
+    )
+    tx = _build(self, sender, self.w3.to_checksum_address(validator), data)
+    return _send(self, sender, tx)
+
+
+def complete_operator_transfer(
+    self: "GenLayerClient",
+    validator: AddressLike,
+    account: Optional[LocalAccount] = None,
+) -> HexBytes:
+    """Finalises a pending rotation. Callable by the wallet owner or the
+    pending operator, once the factory's operatorTransferDelay has elapsed."""
+    sender = _sender(self, account)
+    wallet = _wallet(self, validator)
+    data = wallet.encode_abi("completeOperatorTransfer", args=[])
+    tx = _build(self, sender, self.w3.to_checksum_address(validator), data)
+    return _send(self, sender, tx)
+
+
+def cancel_operator_transfer(
+    self: "GenLayerClient",
+    validator: AddressLike,
+    account: Optional[LocalAccount] = None,
+) -> HexBytes:
+    """Abandons a pending rotation, leaving the current operator in place."""
+    sender = _sender(self, account)
+    wallet = _wallet(self, validator)
+    data = wallet.encode_abi("cancelOperatorTransfer", args=[])
+    tx = _build(self, sender, self.w3.to_checksum_address(validator), data)
+    return _send(self, sender, tx)
+
+
+def get_pending_operator(self: "GenLayerClient", validator: AddressLike) -> dict:
+    """Pending operator and when its transfer was initiated (0 when none)."""
+    wallet = _wallet(self, validator)
+    operator, initiated_at = wallet.functions.getPendingOperator().call()
+    return {
+        "operator": self.w3.to_checksum_address(operator),
+        "initiated_at": int(initiated_at),
+    }
+
+
 def set_identity(
     self: "GenLayerClient",
     validator: AddressLike,

genlayer_py/staking/operator_registration.py

@@ -0,0 +1,135 @@
+"""Proof of possession for operator keys.
+
+Consensus requires an operator to prove control of its key before that key is
+bound to a validator wallet. The proof is an EIP-191 signature, by the operator
+key, over a domain-separated hash of (chainId, registrar, owner, pubKey).
+
+`registrar` is whichever contract verifies the proof, and it differs by flow:
+the ValidatorWalletFactory for a validator join, the wallet itself for an
+operator rotation (ValidatorWalletBlueprint.initiateOperatorTransfer calls
+PubKeyUtils.validateWithPossession(pubKey, address(this), owner(), proof)).
+Passing the wrong one produces a proof that simply fails to verify.
+
+The encoding mirrors genlayer-js's createOperatorRegistration exactly — the
+shared vector in tests/unit/test_operator_registration.py pins the two
+implementations together.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Tuple
+
+from eth_abi.abi import encode as abi_encode
+from eth_account import Account
+from eth_account.messages import encode_defunct
+from eth_typing import ChecksumAddress
+from eth_utils.address import to_checksum_address
+from eth_utils.crypto import keccak
+
+OPERATOR_REGISTRATION_DOMAIN = keccak(
+    text="GenLayer/operatorPubKey/proof-of-possession/v1"
+)
+
+OperatorPublicKey = Tuple[int, int]
+
+
+@dataclass(frozen=True)
+class OperatorRegistrationContext:
+    """Who verifies the proof, on whose behalf, and on which chain."""
+
+    registrar: ChecksumAddress
+    owner: ChecksumAddress
+    chain_id: int
+
+
+@dataclass(frozen=True)
+class OperatorRegistrationProof:
+    operator: ChecksumAddress
+    operator_pub_key: OperatorPublicKey
+    possession_proof: bytes
+
+
+def operator_public_key_from_private_key(private_key: str) -> OperatorPublicKey:
+    """Splits the uncompressed secp256k1 public key into the contract's
+    uint256[2] tuple."""
+    account = Account.from_key(private_key)
+    public_key = account._key_obj.public_key.to_bytes()
+    return (
+        int.from_bytes(public_key[0:32], "big"),
+        int.from_bytes(public_key[32:64], "big"),
+    )
+
+
+def operator_address_from_public_key(pub_key: OperatorPublicKey) -> ChecksumAddress:
+    raw = pub_key[0].to_bytes(32, "big") + pub_key[1].to_bytes(32, "big")
+    return to_checksum_address(keccak(raw)[-20:])
+
+
+def operator_possession_message(
+    pub_key: OperatorPublicKey, context: OperatorRegistrationContext
+) -> bytes:
+    return keccak(
+        abi_encode(
+            ["bytes32", "uint256", "address", "address", "uint256", "uint256"],
+            [
+                OPERATOR_REGISTRATION_DOMAIN,
+                context.chain_id,
+                context.registrar,
+                context.owner,
+                pub_key[0],
+                pub_key[1],
+            ],
+        )
+    )
+
+
+def create_operator_registration(
+    private_key: str, context: OperatorRegistrationContext
+) -> OperatorRegistrationProof:
+    """Builds the proof package the proof-bearing calls consume.
+
+    The private key is used only to sign and is never retained in the result.
+    """
+    account = Account.from_key(private_key)
+    pub_key = operator_public_key_from_private_key(private_key)
+    operator = operator_address_from_public_key(pub_key)
+
+    if operator != to_checksum_address(account.address):
+        raise ValueError(
+            "Operator private key and public key derive different identities."
+        )
+
+    signed = Account.sign_message(
+        encode_defunct(operator_possession_message(pub_key, context)),
+        private_key=private_key,
+    )
+
+    return OperatorRegistrationProof(
+        operator=operator,
+        operator_pub_key=pub_key,
+        possession_proof=bytes(signed.signature),
+    )
+
+
+def verify_operator_registration(
+    registration: OperatorRegistrationProof,
+    context: OperatorRegistrationContext,
+) -> bool:
+    """Checks key identity and the exact registrar/owner/chain binding."""
+    if operator_address_from_public_key(registration.operator_pub_key) != to_checksum_address(
+        registration.operator
+    ):
+        return False
+
+    message = encode_defunct(
+        operator_possession_message(registration.operator_pub_key, context)
+    )
+    try:
+        recovered = Account.recover_message(
+            message, signature=registration.possession_proof
+        )
+    except Exception:
+        return False
+
+    return to_checksum_address(recovered) == to_checksum_address(registration.operator)