FreedomBox/actions/letsencrypt

#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Configuration helper for Let's Encrypt.
"""

import argparse
import filecmp
import glob
import importlib
import inspect
import json
import os
import pathlib
import re
import shutil
import subprocess
import sys

import configobj

from plinth import action_utils
from plinth import app as app_module
from plinth import cfg
from plinth.modules import letsencrypt as le
from plinth.modules.letsencrypt.components import LetsEncrypt

TEST_MODE = False
LE_DIRECTORY = '/etc/letsencrypt/'
ETC_SSL_DIRECTORY = '/etc/ssl/'
RENEWAL_DIRECTORY = '/etc/letsencrypt/renewal/'
AUTHENTICATOR = 'webroot'
WEB_ROOT_PATH = '/var/www/html'
APACHE_PREFIX = '/etc/apache2/sites-available/'
APACHE_CONFIGURATION = '''
Use FreedomBoxTLSSiteMacro {domain}
'''


def parse_arguments():
    """Return parsed command line arguments as dictionary."""
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')

    setup_parser = subparsers.add_parser(
        'setup', help='Run any setup/upgrade activities.')
    setup_parser.add_argument(
        '--old-version', type=int, required=True,
        help='Version number being upgraded from or None if setting up first '
        'time.')

    subparsers.add_parser('get-status',
                          help='Return the status of configured domains.')
    subparser = subparsers.add_parser(
        'get-modified-time',
        help='Return the modified time for a certificate.')
    subparser.add_argument('--domain', required=True,
                           help='Domain name to get modified time for')
    revoke_parser = subparsers.add_parser(
        'revoke', help='Revoke certificate of a domain and disable website.')
    revoke_parser.add_argument('--domain', required=True,
                               help='Domain name to revoke certificate for')
    obtain_parser = subparsers.add_parser(
        'obtain', help='Obtain certificate for a domain and setup website.')
    obtain_parser.add_argument('--domain', required=True,
                               help='Domain name to obtain certificate for')
    delete_parser = subparsers.add_parser(
        'delete', help='Delete certificate for a domain and disable website.')
    delete_parser.add_argument('--domain', required=True,
                               help='Domain name to delete certificate of')

    subparser = subparsers.add_parser(
        'copy-certificate',
        help='Copy LE certificate to a daemon\'s directory')
    subparser.add_argument('--managing-app', required=True,
                           help='App needing the certificate')
    subparser.add_argument('--user-owner', required=True,
                           help='User who should own the certificate')
    subparser.add_argument('--group-owner', required=True,
                           help='Group that should own the certificate')
    subparser.add_argument('--source-private-key-path', required=True,
                           help='Path to the source private key')
    subparser.add_argument(
        '--source-certificate-path', required=True,
        help='Path to the source certificate with public key')
    subparser.add_argument('--private-key-path', required=True,
                           help='Path to the private key')
    subparser.add_argument('--certificate-path', required=True,
                           help='Path to the certificate with public key')

    subparser = subparsers.add_parser(
        'compare-certificate',
        help='Compare LE certificate to one in daemon\'s directory')
    subparser.add_argument('--managing-app', required=True,
                           help='App needing the certificate')
    subparser.add_argument('--source-private-key-path', required=True,
                           help='Path to the source private key')
    subparser.add_argument(
        '--source-certificate-path', required=True,
        help='Path to the source certificate with public key')
    subparser.add_argument('--private-key-path', required=True,
                           help='Path to the private key')
    subparser.add_argument('--certificate-path', required=True,
                           help='Path to the certificate with public key')

    help_hooks = 'Does nothing, kept for compatibility.'
    subparser = subparsers.add_parser('run_pre_hooks', help=help_hooks)
    subparser.add_argument('--domain')
    subparser.add_argument('--modules', nargs='+', default=[])

    subparser = subparsers.add_parser('run_renew_hooks', help=help_hooks)
    subparser.add_argument('--domain')
    subparser.add_argument('--modules', nargs='+', default=[])

    subparser = subparsers.add_parser('run_post_hooks', help=help_hooks)
    subparser.add_argument('--domain')
    subparser.add_argument('--modules', nargs='+', default=[])

    subparsers.required = True
    return parser.parse_args()


def get_certificate_expiry(domain):
    """Return the expiry date of a certificate."""
    certificate_file = os.path.join(le.LIVE_DIRECTORY, domain, 'cert.pem')
    output = subprocess.check_output(
        ['openssl', 'x509', '-enddate', '-noout', '-in', certificate_file])
    return output.decode().strip().split('=')[1]


def get_modified_time(domain):
    """Return the last modified time of a certificate."""
    certificate_file = pathlib.Path(le.LIVE_DIRECTORY) / domain / 'cert.pem'
    return int(certificate_file.stat().st_mtime)


def get_validity_status(domain):
    """Return validity status of a certificate, e.g. valid, revoked, expired"""
    output = subprocess.check_output(['certbot', 'certificates', '-d', domain])
    output = output.decode(sys.stdout.encoding)

    match = re.search(r'INVALID: (.*)\)', output)
    if match is not None:
        validity = match.group(1).lower()
    elif re.search('VALID', output) is not None:
        validity = 'valid'
    else:
        validity = 'unknown'

    return validity


def get_status():
    """
    Return Python dictionary of currently configured domains.
    Should be run as root, otherwise might yield a wrong, empty answer.
    """
    try:
        domains = os.listdir(le.LIVE_DIRECTORY)
    except OSError:
        domains = []

    domains = [
        domain for domain in domains
        if os.path.isdir(os.path.join(le.LIVE_DIRECTORY, domain))
    ]

    domain_status = {}
    for domain in domains:
        domain_status[domain] = {
            'certificate_available':
                True,
            'expiry_date':
                get_certificate_expiry(domain),
            'web_enabled':
                action_utils.webserver_is_enabled(domain, kind='site'),
            'validity':
                get_validity_status(domain),
            'lineage':
                str(pathlib.Path(le.LIVE_DIRECTORY) / domain),
            'modified_time':
                get_modified_time(domain)
        }
    return domain_status


def subcommand_setup(arguments):
    """Upgrade old site configuration to new macro based style.

    Nothing to do for first time setup and for newer versions.
    """
    if arguments.old_version == 2:
        _remove_old_hooks()
        return

    if arguments.old_version != 1:
        return

    domain_status = get_status()
    with action_utils.WebserverChange() as webserver_change:
        for domain in domain_status:
            setup_webserver_config(domain, webserver_change)


def subcommand_get_status(_):
    """Print a JSON dictionary of currently configured domains."""
    domain_status = get_status()
    print(json.dumps({'domains': domain_status}))


def subcommand_get_modified_time(arguments):
    """Print the modified time of a certificate as integer."""
    print(get_modified_time(arguments.domain))


def subcommand_revoke(arguments):
    """Disable a domain and revoke the certificate."""
    domain = arguments.domain

    cert_path = pathlib.Path(le.LIVE_DIRECTORY) / domain / 'cert.pem'
    if cert_path.exists():
        command = [
            'certbot', 'revoke', '--non-interactive', '--domain', domain,
            '--cert-path', cert_path
        ]
        if TEST_MODE:
            command.append('--staging')

        process = subprocess.Popen(command, stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE)
        _, stderr = process.communicate()
        if process.returncode:
            print(stderr.decode(), file=sys.stderr)
            sys.exit(1)

    action_utils.webserver_disable(domain, kind='site')


def subcommand_obtain(arguments):
    """Obtain a certificate for a domain and setup website."""
    domain = arguments.domain

    command = [
        'certbot', 'certonly', '--non-interactive', '--text', '--agree-tos',
        '--register-unsafely-without-email', '--domain', arguments.domain,
        '--authenticator', AUTHENTICATOR, '--webroot-path', WEB_ROOT_PATH,
        '--renew-by-default'
    ]
    if TEST_MODE:
        command.append('--staging')

    process = subprocess.Popen(command, stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    _, stderr = process.communicate()
    if process.returncode:
        print(stderr.decode(), file=sys.stderr)
        sys.exit(1)

    with action_utils.WebserverChange() as webserver_change:
        setup_webserver_config(domain, webserver_change)


def _remove_old_hooks():
    """Remove old style renewal hooks from individual configuration files.

    This has been replaced with global hooks by adding script files in
    directory /etc/letsencrypt/renewal-hooks/{pre,post,deploy}/.

    """
    for file_path in glob.glob(RENEWAL_DIRECTORY + '*.conf'):
        try:
            _remove_old_hooks_from_file(file_path)
        except Exception as exception:
            print('Error removing hooks from file:', file_path, exception)


def _remove_old_hooks_from_file(file_path):
    """Remove old style hooks from a single configuration file."""
    config = configobj.ConfigObj(file_path)
    edited = False
    for line in config.initial_comment:
        if 'edited by plinth' in line.lower():
            edited = True

    if not edited:
        return

    config.initial_comment = [
        line for line in config.initial_comment
        if 'edited by plinth' not in line.lower()
    ]

    if 'pre_hook' in config['renewalparams']:
        del config['renewalparams']['pre_hook']

    if 'renew_hook' in config['renewalparams']:
        del config['renewalparams']['renew_hook']

    if 'post_hook' in config['renewalparams']:
        del config['renewalparams']['post_hook']

    config.write()


def subcommand_copy_certificate(arguments):
    """Copy certificate from LE directory to daemon's directory.

    Set ownership and permissions as requested needed by the daemon.

    """
    source_private_key_path = pathlib.Path(
        arguments.source_private_key_path).resolve()
    _assert_source_directory(source_private_key_path)
    source_certificate_path = pathlib.Path(
        arguments.source_certificate_path).resolve()
    _assert_source_directory(source_certificate_path)

    private_key_path = pathlib.Path(arguments.private_key_path).resolve()
    _assert_managed_path(arguments.managing_app, private_key_path)
    certificate_path = pathlib.Path(arguments.certificate_path).resolve()
    _assert_managed_path(arguments.managing_app, certificate_path)

    # Create directories, owned by root
    private_key_path.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
    certificate_path.parent.mkdir(mode=0o755, parents=True, exist_ok=True)

    # Private key is only accessible to the user owner
    old_mask = os.umask(0o177)
    shutil.copyfile(source_private_key_path, private_key_path)

    if certificate_path != private_key_path:
        # Certificate is only writable by the user owner
        os.umask(0o133)
        shutil.copyfile(source_certificate_path, certificate_path)
    else:
        # If private key and certificate are the same file, append one after
        # the other.
        source_certificate = source_certificate_path.read_bytes()
        with private_key_path.open(mode='a+b') as file_handle:
            file_handle.write(source_certificate)

    os.umask(old_mask)

    shutil.chown(certificate_path, user=arguments.user_owner,
                 group=arguments.group_owner)
    shutil.chown(private_key_path, user=arguments.user_owner,
                 group=arguments.group_owner)


def subcommand_compare_certificate(arguments):
    """Compare LE certificate with an app certificate."""
    source_private_key_path = pathlib.Path(arguments.source_private_key_path)
    source_certificate_path = pathlib.Path(arguments.source_certificate_path)
    _assert_source_directory(source_private_key_path)
    _assert_source_directory(source_certificate_path)

    private_key_path = pathlib.Path(arguments.private_key_path)
    certificate_path = pathlib.Path(arguments.certificate_path)
    _assert_managed_path(arguments.managing_app, private_key_path)
    _assert_managed_path(arguments.managing_app, certificate_path)

    result = False
    try:
        if filecmp.cmp(source_certificate_path, certificate_path) and \
           filecmp.cmp(source_private_key_path, private_key_path):
            result = True
    except FileNotFoundError:
        result = False

    print(json.dumps({'result': result}))


def _assert_source_directory(path):
    """Assert that a path is a valid source of a certificates."""
    assert (str(path).startswith(LE_DIRECTORY)
            or str(path).startswith(ETC_SSL_DIRECTORY))


def _get_managed_path(path):
    """Return the managed path given a certificate path."""
    if '{domain}' in path:
        return pathlib.Path(path.partition('{domain}')[0])

    return pathlib.Path(path).parent


def _assert_managed_path(module, path):
    """Check that path is in fact managed by module."""
    cfg.read()
    module_file = pathlib.Path(cfg.config_dir) / 'modules-enabled' / module
    module_path = module_file.read_text().strip()

    module = importlib.import_module(module_path)
    module_classes = inspect.getmembers(module, inspect.isclass)
    app_classes = [
        cls[1] for cls in module_classes if issubclass(cls[1], app_module.App)
    ]

    managed_paths = []
    for cls in app_classes:
        app = cls()
        components = app.get_components_of_type(LetsEncrypt)
        for component in components:
            if component.private_key_path:
                managed_paths.append(
                    _get_managed_path(component.private_key_path))
            if component.certificate_path:
                managed_paths.append(
                    _get_managed_path(component.certificate_path))

    assert set(path.parents).intersection(set(managed_paths))


def subcommand_run_pre_hooks(_):
    """Do nothing, kept for legacy LE configuration.

    If new version of Plinth is deployed and before it can update the Let's
    Encrypt configuration and remove these old hooks, if a renew operation is
    run, then we don't want it to exit with non-zero error code because this
    hook could not be run.

    Remove at some point in the future.

    """


def subcommand_run_renew_hooks(_):
    """Do nothing, kept for legacy LE configuration.

    If new version of Plinth is deployed and before it can update the Let's
    Encrypt configuration and remove these old hooks, if a renew operation is
    run, then we don't want it to exit with non-zero error code because this
    hook could not be run.

    Remove at some point in the future.

    """


def subcommand_run_post_hooks(_):
    """Do nothing, kept for legacy LE configuration.

    If new version of Plinth is deployed and before it can update the Let's
    Encrypt configuration and remove these old hooks, if a renew operation is
    run, then we don't want it to exit with non-zero error code because this
    hook could not be run.

    Remove at some point in the future.

    """


def subcommand_delete(arguments):
    """Disable a domain and delete the certificate."""
    domain = arguments.domain
    command = ['certbot', 'delete', '--non-interactive', '--cert-name', domain]
    process = subprocess.Popen(command, stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    _, stderr = process.communicate()
    if process.returncode:
        print(stderr.decode(), file=sys.stderr)
        sys.exit(1)

    action_utils.webserver_disable(domain, kind='site')


def setup_webserver_config(domain, webserver_change):
    """Create SSL web server configuration for a domain.

    Do so only if there is no configuration existing.
    """
    file_name = os.path.join(APACHE_PREFIX, domain + '.conf')
    if os.path.isfile(file_name):
        os.rename(file_name, file_name + '.fbx-bak')

    with open(file_name, 'w') as file_handle:
        file_handle.write(APACHE_CONFIGURATION.format(domain=domain))

    webserver_change.enable('freedombox-tls-site-macro', kind='config')
    webserver_change.enable(domain, kind='site')


def main():
    """Parse arguments and perform all duties."""
    arguments = parse_arguments()

    subcommand = arguments.subcommand.replace('-', '_')
    subcommand_method = globals()['subcommand_' + subcommand]
    subcommand_method(arguments)


if __name__ == '__main__':
    main()