Created
March 18, 2020 17:23
-
-
Save zaneclaes/64e4b0b3fbb4ffca830fd499c5fbe1f8 to your computer and use it in GitHub Desktop.
Update route53 via AWS CLI with the current IP address. Usage: "update-route53.py subdomain1 subdomain2 domain-name.com"
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment