letsenncrypt_dns_hook.py

#! /usr/bin/env python3
# vim: set fileencoding=UTF-8

""" letsencrypt_dns_hook.py: Used as a hook script for dehydrated.sh during
validation for a given name to modify the route53 DNS records necessary to
provide proof of domain ownership.
"""

__author__    = "Zach Leslie"
__copyright__ = "Copyright 2016, OtoAnalytics"
__license__   = "Apache2"
__version__   = "1.0.1"
__maintainer__ = "Zach Leslie"


import logging
import sys
import boto3
import time


class LetsEncryptRoute53HookThing():
    def __init__(self, *, certname, token, external_zone, external_zone_id):
        self.external_zone = external_zone
        self.external_zone_id = external_zone_id

        self.route53 = boto3.client('route53')
        self.hostname, self.domain = certname.split(self.external_zone, 1)
        self.fakezone = certname.split('.', 1)[-1]
        self.certname = certname
        self.token = token

    def create_txt_record(self):
        results = self.route53.change_resource_record_sets(
            HostedZoneId=self.external_zone_id,
            ChangeBatch={
                'Changes': [
                    {
                        'Action': 'UPSERT',
                        'ResourceRecordSet': {
                            'Name': '_acme-challenge.%s' % self.fakezone,
                            'Type': 'TXT',
                            'TTL': 0,
                            'ResourceRecords': [
                                {
                                    'Value': '"%s"' % str(self.token)
                                }
                            ]
                        }
                    },
                    {
                        'Action': 'UPSERT',
                        'ResourceRecordSet': {
                            'Name': '_acme-challenge.%s' % self.certname,
                            'Type': 'TXT',
                            'TTL': 0,
                            'ResourceRecords': [
                                {
                                    'Value': '"%s"' % str(self.token)
                                }
                            ]
                        }
                    }
                ]
            }
        )

        logging.info('sleeping 20 seconds')
        time.sleep(20)

    def delete_txt_record(self):
        results = self.route53.change_resource_record_sets(
            HostedZoneId=self.external_zone_id,
            ChangeBatch={
                'Changes': [
                    {
                        'Action': 'DELETE',
                        'ResourceRecordSet': {
                            'Name': '_acme-challenge.%s' % self.fakezone,
                            'Type': 'TXT',
                            'TTL': 0,
                            'ResourceRecords': [
                                {
                                    'Value': '"%s"' % str(self.token)
                                }
                            ]
                        }
                    },
                    {
                        'Action': 'DELETE',
                        'ResourceRecordSet': {
                            'Name': '_acme-challenge.%s' % self.certname,
                            'Type': 'TXT',
                            'TTL': 0,
                            'ResourceRecords': [
                                {
                                    'Value': '"%s"' % str(self.token)
                                }
                            ]
                        }
                    }
                ]
            }
        )

    def deploy_certificate(self):
        pass

    def unchanged_cert(self):
        pass


def main(args):
    logger = logging.getLogger()
    print(args)

    operation = args[1]
    certname = args[2]
    token = args[4]

    le = LetsEncryptRoute53HookThing(
        certname=certname,
        token=token,
        external_zone='example.com',
        external_zone_id='1234',
    )

    ops = {
        'deploy_challenge': le.create_txt_record,
        'clean_challenge': le.delete_txt_record,
        'deploy_cert': le.deploy_certificate,
        'unchanged_cert': le.unchanged_cert,
    }

    logger.info('route53 hook executing %s' % operation)
    ops[operation]()


if __name__ == '__main__':
    main(sys.argv)