Poor Man's Route53 DDNS

route53_ddns.py

'''
Run this script periodically to automagically update
a domain name with your current public IP. Schedule
this to run periodically on a Raspberry Pi for a
poor man's Dynamic DNS (DDNS).

Run "python route53_ddns.py --help" for more info.
'''
import boto3
import requests
import time

def update_dns(hosted_zone, domain_name, profile_name=None, what_if=False):
    route53 = boto3.Session(profile_name=profile_name).client('route53')
    my_ip = get_my_public_ip()
    current_dns_ip, current_dns_ttl = get_dns(route53, hosted_zone, domain_name)
    if my_ip != current_dns_ip:
        message = f'Updating {domain_name} from {current_dns_ip} to {my_ip}'
        if what_if:
            print('What if:', message)
        else:
            print(message)
            start = time.time()
            set_dns_ip(route53, hosted_zone, domain_name, my_ip, current_dns_ttl)
            print('Took', int(time.time() - start), 'seconds')
    else:
        print(f'{domain_name} is already set to {my_ip}. Skipping...')

def get_my_public_ip():
    response = requests.get('https://api.ipify.org/?format=json')
    response.raise_for_status()
    return response.json()['ip']

def get_dns(route53, hosted_zone, domain_name):
    paginator = route53.get_paginator('list_resource_record_sets')
    page_iterator = paginator.paginate(
        HostedZoneId=hosted_zone,
        StartRecordName=domain_name,
        StartRecordType='A',
        PaginationConfig={
            'MaxItems': 1,
        },
    )
    filtered_iterator = page_iterator.search(f'ResourceRecordSets[?Name == `{domain_name}.` && Type == `A`]')
    records = list(filtered_iterator)
    if not records:
        raise Exception(f'Record not found. Create a Route53 A record for {domain_name} before running this script.')
    return records[0]['ResourceRecords'][0]['Value'], records[0]['TTL']

def set_dns_ip(route53, hosted_zone, domain_name, new_ip, ttl):
    response = route53.change_resource_record_sets(
        HostedZoneId=hosted_zone,
        ChangeBatch={
            'Changes': [{
                'Action': 'UPSERT',
                'ResourceRecordSet': {
                    'Name': domain_name,
                    'Type': 'A',
                    'TTL': ttl,
                    'ResourceRecords': [{ 'Value': new_ip }],
                },
            }],
        },
    )
    waiter = route53.get_waiter('resource_record_sets_changed')
    waiter.wait(
        Id=response['ChangeInfo']['Id'],
        WaiterConfig={
            'Delay': 10,
            'MaxAttempts': 30,
        },
    )

def _parse_args():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument('-z', '--zone', default='Z3M3LMPEXAMPLE', help='the Route53 hosted zone id')
    parser.add_argument('-d', '--domain', default='home.example.com', help='the domain name')
    parser.add_argument('--profile', default=None, help='the AWS named profile with access to Route53')
    parser.add_argument('--whatif', action='store_true', help='displays a message that describes the effect of the script instead of executing the script')
    return parser.parse_args()

if __name__ == '__main__':
    args = _parse_args()
    update_dns(args.zone, args.domain, args.profile, args.whatif)