Skip to content

Instantly share code, notes, and snippets.

@alex-miller-0
Last active April 20, 2023 19:17
Show Gist options
  • Save alex-miller-0/52385d4666ae57913a5f37d73ce13b09 to your computer and use it in GitHub Desktop.
Save alex-miller-0/52385d4666ae57913a5f37d73ce13b09 to your computer and use it in GitHub Desktop.
Upcoming Ethereum Validator Proposal Duties (Lighthouse)
# Validator proposals are determined two epochs (~12 min) in advance, so we
# can do a quick check and see if any of our validators have upcoming proposals.
# This is helpful for determining when to restart our consensus client, as it
# will force the validators offline for two epochs.
# NOTE: This was designed for use with Lighthouse. It will need to be updated
# for use with other clients.
from datetime import datetime, timezone
from math import trunc
import json
import os
import urllib.request
# Define the beacon node API location
BN_API_URL = "http://127.0.0.1:" + (os.environ.get("BN_API_PORT") or str(5052))
# Define the validator API location
VAL_API_URL = "http://127.0.0.1:" + (os.environ.get("VAL_API_PORT") or str(5062))
# Make a GET HTTP request
def GET(url, headers=None):
if headers is None:
req = urllib.request.Request(url)
else:
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req) as response:
data = json.loads(response.read().decode("utf-8"))
return data
# Get the current UTC timestamp using system clock
def get_current_utc_timestamp():
dt = datetime.now(timezone.utc)
return trunc(dt.replace(tzinfo=timezone.utc).timestamp())
# Get the current epoch by calculating distance from genesis time
# (12 s/slot, 32 slots/epoch)
def get_current_epoch():
genesis = int(GET(BN_API_URL + "/eth/v1/beacon/genesis")["data"]["genesis_time"])
return trunc((get_current_utc_timestamp() - genesis) / (32 * 12))
# Get the authorization token so that we can make protected request
# to the validator API. This secret should be in a local file.
def get_val_api_token():
token_loc = GET(VAL_API_URL + "/lighthouse/auth")["token_path"]
with open(token_loc) as f:
token = f.readline()
f.close()
return token
# Get a set of currently known validator public keys
def get_validator_pubkeys():
token = get_val_api_token()
headers = { "Authorization": "Basic " + token }
validators = GET(VAL_API_URL + "/lighthouse/validators", headers)["data"]
val_pubkeys = []
for val in validators:
val_pubkeys.append(val["voting_pubkey"])
return val_pubkeys
# Get a list of validators that will make a proposal in a
# given epoch, if any
def get_proposals(epoch, validator_pubkeys):
# First get all validators that will make a proposal in this epoch
all_proposals = GET(BN_API_URL + "/eth/v1/validator/duties/proposer/" + str(epoch))["data"]
# Cross reference with our validator pubkeys
proposals = []
for prop in all_proposals:
if prop["pubkey"] in validator_pubkeys:
proposals.append(prop)
return proposals
############
# The script
############
# First, get our validators
validators = get_validator_pubkeys()
# Get the current epoch
epoch = get_current_epoch()
# We can only check this and the next epoch, as duties beyond
# next epoch have not yet been assigned
e0_props = get_proposals(epoch, validators)
e1_props = get_proposals(epoch + 1, validators)
# Print stuff if we found upcoming proposals
print("-----------------------------")
print(
("💰" if len(e0_props) > 0 else "") + str(len(e0_props)) + " proposals in current epoch"
)
print(
("💰" if len(e1_props) > 0 else "") + str(len(e1_props)) + " proposals in next epoch"
)
print("-----------------------------")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment