Upcoming Ethereum Validator Proposal Duties (Lighthouse)

check_validator_proposal_duties.py

# 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("-----------------------------")