Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Update route53 via AWS CLI with the current IP address. Usage: " subdomain1 subdomain2"
#!/usr/bin/env python3
import subprocess, re, os, json, sys
zones = None
def _sh(cmd):
return, shell=True, check=True, capture_output=True, text=True).stdout.strip()
def _whatsmyip():
return _sh('dig +short')
# 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')
return zones[domain]
def _route53(method, domain, params = {}):
qp = ['aws', 'route53', method, '--hosted-zone-id', _get_hosted_zone_id(domain)]
for key in params:
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}]
print('unknown destination', dest)
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]'"
def _get_elb_zone(elb, classic = False):
elbid = elb.split('-')[0]
if classic:
elbv = 'elb describe-load-balancers --load-balancer-names'
elbv = 'elbv2 describe-load-balancers --names'
rd = _sh(f'aws {elbv} {elbid}')
jd = json.loads(rd)
raise Exception(f'json {rd}')
if classic:
return jd['LoadBalancerDescriptions'][0]['CanonicalHostedZoneNameID']
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 =
if cur == dest:
print(f'Skipping; already set to {dest}')
# 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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment