acme-updater/amulib/apache.py

# SPDX-FileCopyrightText: 2016-2023 Helmut Pozimski <helmut@pozimski.eu>
#
# SPDX-License-Identifier: GPL-2.0-only

# -*- coding: utf8 -*-

"""
Contains the apache module which takes care of maintaining the apache 2 SSL
certificates.
"""

import datetime
import logging
import os
import subprocess

import OpenSSL

from amulib import helpers
from amulib.cert_path_provider import CertPathProvider

LOGGER = logging.getLogger("acme-updater")


def run(cert_path_provider: CertPathProvider, config=None,
        named_key_path="/run/named/session.key", dns_server="localhost"):
    """
    Main method of the apache module, actually replaces the certificates,
    manages the service and writes TLSA records if necessary.

    :param cert_path_provider: provider of the certificate path
    :type cert_path_provider: CertPathProvider
    :param config: configuration for the module.
    :type config: dict
    :param named_key_path: path to the named session.key
    :type named_key_path: str
    :param dns_server: DNS server to use to create TLSA records
    :type dns_server: str
    """
    cert_renewed = False
    parsed_vhosts = []
    if config:
        vhosts_dir = config["vhosts_dir"]
        tlsa = config["tlsa"]
        exclude_vhosts = config["exclude_vhosts"]
        tlsa_exclude = config["tlsa_exclude"]
    else:
        # Default parameters based on best guesses
        vhosts_dir = "/etc/apache2/sites-enabled"
        tlsa = False
        exclude_vhosts = []
        tlsa_exclude = []

    for vhost in os.listdir(vhosts_dir):
        if vhost.endswith(".conf") and vhost not in exclude_vhosts:
            vhost_absolute = os.path.join(vhosts_dir, vhost)
            with open(vhost_absolute, "r") as vhost_file:
                parsed_vhosts.extend(helpers.parse_apache_vhost(vhost_file))

    for entry in parsed_vhosts:
        try:
            with open(entry[1], "r") as cert_file:
                cert_text = cert_file.read()
        except IOError:
            LOGGER.error("Error while opening cert file %s ", entry[1])
        else:
            x509_current_cert = OpenSSL.crypto.load_certificate(
                OpenSSL.crypto.FILETYPE_PEM, cert_text)
            if "Let's Encrypt" in x509_current_cert.get_issuer().__str__():
                cert_path = cert_path_provider.provide_cert_path(entry[0])
                fullchain_path = cert_path_provider.provide_fullchain_path(entry[0])
                try:
                    with open(cert_path, "r") as acme_cert_file:
                        acme_cert_text = acme_cert_file.read()
                except IOError:
                    LOGGER.error("Could not open certificate for %s in acme "
                                 "state directory", entry[0])
                else:
                    x509_acme_cert = OpenSSL.crypto.load_certificate(
                        OpenSSL.crypto.FILETYPE_PEM, acme_cert_text
                    )
                    expiry_date = \
                        x509_acme_cert.get_notAfter().decode("utf-8")
                    expiry_datetime = helpers.parse_asn1_time(expiry_date)
                    if expiry_datetime < datetime.datetime.utcnow():
                        LOGGER.warning(
                            "Certificate for %s is expired and no newer "
                            "one is available, bailing out!", entry[0])
                    else:
                        serial_current_cert = \
                            x509_current_cert.get_serial_number()
                        serial_acme_cert = x509_acme_cert.get_serial_number()
                        if serial_current_cert == serial_acme_cert:
                            LOGGER.debug("Cert for %s matches with the one "
                                         "installed, nothing to do.", entry[1])
                        else:
                            if tlsa:
                                for domain in entry[3]:
                                    if domain not in tlsa_exclude:
                                        helpers.create_tlsa_records(
                                            domain, "443", x509_acme_cert,
                                            named_key_path, dns_server)
                            if helpers.copy_file(fullchain_path, entry[1]):
                                acme_key_path = cert_path_provider.provide_key_path(entry[0])
                                if helpers.copy_file(acme_key_path, entry[2]):
                                    LOGGER.info(
                                        "Successfully renewed cert for %s",
                                        entry[0])
                                    cert_renewed = True
                                else:
                                    LOGGER.error(
                                        "Renewal of cert for %s failed, "
                                        "please clean up manually and "
                                        "check the backup files!",
                                        entry[0])
                            else:
                                LOGGER.error("Renewal of cert for %s failed, "
                                             "please clean up manually and "
                                             "check the backup files!",
                                             entry[0])
    if cert_renewed:
        LOGGER.debug("Checking apache configuration")
        try:
            subprocess.check_call(["/usr/sbin/apache2ctl", "-t"])
        except subprocess.CalledProcessError:
            LOGGER.error("Error in apache configuration, will not restart the "
                         "web server")
        else:
            try:
                subprocess.check_call(["/etc/init.d/apache2", "restart"])
            except subprocess.CalledProcessError:
                LOGGER.error("Apache restart failed!")
            else:
                LOGGER.info("Apache restarted successfully")