path: root/acme_dns_tiny.py

#!/usr/bin/env python3
#
"""\
Tiny ACME client to get TLS certificate by responding to DNS challenges. It
uses Let's Encrypt servers by default.

As the script requires access to your private ACME account key and dns server,
please read through it. It's about 326 lines, 225 SLOC with no line
exceeding 80cols.

Usage:
  acme_dns_tiny.py [--contact=MAIL]... [--quiet] [--verbose]
                   [--acme-directory=URL] [--ttl=SECONDS]
                   [--separator=STR]
                   (--account-key=PATH) (--csr=PATH) (--script=PATH)
  acme_dns_tiny.py -h|--help

Options:

  --account-key=PATH    path to your Let's Encrypt account private key
  --csr=PATH            path to your certificate signing request
  --quiet               suppress output except for errors
  --verbose             show all debug information on stderr
  --acme-directory=URL  where to make acme request, default is Let's Encrypt
  --contact=MAIL        create/update contact info (ex "mailto:x@example.com")
  --ttl=SECONDS         time before (re)try self check
  --separator=STR       list all chains joined by STR if specified
  --script=PATH         script to run to update the DNS server record

Called script:

Script is called with the following arguments: action zone challenge

  action     can be "add" or "delete"
  zone       in the form _acme-challenge.domain. domains are extracted from CSR
  challenge  given by the ACME server, to be set as a TXT record

Example:

|  path/to/ \\
|        --account-key=./account.key \\
|        --csr=./domain.csr \\
|        --contact=contact@domain \\
|        --ttl=300 \\
|        --script=./update-dns-record > signed.crt

Example Crontab Renewal (once per month):

0 0 1 * * path/to/acme_dns_tiny.py --account-key /path/to/account.key \
  --csr /path/to/domain.csr > /path/to/signed.crt \
  2>> /var/log/acme_tiny.log

Example of called script:

#!/bin/bash
case "$1" in
    add) ssh authoritative.dns.host nsupdate \
            <<<$"server ::1\\nupdate add $2 60 TXT $3\\nsend";;
    delete) ssh authoritative.dns.host nsupdate \
            <<<$"server ::1\\nupdate delete $2 TXT";;
esac
"""


import subprocess
import json
import base64
import time
import hashlib
import re
import logging
import collections
import contextlib

import docopt
import requests
import dns
import dns.resolver as resolver


ACME_DEFAULT_DIRECTORY = 'https://acme-v02.api.letsencrypt.org/directory'

# consistent logging: global instance share accross all this module
log = logging.getLogger('acme_dns_tiny')
log_configure = True # useful for unit tests


def b64(b):
    """Encodes string as base64 as specified in ACME RFC"""
    return base64.urlsafe_b64encode(b).decode("utf8").rstrip("=")

def b64sha256(payload):
    return b64(hashlib.sha256(payload.encode('utf8')).digest())

def b64json(obj):
    return b64(json.dumps(obj).encode("utf8"))

def openssl(subcommand, stdin=None):
    """Run openssl command line and raise IOError on non-zero return."""
    binary = True if '-sign' in subcommand or 'DER' in subcommand else False
    return subprocess.run(
        ['openssl'] + subcommand,
        input=stdin.encode('utf8') if binary and stdin is not None else stdin,
        stdout=subprocess.PIPE,
        check=True,
        encoding=None if binary else 'utf8',
    ).stdout

def dns_script(script=None, action=None, zone=None, content=None):
    log.info('Calling script: "%s %s %s %s"', script, action, zone, content)
    subprocess.run([script, action, zone, content], check=True)

def extract_domains_from_csr(csr_path):
    log.info('Extract domains from CSR.')
    csr = openssl(['req', '-in', csr_path, '-noout', '-text'])
    san = re.findall(
            r'X509v3 Subject Alternative Name:\s+([^\r\n]+)\r?\n',
            csr, re.MULTILINE)
    domains = set(re.findall(r'Subject:.*?\s+?CN\s*?=\s*?([^\s,;/]+)', csr)
            + re.findall(r'DNS:\s*(\S+)\s*(?:,|$)', san and san[0] or ''))
    if not domains:
        raise ValueError('No domain to validate in the provided CSR.')
    return domains

def self_challenge_check(*, domain=None, ttl=None, challenge=None):
    log.info('Self testing %s with challenge "%s"', domain, challenge)
    for checknum in range(10):
        log.info('Sleep 1 TTL (%ss) before self check', ttl)
        time.sleep(ttl)
        log.debug('Self check number %s', checknum)
        try:
            try:
                responses = resolver.resolve(domain, rdtype='TXT').rrset
            except AttributeError:
                # backward compatibility for older dns dnspython releases
                responses = resolver.query(domain, rdtype='TXT').rrset
        except dns.exception.DNSException as e:
            log.debug('  - DNS error: %s: %s', type(e).__name__, e)
            continue
        responses = [x.to_text() for x in responses]
        log.debug('  - Found values: %s', responses)
        if f'"{challenge}"' in responses:
            break
    else:
        raise ValueError(f'Challenge not found: {challenge}')


class ACME:

    def __init__(self, *, account_key_path=None):
        self.directory = None
        self.account_key_path = account_key_path
        self.dirheaders = headers = {
            'User-Agent': 'acme-dns-tiny/2.1',
            'Accept-Language': 'en',
        }
        self.headers = {**headers, 'Content-Type': 'application/jose+json'}
        self.jws_header = None
        self.jwk_thumbprint = None

    def generate_jws_header(self, account_key_path):
        log.debug('Read account key.')
        out = openssl(['pkey', '-in', account_key_path, '-noout', '-text'])
        pub_hex, pub_exp = re.search(
            r'modulus:\r?\n\s+00:([a-f0-9\:\s]+?)'
            r'\r?\npublicExponent: ([0-9]+)',
            out, re.MULTILINE | re.DOTALL).groups()
        self.jws_header = header = {
            'alg': 'RS256',
            'jwk': {
                'e': b64(int(pub_exp).to_bytes(100, 'big').lstrip(b'\0')),
                'kty': 'RSA',
                'n': b64(bytes.fromhex(re.sub(r'(\s|:)', '', pub_hex))),
            },
            'nonce': None,
        }
        self.jwk_thumbprint = b64sha256(json.dumps(header['jwk'],
            sort_keys=True, separators=(',', ':')))

    def request_new_nonce(self):
        log.info('Request new nonce')
        self.jws_header['nonce'] = requests.get(self.directory["newNonce"]
                ).headers['Replay-Nonce']

    def init_sreqs(self, directory_url=None):
        self.generate_jws_header(self.account_key_path)
        log.info('Fetch information from the ACME directory.')
        self.directory = dirmap = requests.get(directory_url,
                headers=self.dirheaders).json()
        self.request_new_nonce()

    def sreq(self, url=None, payload=None, headers=None, max_retry=10):
        """Sends signed requests to ACME server."""
        if max_retry < 0:
            raise ValueError('max_retry exceeded')
        # POST-as-GET (payload=None != {}), payload64 is an empty string
        payload64   = '' if payload is None else b64json(payload)
        protected64 = b64json({**self.jws_header, 'url': url})
        jose = {
            'protected':    protected64,
            'payload':      payload64,
            'signature':    b64(openssl(["dgst", "-sha256", "-sign",
                self.account_key_path], f'{protected64}.{payload64}'))
        }
        try:
            req = requests.post(url, json=jose,
                    headers={**self.headers, **(headers or {})})
        except requests.exceptions.RequestException as error:
            req = error.response
        self.jws_header['nonce'] = req.headers['Replay-Nonce']
        jmap = {}
        with contextlib.suppress(ValueError):
            jmap = req.json()
        if req.status_code not in (200, 201):
            if jmap['type'] == 'urn:ietf:params:acme:error:badNonce':
                log.info('nonce expired or invalid, retry (left %d)', max_retry)
                self.request_new_nonce()
                return self.sreq(url, payload, headers, max_retry-1)
            raise ValueError(f'Request failed: {req.status_code} {jmap}, '
                'hint: {sreq.headers.get("Link", "empty")}.')
        return collections.namedtuple('sreq', ['code', 'headers', 'map',
            'text'])(req.status_code, req.headers, jmap, req.text)

    def register_account(self, *, contacts=None):
        log.info('Register/Login ACME Account.')
        srv_terms = self.directory.get('meta', {}).get('termsOfService', '')
        if srv_terms:
            log.warning('Terms of service auto agreed: %s', srv_terms)
        account_request = {
            **dict({'termsOfServiceAgreed': True} if srv_terms else {}),
            **dict({'contact': contacts} if contacts else {}),
        }
        sreq = self.sreq(self.directory['newAccount'], account_request)
        self.jws_header['kid'] = kid = sreq.headers['Location']
        del self.jws_header['jwk']
        if sreq.code == 201:
            log.info('  - Registered a new account: "%s"', kid)
        elif sreq.code == 200:
            log.debug('  - Account is already registered: "%s"', kid)
            sreq = self.sreq(self.jws_header['kid'], {})
        if contacts and (set(contacts) != set(sreq.map['contact'])):
            self.sreq(self.jws_header["kid"], account_request)
            log.info('  - Account updated with latest contact information.')

    def new_order(self, *, domains=None):
        log.info('Request a new ACME order.')
        sreq = self.sreq(self.directory['newOrder'], {'identifiers':
            [{'type': 'dns', 'value': x} for x in domains]})
        # note: standard says only 201 is returned, assume 200 is not ok
        if sreq.code != 201 or sreq.map['status'] not in ('pending', 'ready'):
            raise ValueError(f'newOrder failed: {sreq.code} {sreq.map}')
        order_location = sreq.headers['Location']
        log.debug('  - Order received: %s', order_location)
        if sreq.map['status'] == 'ready':
            log.info('No challenge to process: order is already ready.')
            sreq.map['authorizations'] = []
        return sreq.map, order_location

    def validate_order(self, finalize_url, order_location, csr_der):
        log.info('Request to finalize the order (all chalenge completed)')
        self.sreq(finalize_url, {'csr': csr_der})
        while True:
            sreq = self.sreq(order_location)
            if sreq.map['status'] == 'processing':
                time.sleep(sreq.headers.get('Retry-After', 5))
            elif sreq.map['status'] == 'valid':
                log.info('Order finalized')
                break
            else:
                raise ValueError(f'Error finalizing order: {sreq.map}')
        return sreq.map

    def validate_challenge(self, *, url=None, keyauth=None):
        log.info('Asking ACME server to validate challenge')
        self.sreq(url, {'keyAuthorization': keyauth})
        while True:
            sreq = self.sreq(url)
            if sreq.map['status'] == 'pending':
                time.sleep(sreq.headers.get('Retry-After', 5))
            elif sreq.map['status'] == 'valid':
                log.info('ACME has verified challenge for domain.')
                break
            else:
                raise ValueError(f'Challenge for domain failed: {sreq.map}')

    def get_auth(self, authurl):
        log.info('Process challenge for authorization: %s', authurl)
        sreq = self.sreq(authurl).map
        domain = sreq['identifier']['value']
        challenge = [c for c in sreq['challenges'] if c['type'] == 'dns-01'][0]
        token = re.sub(r'[^A-Za-z0-9_-]', '_', challenge['token']) # b64url
        keyauth = f'{token}.{self.jwk_thumbprint}'
        keydigest64 = b64sha256(keyauth)
        domain = f'_acme-challenge.{domain}.'
        return domain, keydigest64, challenge['url'], keyauth

    def get_chains(self, crt_url, separator=None):
        headers = {'Accept': 'application/pem-certificate-chain'}
        log.info('Signed certificate at: %s', crt_url)
        sreq = self.sreq(crt_url, headers=headers)

        if separator is None:
            return sreq.text

        chains = [sreq.text]
        regex = re.compile(r'\s*<(\S+)>\s*;\s*rel\s*=\s*"alternate"\s*')
        for match in regex.finditer(sreq.headers.get('Link', '')):
            chains.append(self.sreq(match.group(1), headers=headers).text)

        return separator.join(chains)


def get_crt(args):
    ttl = int(args['--ttl'] or "60")
    domains = extract_domains_from_csr(args['--csr'])
    acme = ACME(account_key_path=args['--account-key'])
    acme.init_sreqs(args['--acme-directory'] or ACME_DEFAULT_DIRECTORY)
    acme.register_account(contacts=args['--contact'])
    order, order_location = acme.new_order(domains=domains)

    log.info('Completing each each authorization challenge')
    for authurl in order['authorizations']:
        domain, keydigest64, challenge_url, keyauth = acme.get_auth(authurl)
        dns_script(args['--script'], 'add', domain, keydigest64)
        self_challenge_check(domain=domain, ttl=ttl, challenge=keydigest64)
        try:
            acme.validate_challenge(url=challenge_url, keyauth=keyauth)
        finally:
            dns_script(args['--script'], 'delete', domain, '')

    csr_der = b64(openssl(['req', '-in', args['--csr'], '-outform', 'DER']))
    order = acme.validate_order(order['finalize'], order_location, csr_der)
    separator = None
    if args['--separator'] is not None:
        separator = args['--separator'].encode('utf8').decode('unicode_escape')
    chains = acme.get_chains(order['certificate'], separator)

    return chains


def main(args):
    args = args or docopt.docopt(__doc__)
    if log_configure:
        log.addHandler(logging.StreamHandler())
        log.setLevel(
                (args['--verbose'] and logging.DEBUG) or
                (args['--quiet']   and logging.ERROR) or
                logging.INFO
        )
    print(get_crt(args), end='')


if __name__ == "__main__":
    main(None)