Update route53 via AWS CLI with the current IP address. Usage: "update-route53.py subdomain1 subdomain2 domain-name.com"

update-route53.py

#!/usr/bin/env python3
import subprocess, re, os, json, sys

zones = None

def _sh(cmd):
  return subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True).stdout.strip()

def _whatsmyip():
  return _sh('dig +short myip.opendns.com @resolver1.opendns.com')

# Get the existing record set
def _record_set(subdomain, domain, t = 'A'):
  fqdn = _fqdn(subdomain, domain)
  j = json.loads(_route53('list-resource-record-sets', domain, {
    '--start-record-name': fqdn,
    # '--max-items': "1",
    '--query': f'"ResourceRecordSets[?Type == \'{t}\']"',
    '--output': 'json' # | jq -r \'.ResourceRecordSets[]\''
  }))
  j = [x for x in j if x['Name'] == '%s.' % fqdn]
  if len(j) == 0: return None
  if len(j) > 1: raise Exception('more than one record found: %s' % j)
  return j[0] # ['AliasTarget']['DNSName'] OR

def _fqdn(subdomain, domain):
  if len(subdomain) <= 0: return domain
  return '%s.%s' % (subdomain, domain)

# Use AWS to populate the global `zones` as a map of domain name => zone ID
# Then return the zone ID for this domain
def _get_hosted_zone_id(domain):
  global zones
  if not zones:
    zones = {}
    zd = json.loads(_sh('aws route53 list-hosted-zones'))['HostedZones']
    for z in zd:
      zones[z['Name'][:-1]] = z['Id'].split('/')[2]
  if not domain in zones:
    print(domain, 'is not hosted by route53')
    exit(-1)
  return zones[domain]

def _route53(method, domain, params = {}):
  qp = ['aws', 'route53', method, '--hosted-zone-id', _get_hosted_zone_id(domain)]
  for key in params:
    qp.append(key)
    qp.append(params[key])
  return _sh(' '.join(qp))

# Route some subdomain + domain to a given dest (IP or load balancer)
def _update(subdomain, domain, dest, t = 'A', ttl = 60):
  print(f'update: {subdomain} -- {domain} -- {dest}')
  fqdn = _fqdn(subdomain, domain)
  existing = _record_set(subdomain, domain)

  rs = { 'Name': fqdn, 'Type': t, 'TTL': ttl }
  if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", dest):
    if existing and 'ResourceRecords' in existing and existing['ResourceRecords'][0]['Value'] == dest:
      print(fqdn, 'is already pointed at', dest)
      return False
    rs['ResourceRecords'] = [{'Value': dest}]
  else:
    print('unknown destination', dest)
    exit(-1)

  update_data = {
    "Comment": "Dynamic DNS",
    "Changes": [ { "Action":"UPSERT", "ResourceRecordSet": rs } ]
  }

  tf = 'route53.json'
  with open(tf, 'w') as outfile:
    json.dump(update_data, outfile)

  print('update', existing, 'to', rs)
  return _route53('change-resource-record-sets', domain, {
    '--change-batch': 'file://' + tf,
    '--query': "'[ChangeInfo.Comment, ChangeInfo.Id, ChangeInfo.Status, ChangeInfo.SubmittedAt]'"
  })
  os.remove(tf)

def _get_elb_zone(elb, classic = False):
  elbid = elb.split('-')[0]
  if classic:
    elbv = 'elb describe-load-balancers --load-balancer-names'
  else:
    elbv = 'elbv2 describe-load-balancers --names'
  rd = _sh(f'aws {elbv} {elbid}')
  try:
    jd = json.loads(rd)
  except:
    raise Exception(f'json {rd}')
  if classic:
    return jd['LoadBalancerDescriptions'][0]['CanonicalHostedZoneNameID']
  else:
    return jd['LoadBalancers'][0]['CanonicalHostedZoneId']

if __name__ == "__main__":
  if len(sys.argv) < 3:
    raise Exception("Please provide at least one subdomain + domain.")
  cwd = os.path.dirname(__file__)

  subdomains = sys.argv[1:]
  domain = subdomains.pop()

  dest = _whatsmyip()
  cur = ''
  cachefn = os.path.join(cwd, f'{domain}.txt')
  if os.path.isfile(cachefn):
    with open(cachefn, 'r') as f: cur = f.read().strip()

  if cur == dest:
    print(f'Skipping; already set to {dest}')
    exit(0)

  # Route each of the hosts to a dest
  for subdomain in subdomains:
    _update(subdomain, domain, dest)

  with open(cachefn, 'w+') as f: f.write(dest)