AWS hook for letsencrypt.sh

letsencrypt-hook-aws.py

#!/usr/bin/env python -u

from __future__ import print_function
import os, sys, re, boto3
route53 = boto3.client('route53')
iam = boto3.client('iam')

def zone_cmp(first, second):
  return first['Name'].count('.') - second['Name'].count('.')

def hosted_zones(name):
  resp = route53.list_hosted_zones()
 
  while True:
    for zone in resp['HostedZones']:
      zone_name = zone['Name'][0:-1]
      if name == zone_name or name.endswith('.%s' % zone_name):
        yield zone

    if not resp['IsTruncated']:
      break

    resp = route53.list_hosted_zones_by_name(Marker=resp['NextMarker'])

def make_change(action, domain, token, wait=False):
  zones = sorted(hosted_zones(domain), cmp=zone_cmp, reverse=True)
  if len(zones) == 0:
    print('error: no zones found for \'%s\'' % domain, file=sys.stderr)
    sys.exit(2)

  zone = zones[0]
  resp = route53.change_resource_record_sets(
      HostedZoneId=zone['Id'],
      ChangeBatch={
        'Changes': [
          {
            'Action': action,
            'ResourceRecordSet': {
              'Name': '_acme-challenge.%s' % domain,
              'Type': 'TXT',
              'TTL': 60,
              'ResourceRecords': [
                {
                  'Value': '"%s"' % token  
                }
              ]
            }
          }
        ]  
      }
    )

  print(resp['ChangeInfo'])
  if wait:
    print('Waiting for change...' , end='')
    waiter = route53.get_waiter('resource_record_sets_changed')
    waiter.wait(Id=resp['ChangeInfo']['Id'])
    print(' OK')

def read_file(path):
  with open(path, 'r') as io:
    return io.read()

def deploy_challenge(domain, challenge_token, token):
  make_change('UPSERT', domain, token, wait=True)

def clean_challenge(domain, challenge_token, token):
  make_change('DELETE', domain, token)

def deploy_cert(domain, keyfile, certfile, chainfile):
  ts = re.search(r'-(.+?)\.pem$', os.readlink(certfile))

  resp = iam.upload_server_certificate(
    ServerCertificateName='%s-%s' % (domain, ts.group(1)),
    CertificateBody=read_file(certfile),
    PrivateKey=read_file(keyfile),
    CertificateChain=read_file(chainfile.replace('/full', '/'))
  )

  print(resp['ServerCertificateMetadata'])

def main(op, *args):
  if op == 'deploy_challenge':
    deploy_challenge(*args)
  elif op == 'clean_challenge':
    clean_challenge(*args)
  elif op == 'deploy_cert':
    deploy_cert(*args)
  else:
    print('error: unsupported operation \'%s\'' % op, file=sys.stderr)
    sys.exit(1)

if __name__ == "__main__":
  main(*sys.argv[1:])