Change contact email addresses in Apache httpd mod_md account files

md-accounts-contact-update.py

#!/usr/bin/env python3

# TODO: File locking using fcntl.lock

import argparse
import base64
import json
import logging
import pathlib
import acme.client
import acme.messages
import josepy as jose
from cryptography.hazmat.primitives import serialization


MD_STORE_DIR_DEFAULT = '.'
MD_STORE_JSON = 'md_store.json'
MD_ACCOUNTS_DIR = 'accounts'
ACCOUNT_JSON = 'account.json'
ACCOUNT_PEM = 'account.pem'

logger = logging.getLogger(__name__)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description='Change contact email addresses in Apache httpd mod_md account files')
    parser.add_argument('-d', '--debug', action='store_true')
    parser.add_argument('-n', '--dry-run', action='store_true')
    parser.add_argument('-m', '--md-store-dir', default=MD_STORE_DIR_DEFAULT, type=pathlib.Path)
    group = parser.add_mutually_exclusive_group()
    _account_argument = group.add_argument('-a', '--account')  # Needed for nice error message
    group.add_argument('-A', '--account-pattern')  # default='*' is implemented below
    parser.add_argument('-e', '--email', required=True)
    args = parser.parse_args()

    if args.debug:
        logging.basicConfig(level=logging.DEBUG)

    md_accounts_dir = args.md_store_dir / MD_ACCOUNTS_DIR

    if args.account:
        account = pathlib.Path(args.account)
        # Try very hard to find exactly one account
        if account.is_dir():
            # Full path to account directory
            # XXX: Should we check whether this realpath is a child of md_accounts_dir?
            accounts = [account]
        elif (md_accounts_dir / account).is_dir():
            # Account directory basename
            accounts = [(md_accounts_dir / account)]
        else:
            # Something else, try to glob match one account dir
            _a = list(md_accounts_dir.glob(f'*{account}*'))
            if len(_a) == 1:
                accounts = _a
            elif not _a:
                # Print nice error message
                parser.error(str(argparse.ArgumentError(
                    _account_argument,
                    f'Could not find an account matching "{account}"')))
            else:
                # Print nice error message
                parser.error(str(argparse.ArgumentError(
                    _account_argument,
                    f'Found more than one account matching "{account}"')))
    else:
        # Here we set account pattern '*' as default
        account_pattern = args.account_pattern or '*'
        accounts = list(md_accounts_dir.glob(account_pattern))

    logger.debug(accounts)

    # First read the account key password from MD_STORE_JSON
    with open(args.md_store_dir / MD_STORE_JSON, encoding='utf-8') as md_store_json:
        md_store = json.load(md_store_json)
        password = base64.urlsafe_b64decode(md_store['key'])

    for account in accounts:
        logger.debug(account)

        with open(account / ACCOUNT_PEM, 'rb') as account_pem, \
                open(account / ACCOUNT_JSON, encoding='utf-8') as account_json:
            # Read and decrypt account key
            md_account_key = serialization.load_pem_private_key(
                account_pem.read(),
                password)
            # Convert to JWK
            jwkrsakey = jose.jwk.JWKRSA(key=md_account_key)

            # Load account info
            md_account = json.load(account_json)
            logger.debug(md_account)

        # Stub the existing registration for the acme library
        regr = acme.messages.RegistrationResource(
            body=acme.messages.Registration(
                key=jwkrsakey,
                contact=md_account['contact'],
                status=md_account['status']),
            uri=md_account['url'])
        logger.debug(regr)

        # Create the update message with the new email address
        upd = acme.messages.UpdateRegistration(
            key=jwkrsakey,
            contact=[f'mailto:{args.email}'],
            status='valid')  # XXX: Required by server API
        logger.debug(upd)

        if not args.dry_run:
            # API client preparations
            net = acme.client.ClientNetwork(key=jwkrsakey)
            try:
                # acme >= 2
                directory = acme.client.ClientV2.get_directory(
                    url=md_account['ca-url'],
                    net=net)
            except AttributeError:
                # acme < 2
                directory = acme.messages.Directory.from_json(net.get(md_account['ca-url']).json())
            client = acme.client.ClientV2(directory=directory, net=net)

            # Actually perform the update API call
            regr = client.update_registration(regr=regr, update=upd)
            logger.debug(regr)

            # Store updated contact address in account info
            md_account['contact'] = regr.body['contact']
            logger.debug(md_account)

            with open(account / ACCOUNT_JSON, 'w', encoding='utf-8') as account_json:
                json.dump(md_account, account_json)