route53 hook for dehydrated - python2 / python3 + boto2 version. Tested on Ubuntu 16.04

route53.py

#!/usr/bin/env python

# How to use:
#
# Ubuntu 16.04: apt install -y python-boto OR apt install -y python3-boto
#
# Specify the default profile on aws/boto profile files or use the optional AWS_PROFILE env var:
# AWS_PROFILE=example ./dehydrated -c -d example.com -t dns-01 -k /etc/dehydrated/hooks/route53.py
#
# Manually specify hosted zone:
# HOSTED_ZONE=example.com ./dehydrated -c -d example.com -t dns-01 -k /etc/dehydrated/hooks/route53.py
#
# Manually specify the TXT record to update (useful when _acme-challenge.example.com is a CNAME not hosted on AWS):
# NAME_OVERRIDE=_acme-challenge.example.net ./dehydrated -c -d example.com -t dns-01 -k /etc/dehydrated/hooks/route53.py
#
# If the CNAME itself is also hosted on the same AWS account, the override is not necessary.
#
# More info about dehaydrated and dns challenge: https://github.com/lukas2511/dehydrated/wiki/Examples-for-DNS-01-hooks
# Using AWS Profiles: http://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html#cli-multiple-profiles
#
# This hook also works with dehydrated's HOOK_CHAIN="yes", which passes all domains in one invocation.
# It is recommended to use this hook with HOOK_CHAIN="yes" because it is faster to challenge domains in batch.
#
# This hook also works with wildcard certificates, and also when _acme-challenge is a CNAME.

import os
import sys
from boto.route53 import *
from time import sleep


USAGE_TEXT = "USAGE: route53.py CHALLENGE_TYPE DOMAIN TOKEN_FILENAME_IGNORED TOKEN_VALUE [DOMAIN TOKEN_FILENAME_IGNORED TOKEN_VALUE]..."


def get_zone_id(conn, domain):
    if 'HOSTED_ZONE' in os.environ:
        hosted_zone = "{0}.".format(os.environ['HOSTED_ZONE'])
        if not domain.endswith(hosted_zone):
            raise Exception("Incorrect hosted zone for domain {0}".format(domain))
        zone = conn.get_hosted_zone_by_name(hosted_zone)
        zone_id = zone['GetHostedZoneResponse']['HostedZone']['Id'].replace('/hostedzone/', '')
    else:
        zones = conn.get_all_hosted_zones()
        candidate_zones = []
        for zone in zones['ListHostedZonesResponse']['HostedZones']:
            if domain.endswith(zone['Name']):
                candidate_zones.append((domain.find(zone['Name']), zone['Id'].replace('/hostedzone/', '')))

        if len(candidate_zones) == 0:
            raise Exception("Hosted zone not found for domain {0}".format(domain))

        candidate_zones.sort()
        zone_id = candidate_zones[0][1]
    return zone_id


def wait_for_dns_update(conn, response, time_elapsed=0):
    timeout = 300
    sleep_time = 5
    st = status.Status(conn, response['ChangeResourceRecordSetsResponse']['ChangeInfo'])
    while st.update() != 'INSYNC' and time_elapsed <= timeout:
        print("Waiting for DNS change to complete... ({0}; elapsed {1} seconds)".format(st, time_elapsed))
        sleep(sleep_time)
        time_elapsed += sleep_time

    if st.update() != 'INSYNC' and time_elapsed > timeout:
        raise Exception("Timed out while waiting for DNS record to be ready. Waited {0} seconds but the last status was {1}".format(time_elapsed, st))

    print("DNS change completed")
    return time_elapsed


def route53_dns(domain_challenges_dict, action):
    action = action.upper()
    assert action in ['UPSERT', 'DELETE']

    conn = connection.Route53Connection()

    responses = []
    for domain, txt_challenges in domain_challenges_dict.items():

        print("domain: {0}".format(domain))
        print("txt_challenges: {0}".format(txt_challenges))

        name = u'_acme-challenge.{0}.'.format(domain) # note u'' and trailing . are important here for the == below
        if 'NAME_OVERRIDE' in os.environ:
            name = u'{0}.'.format(os.environ['NAME_OVERRIDE'])

        orig_name = name
        # Get existing record set, so we can add our challenges to it.
        # It's important that we add instead of override, to support dehydrated's HOOK_CHAIN="no",
        # (in which case we as the hook can't see all changes to make upfront).
        for attempt in xrange(5):
            zone_id = get_zone_id(conn, name)
            record_set = conn.get_all_rrsets(zone_id, name=name)

            cname_found = False
            for record in record_set:
                if record.name == name and record.type == "CNAME":
                    name = record.resource_records[0]
                    if not name.endswith(u'.'):
                        name = u'{0}.'.format(name)
                    cname_found = True
                    print("Found CNAME: {0} -> {1}".format(record.name, name))
            
            if not cname_found:
                break
        else:
            raise Exception("Too long chain of CNAMEs for {0}".format(orig_name))

        record_exists = False
        existing_quoted_txt_challenges = [] # include "" quotes already; not a set because 'DELETE' may care about order
        for record in record_set:
            if record.name == name and record.type == "TXT":
                record_exists = True
                existing_quoted_txt_challenges += record.resource_records

        if action == 'UPSERT':
            needed_quoted_txt_challenges = set('"{0}"'.format(c) for c in txt_challenges)
            all_quoted_txt_challenges = set(existing_quoted_txt_challenges) | needed_quoted_txt_challenges

            change = record_set.add_change('UPSERT', name, type='TXT', ttl=60)
            for txt_challenge in all_quoted_txt_challenges:
                change.add_value(txt_challenge)
            response = record_set.commit()
            responses.append(response)

        elif action == 'DELETE':
            if record_exists:
                change = record_set.add_change('DELETE', name, type='TXT', ttl=60)
                for txt_challenge in existing_quoted_txt_challenges:
                    change.add_value(txt_challenge)
                response = record_set.commit()
                # We don't block the hook to wait for deletion to complete.
                # responses.append(response)
            else:
                print("Challenge record " + name + " is already gone!")

    if responses != []:
        print("Waiting for all responses...")
        time_elapsed = 0
        for response in responses:
            time_elapsed = wait_for_dns_update(conn, response, time_elapsed)


def deploy_hook_args_to_domain_challenge_dict(hook_args):
    assert len(hook_args) % 3 == 0, "wrong number of arguments, hook arguments must be multiple of 3; " + USAGE_TEXT
    domain_dict = {}
    for i in xrange(0, len(hook_args), 3):
        domain = hook_args[i]
        txt_challenge = hook_args[i+2]
        domain_dict.setdefault(domain, []).append(txt_challenge)
    return domain_dict


if __name__ == "__main__":

    assert len(sys.argv) >= 2, "wrong number of arguments, need at least 1; " + USAGE_TEXT
    hook = sys.argv[1]

    print("hook: {0}".format(hook))

    if hook == "deploy_challenge":
        hook_args = sys.argv[2:]
        domain_challenges_dict = deploy_hook_args_to_domain_challenge_dict(hook_args)
        route53_dns(domain_challenges_dict, action='upsert')
    elif hook == "clean_challenge":
        hook_args = sys.argv[2:]
        domain_challenges_dict = deploy_hook_args_to_domain_challenge_dict(hook_args)
        route53_dns(domain_challenges_dict, action='delete')
    elif hook == "startup_hook":
        print("Ignoring startup_hook")
        exit(0)
    elif hook == "exit_hook":
        print("Ignoring exit_hook")
        exit(0)
    elif hook == "deploy_cert":
        print("Ignoring deploy_cert hook")
        exit(0)
    elif hook == "unchanged_cert":
        print("Ignoring unchanged_cert hook")
        exit(0)
    else:
        print("Ignoring unknown hook %s", hook)
        exit(0)

This doesn't work with python 3. https://stackoverflow.com/a/57389618/244008

Traceback (most recent call last):
  File "/usr/local/bin/route53.py", line 165, in <module>
    domain_challenges_dict = deploy_hook_args_to_domain_challenge_dict(hook_args)
  File "/usr/local/bin/route53.py", line 149, in deploy_hook_args_to_domain_challenge_dict
    for i in xrange(0, len(hook_args), 3):
NameError: name 'xrange' is not defined
ERROR: deploy_challenge hook returned with non-zero exit code

Replace xrange with range

This doesn't work with python 3. https://stackoverflow.com/a/57389618/244008

Diff to fix

diff --git a/route53.py b/route53.py
index eec1399..aa4e20a 100644
--- a/route53.py
+++ b/route53.py
@@ -90,7 +90,7 @@ def route53_dns(domain_challenges_dict, action):
         # Get existing record set, so we can add our challenges to it.
         # It's important that we add instead of override, to support dehydrated's HOOK_CHAIN="no",
         # (in which case we as the hook can't see all changes to make upfront).
-        for attempt in xrange(5):
+        for attempt in range(5):
             zone_id = get_zone_id(conn, name)
             record_set = conn.get_all_rrsets(zone_id, name=name)

@@ -146,7 +146,7 @@ def route53_dns(domain_challenges_dict, action):
 def deploy_hook_args_to_domain_challenge_dict(hook_args):
     assert len(hook_args) % 3 == 0, "wrong number of arguments, hook arguments must be multiple of 3; " + USAGE_TEXT
     domain_dict = {}
-    for i in xrange(0, len(hook_args), 3):
+    for i in range(0, len(hook_args), 3):
         domain = hook_args[i]
         txt_challenge = hook_args[i+2]
         domain_dict.setdefault(domain, []).append(txt_challenge)

Replace xrange with range

Yeah figured this out thanks heaps !