The script below will unregister a validator node given its node-keystore.ks file and password. This is really only for emergency use where you’re unable to unregister a node because it is still syncing.
The script reads the node-keystore.ks file which is in a PKCS12 format, extracts its private key, generates the corresponding public key, validator address and validator wallet address. Then using this data it submits a build transaction request to https://mainnet.radixdlt.com/construction with an UnregisterValidator action.
The response is read, the blobToSign is signed using the keystore private key, and a finalize transaction request is built and submitted to https://mainnet.radixdlt.com/construction again.
*Ensure that your validator wallet address has at least 5.1 XRD to successfully unregister the node. You will require an additional 5.1 XRD to re-register the node again later
import bech32
import ecdsa
import hashlib
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, NoEncryption
from cryptography.hazmat.backends import default_backend
from ecdsa.curves import SECP256k1
from ecdsa.util import sigencode_der
from getpass import getpass
import requests
import json
print ("Enter your Keystore Password:")
pw = getpass()
password = pw.encode()
# Read the Radix Keystore File (which is in PKCS12 format)
with open("node-keystore.ks", "rb") as f:
private_key, certificate, additional_certificates = pkcs12.load_key_and_certificates(f.read(), password, default_backend())
# Extract the unencrypted Private Key bytes
private_key_bytes = private_key.private_bytes(Encoding.DER, PrivateFormat.PKCS8, NoEncryption())
# Convert into Elliptic Curve Digital Signature Algorithm (ecdsa) private key object
private_key = ecdsa.SigningKey.from_der(private_key_bytes, hashfunc=hashlib.sha256)
# Derive public key from private key
verifying_key = private_key.get_verifying_key()
# Convert public key into compressed format so that we can generate the Validator Address
public_key_compressed_bytes = verifying_key.to_string("compressed")
public_key_compressed_bytes_hex = public_key_compressed_bytes.hex()
print("Validator Public Key (Compressed): ", public_key_compressed_bytes_hex)
# Generate Validator Address from the Compressed Public Key
public_key_bytes5 = bech32.convertbits(public_key_compressed_bytes, 8, 5)
validator_address = bech32.bech32_encode("rv", public_key_bytes5)
print("Validator Address: ", validator_address)
# Convert Compressed Public Key into a Radix Engine Address
readdr_bytes = b"\x04" + public_key_compressed_bytes
# Convert Radix Engine Address into Validator Wallet Address
readdr_bytes5 = bech32.convertbits(readdr_bytes, 8, 5)
validator_wallet_address = bech32.bech32_encode("rdx", readdr_bytes5)
print("Validator Wallet Address: ", validator_wallet_address)
# Construct RPC request
data = f"""
{{
"network_identifier": {{
"network": "mainnet"
}},
"actions": [
{{
"type": "UnregisterValidator",
"validator": {{
"address": "{validator_address}"
}}
}}
],
"fee_payer": {{
"address": "{validator_wallet_address}"
}},
"disable_token_mint_and_burn": true
}}
"""
print("Build Transaction Request JSON: ", data)
req = requests.Request('POST', 'https://mainnet.radixdlt.com/transaction/build', data=data)
prepared = req.prepare()
prepared.headers['Content-Type'] = 'application/json'
s = requests.Session()
# Send Request to Unregister Validator
resp = s.send(prepared)
# Get JSON Response
resp_json = resp.json()
print("Build Transaction Response JSON: \n", json.dumps(resp_json, indent=3))
# Extract fields from JSON Response
blob = resp_json['transaction_build']['unsigned_transaction']
blob_to_sign = resp_json['transaction_build']['payload_to_sign']
# Sign the blob_to_sign with the Keystore Private Key and convert to DER format
signature_der = private_key.sign_digest(bytearray.fromhex(blob_to_sign), sigencode=sigencode_der).hex()
# Finalize RPC Request
data = f"""
{{
"network_identifier":{{
"network": "mainnet"
}},
"unsigned_transaction": "{blob}",
"signature": {{
"bytes": "{signature_der}",
"public_key": {{
"hex": "{public_key_compressed_bytes_hex}"
}}
}},
"submit": true
}}
"""
print("Finalize Transaction Request JSON: ", data)
req = requests.Request('POST', 'https://mainnet.radixdlt.com/transaction/finalize', data=data)
prepared = req.prepare()
prepared.headers['Content-Type'] = 'application/json'
s = requests.Session()
# Send Request to Unregister Validator
resp = s.send(prepared)
# Get JSON Response
resp_json = resp.json()
print("Finalize Transaction Response JSON: \n", json.dumps(resp_json, indent=3))
Updates:
- 2022-02-25: Changed to use new Gateway API (Thanks @Faraz)
- 2022-08-03: Gateway host name updated to mainnet.radixdlt.com (Faraz)
This work by RadixPool.com is licensed under a Creative Commons Attribution 4.0 International License.