lets-encrypt-update

#!/usr/bin/env python3
"""Update or acquire Let's Encrypt certificates for OCF's traditional
vhosts.

This script is intended to be run by a cronjob. All it actually does is:

    1. Fetches the list of vhost domains via ocflib.

    2. Checks the DNS to see which ones point to the right server.

    3. Passes the valid domains to ocf-lets-encrypt.

It works equally well for web and app vhosts and takes an argument to
specify which to update.
"""
import argparse
import functools
import subprocess
import sys
from itertools import chain

import dns.resolver
from ocflib.vhost.application import get_app_vhosts
from ocflib.vhost.web import get_vhosts


def debug(*args, **kwargs):
    pass


def all_domains(vhosts):
    """Get all domains being hosted from the list of vhosts.

    This includes:
      * base virtual host domains
      * aliases which redirect to the base domain
    """
    return (
        set(vhosts.keys()) |
        set(chain.from_iterable(
            vhost.get('aliases', [])
            for vhost in vhosts.values()
        ))
    )


def eligible_domains(domains, target_domain):
    """Picks out the domains which we can get certs for.

    We check that the domains appear to be served by us (in particular,
    we check public DNS to see whether they seem to point to the right
    host, i.e. death or the apphost).
    """
    resolver = dns.resolver.Resolver()

    # Use Google's public DNS servers, OpenDNS servers, and Quad9 to test some
    # public DNS services to see what the domains point to. If one fails, then
    # some of the others should hopefully be working fine
    resolver.nameservers = [
        '8.8.8.8', '8.8.4.4', '208.67.222.222', '208.67.220.220', '9.9.9.9',
    ]

    def resolve(fqdn):
        try:
            answers = list(resolver.query(fqdn, 'A'))
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            return
        else:
            if answers:
                return str(answers[0])

    ip = resolve(target_domain)

    eligible_domains = {
        domain
        for domain in domains
        if resolve(domain) == ip
    }

    bad_domains = domains - eligible_domains
    if bad_domains:
        debug('Excluded {} domains for bad DNS:\n    {}'.format(
            len(bad_domains),
            '\n    '.join(bad_domains),
        ))
    else:
        debug('Did not exclude any domains for bad DNS.')

    return eligible_domains


def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument('target', choices=('web', 'app'))
    parser.add_argument('-n', '--dry-run', action='store_true')
    parser.add_argument('-v', '--verbose', action='store_true')
    args = parser.parse_args()

    if args.verbose:
        global debug
        debug = functools.partial(print, flush=True)

    if args.target == 'web':
        vhosts, target_domain = get_vhosts(), 'death.ocf.berkeley.edu'
    else:
        vhosts, target_domain = get_app_vhosts(), 'vampires.ocf.berkeley.edu'

    domains = all_domains(vhosts)
    eligible = eligible_domains(domains, target_domain)
    acquired = set()
    renewed = set()
    for domain in eligible:
        ret = subprocess.call(chain(
            ('ocf-lets-encrypt', domain),
            ('--verbose',) if args.verbose else (),
            ('--dry-run',) if args.dry_run else (),
            ('--extended-return-codes',),
            ('--private-key', '/etc/ssl/lets-encrypt/le-vhost.key'),
            ('--cert', '/services/http/ssl/{}.crt'.format(domain)),
        ))
        if ret == 255:
            acquired.add(domain)
        elif ret == 254:
            renewed.add(domain)
        else:
            assert ret == 0, 'ocf-lets-encrypt returned {}'.format(ret)

    debug('Total domains: {}'.format(len(domains)))
    debug('Total eligible domains: {}'.format(len(eligible)))
    debug('Certs acquired: {}'.format(len(acquired)))
    debug('Certs renewed: {}'.format(len(renewed)))


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