Examples

The example scripts are located in the examples directory.

Local Key Manager

Import keystores - synchronous

import eth_2_key_manager_api_client

# Create lists of keystore and password strings, note: the order of the lists must match `passwords[i]` must unlock `keystores[i]`
list_of_keystore_strs: list[str] = []
list_of_keystore_password_strs: list[str] = []

keystore_files = [
    "mock_validator_keystores/keystore-m_12381_3600_0_0_0-1669980799.json",
    "mock_validator_keystores/keystore-m_12381_3600_1_0_0-1680087924.json",
    "mock_validator_keystores/keystore-m_12381_3600_2_0_0-1680087924.json",
]

# Read the keystores as strings
for keystore_file in keystore_files:
    with open(keystore_file, "r") as f:
        list_of_keystore_strs.append(f.read())

# Read the passwords as strings
keystore_password_files = [
    "mock_validator_keystores/keystore-m_12381_3600_0_0_0-1669980799.txt",
    "mock_validator_keystores/keystore-m_12381_3600_1_0_0-1680087924.txt",
    "mock_validator_keystores/keystore-m_12381_3600_2_0_0-1680087924.txt",
]

for keystore_password_file in keystore_password_files:
    with open(keystore_password_file, "r") as f:
        list_of_keystore_password_strs.append(f.read())

with open("mock_validator_keystores/slashing_protection_db.json", "r") as f:
    slashing_protection_str = f.read()

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager(timeout=10.0)

response = eth_2_key_manager.import_keystores.sync_detailed(list_of_keystore_strs, list_of_keystore_password_strs, slashing_protection_str)

if response.status_code == 200:
    print("Keystores imported successfully")
else:
    print(f"Keystores import failed with status code: {response.status_code}")
assert response.status_code == 200

Import keystores - async

import asyncio

import eth_2_key_manager_api_client
from tests.conftest import parse_file

validators = parse_file("../.env")

# Create lists of keystore and password strings, note: the order of the lists must match `passwords[i]` must unlock `keystores[i]`
list_of_keystore_strs: list[str] = []
list_of_keystore_password_strs: list[str] = []

keystore_files = [
    "mock_validator_keystores/keystore-m_12381_3600_0_0_0-1669980799.json",
    "mock_validator_keystores/keystore-m_12381_3600_1_0_0-1680087924.json",
    "mock_validator_keystores/keystore-m_12381_3600_2_0_0-1680087924.json",
]

# Read the keystores as strings
for keystore_file in keystore_files:
    with open(keystore_file, "r") as f:
        list_of_keystore_strs.append(f.read())

# Read the passwords as strings
keystore_password_files = [
    "mock_validator_keystores/keystore-m_12381_3600_0_0_0-1669980799.txt",
    "mock_validator_keystores/keystore-m_12381_3600_1_0_0-1680087924.txt",
    "mock_validator_keystores/keystore-m_12381_3600_2_0_0-1680087924.txt",
]

for keystore_password_file in keystore_password_files:
    with open(keystore_password_file, "r") as f:
        list_of_keystore_password_strs.append(f.read())

with open("mock_validator_keystores/slashing_protection_db.json", "r") as f:
    slashing_protection_str = f.read()


async def import_keystores_async(validator):

    eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager(base_url=validator[1], token=validator[2])

    response = await eth_2_key_manager.import_keystores.asyncio_detailed(list_of_keystore_strs, list_of_keystore_password_strs, slashing_protection_str)

    if response.status_code == 200:
        print(f"{validator[0]} - {validator[1]} - Keystores imported successfully")
    else:
        print(f"{validator[0]} - {validator[1]} - Keystores import failed with status code: {response.status_code}")


async def main():
    await asyncio.gather(*(import_keystores_async(validator) for validator in validators))


asyncio.run(main())

List keys - synchronous

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
response = eth_2_key_manager.list_keys.sync_detailed()

if response.status_code == 200:
    print(f"Validator: {eth_2_key_manager.client.base_url}")
    assert isinstance(response.parsed, eth_2_key_manager_api_client.models.list_keys_response.ListKeysResponse)
    print(f"Number of keys: {len(response.parsed.data)}")
    if response.parsed.data:
        print("List of keys:")
        for key in response.parsed.data:
            print(f"    {key.validating_pubkey}")
else:
    print(f"List keys failed with status code: {response.status_code}")

List keys - async

import asyncio

import eth_2_key_manager_api_client
from tests.conftest import parse_file

validators = parse_file("../.env")


async def list_keys_async(validator):

    eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager(base_url=validator[1], token=validator[2])

    response = await eth_2_key_manager.list_keys.asyncio_detailed()

    if response.status_code == 200:
        print(f"{validator[0]} - {validator[1]} - Number of keys: {len(response.parsed.data)}")
        if response.parsed.data:
            print("List of keys:")
            for key in response.parsed.data:
                print(f"    {key.validating_pubkey}")
    else:
        print(f"{validator[0]} - {validator[1]} - List keys failed with status code: {response.status_code}")


async def main():
    await asyncio.gather(*(list_keys_async(validator) for validator in validators))


asyncio.run(main())

Delete keys - synchronous

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x874bed7931ba14832198a4070b881f89e7ddf81898dd800446ef382344e9726a5e6265acb21f5c8ee2759c313ec6ca0d"

response = eth_2_key_manager.delete_keys.sync_detailed(pubkeys=[pubkey])
if response.status_code == 200:
    print("Keystore deleted successfully")
else:
    print(f"Keystore delete failed with status code: {response.status_code}")
assert response.status_code == 200

Delete keys - async

import asyncio

import eth_2_key_manager_api_client
from tests.conftest import parse_file

validators = parse_file("../.env")

pubkeys = [
    "0x874bed7931ba14832198a4070b881f89e7ddf81898dd800446ef382344e9726a5e6265acb21f5c8ee2759c313ec6ca0d",
    "0x88a471158d618a8f9997dcb2cc1921411392d82d00e339ccf912fd9335bd42f97c9de046280d9d5f681a8e73a7d3baad",
    "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b",
]


async def delete_keys_async(validator):

    eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager(base_url=validator[1], token=validator[2])

    response = await eth_2_key_manager.delete_keys.asyncio_detailed(pubkeys=pubkeys)

    if response.status_code == 200:
        print(f"{validator[0]} - {validator[1]} - Keystore deleted successfully")
    else:
        print(f"{validator[0]} - {validator[1]} - Keystore delete failed with status code: {response.status_code}")


async def main():
    await asyncio.gather(*(delete_keys_async(validator) for validator in validators))


asyncio.run(main())

Gas limit

Set gas limit

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b"
response = eth_2_key_manager.set_gas_limit.sync_detailed(pubkey=pubkey, gas_limit="999999")
if response.status_code == 202:
    print("Gas limit set successfully")
else:
    print(f"Gas limit set failed with status code: {response.status_code}")
assert response.status_code == 202

Get gas limit

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b"
response = eth_2_key_manager.get_gas_limit.sync_detailed(pubkey=pubkey)
if response.status_code == 200:
    assert isinstance(response.parsed, eth_2_key_manager_api_client.models.list_gas_limit_response.ListGasLimitResponse)
    print(f"Gas limit for pubkey {pubkey} is {response.parsed.data.gas_limit}")
else:
    print(f"Gas limit listing failed with status code: {response.status_code}")
assert response.status_code == 200

Delete gas limit

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b"
response = eth_2_key_manager.delete_gas_limit.sync_detailed(pubkey=pubkey)
if response.status_code == 204:
    print("Gas limit deleted successfully")
else:
    print(f"Gas limit deletion failed with status code: {response.status_code}")
assert response.status_code == 204

Fee recipient

Set fee recipient

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b"
response = eth_2_key_manager.set_fee_recipient.sync_detailed(pubkey=pubkey, ethaddress="0xabcf8e0d4e9587369b2301d0790347320302cc09")
if response.status_code == 202:
    print("Fee Recipient set successfully")
else:
    print(f"Fee Recipient set failed with status code: {response.status_code}")
assert response.status_code == 202

List fee recipient

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b"
response = eth_2_key_manager.list_fee_recipient.sync_detailed(pubkey=pubkey)
if response.status_code == 200:
    assert isinstance(response.parsed, eth_2_key_manager_api_client.models.list_fee_recipient_response.ListFeeRecipientResponse)
    print(f"Fee recipient for pubkey {pubkey} is {response.parsed.data.ethaddress}")
else:
    print(f"Fee Recipient listing failed with status code: {response.status_code}")
assert response.status_code == 200

Delete fee recipient

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x99c4c42fac7d1393956bd9e2785ed67cf5aaca4bf56d2fcda94c42d6042aebb1723ce6bac6f0216ff8c5d4f9f013008b"
response = eth_2_key_manager.delete_fee_recipient.sync_detailed(pubkey=pubkey)
if response.status_code == 204:
    print("Fee Recipient deleted successfully")
else:
    print(f"Fee Recipient deletion failed with status code: {response.status_code}")
assert response.status_code == 204

Remote Key Manager

Import remote keys - synchronous

import eth_2_key_manager_api_client

remote_keys = [
    {
        "pubkey": "0x93247f2209abcacf57b75a51dafae777f9dd38bc7053d1af526f220a7489a6d3a2753e5f3e8b1cfe39b56f43611df74a",
        "url": "https://remote.signer",
    }
]

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
response = eth_2_key_manager.import_remote_keys.sync_detailed(remote_keys=remote_keys)

if response.status_code == 200:
    print("Remote keys imported successfully")
else:
    print(f"Remote keys import failed with status code: {response.status_code}")

Import remote keys - async

import asyncio

import eth_2_key_manager_api_client
from tests.conftest import parse_file

validators = parse_file("../.env")

# validators = [
#     (
#         "LIGHTHOUSE",
#         "http://192.168.121.71:7500",
#         "api-token-0x024fa0a0c597e83a970e689866703a89a555c9ee602d08cf848ee7935509a62f33"
#     ),
#     (
#         "TEKU",
#         "https://192.168.121.245:7500",
#         "5b874584cecab7eadfc3a224f177272f"
#     ),
#     (
#         "NIMBUS",
#         "http://192.168.121.61:7500",
#         "fcb4869b587a71b6d7ff25c85ac312201e53"
#     ),
#     (
#         "LODESTAR",
#         "http://192.168.121.118:7500",
#         "api-token-0xff6b738e030ee41e6bc317c204e2dae8965f6d666dc158e5690940936eeb35d9"
#     )
# ]

remote_keys = [{"pubkey": "0x93247f2209abcacf57b75a51dafae777f9dd38bc7053d1af526f220a7489a6d3a2753e5f3e8b1cfe39b56f43611df74a", "url": "https://remote.signer"}]


async def import_remote_keys_async(validator):

    eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager(base_url=validator[1], token=validator[2])

    response = await eth_2_key_manager.import_remote_keys.asyncio_detailed(remote_keys=remote_keys)

    if response.status_code == 200:
        print(f"{validator[0]} - {validator[1]} - Remote keys imported successfully")
    else:
        print(f"{validator[0]} - {validator[1]} - Remote keys import failed with status code: {response.status_code}")


async def main():
    await asyncio.gather(*(import_remote_keys_async(validator) for validator in validators))


asyncio.run(main())

List remote keys - synchronous

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
response = eth_2_key_manager.list_remote_keys.sync_detailed()

if response.status_code == 200:
    print(f"Validator: {eth_2_key_manager.client.base_url}")

    # Assert that the response is of the expected type to satisfy static type checking.
    # Ref: https://mypy.readthedocs.io/en/stable/error_code_list.html#check-that-attribute-exists-in-each-union-item-union-attr
    assert isinstance(response.parsed, eth_2_key_manager_api_client.models.list_remote_keys_response.ListRemoteKeysResponse)

    print(f"Number of remote keys: {len(response.parsed.data)}")
    if response.parsed.data:
        print("List of keys:")
        for key in response.parsed.data:
            print(f"    {key.pubkey}")
else:
    print(f"List remote keys failed with status code: {response.status_code}")

List remote keys - async

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
response = eth_2_key_manager.list_remote_keys.sync_detailed()

if response.status_code == 200:
    print(f"Validator: {eth_2_key_manager.client.base_url}")

    # Assert that the response is of the expected type to satisfy static type checking.
    # Ref: https://mypy.readthedocs.io/en/stable/error_code_list.html#check-that-attribute-exists-in-each-union-item-union-attr
    assert isinstance(response.parsed, eth_2_key_manager_api_client.models.list_remote_keys_response.ListRemoteKeysResponse)

    print(f"Number of remote keys: {len(response.parsed.data)}")
    if response.parsed.data:
        print("List of keys:")
        for key in response.parsed.data:
            print(f"    {key.pubkey}")
else:
    print(f"List remote keys failed with status code: {response.status_code}")

Delete remote keys - synchronous

import eth_2_key_manager_api_client

eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager()
pubkey = "0x93247f2209abcacf57b75a51dafae777f9dd38bc7053d1af526f220a7489a6d3a2753e5f3e8b1cfe39b56f43611df74a"
response = eth_2_key_manager.delete_remote_keys.sync_detailed(pubkeys=[pubkey])

if response.status_code == 200:
    print("Remote keys deleted successfully")
else:
    print(f"Remote keys delete failed with status code: {response.status_code}")

Delete remote keys - async

import asyncio

import eth_2_key_manager_api_client
from tests.conftest import parse_file

validators = parse_file("../.env")

pubkey = "0x93247f2209abcacf57b75a51dafae777f9dd38bc7053d1af526f220a7489a6d3a2753e5f3e8b1cfe39b56f43611df74a"


async def delete_remote_keys_async(validator):

    eth_2_key_manager = eth_2_key_manager_api_client.Eth2KeyManager(base_url=validator[1], token=validator[2])

    response = await eth_2_key_manager.delete_remote_keys.asyncio_detailed(pubkeys=[pubkey])

    if response.status_code == 200:
        print(f"{validator[0]} - {validator[1]} - Remote keys deleted successfully")
    else:
        print(f"{validator[0]} - {validator[1]} - Remote keys delete failed with status code: {response.status_code}")


async def main():
    await asyncio.gather(*(delete_remote_keys_async(validator) for validator in validators))


asyncio.run(main())