add acme_updater.py

README.md

@@ -8,6 +8,7 @@ This repository contains a collection of scripts written by me which are not big
 * backup_report.sh: Bash script that generates a report about created backups using [storeBackup](http://storebackup.org/). Can be used to track the creation of backups and detect failed backups so they can be cleaned up manually
 * cyber_generator: wsgi script that creates a random word with the prefix "cyber", currently used at [cyber-everything.de](https://cyber-everything.de/)
 * dnsping: Collection of scripts to implement a basic dynamic dns service
+* acme_updater.py: Python scripts to automatically replace letsencrypt certificates managed and renewed by acmetool
 
 ## Copying
 

acme_updater.py

@@ -0,0 +1,195 @@
+#! /usr/bin/env python3
+"""
+This script scans all apache vhosts on the system, checks if they use
+letsencrypt certificates for which acmetool has renewed the certificate and
+replaces the old certificate with the new one automatically. This way,
+acmetool can run as an unprivileged user and acme-updater can take
+care of swapping out certificates while running as root.
+
+Requirements:
+
+* python 3.x
+* module OpenSSL
+"""
+
+import os
+import logging
+import datetime
+import shutil
+import subprocess
+
+import OpenSSL
+
+# directory for enabled apache vhosts, might be different on your distribution
+VHOSTS_DIRECTORY = "/etc/apache2/sites-enabled"
+
+# Log level, default is info
+LOG_LEVEL = logging.INFO
+
+# State directory for acmetool, default is /var/lib/acme
+ACME_STATE_DIR = "/var/lib/acme"
+
+LOGGER = logging.getLogger("acme-updater")
+LOGGER.setLevel(LOG_LEVEL)
+CONSOLE_HANDLER = logging.StreamHandler()
+LOGGER.addHandler(CONSOLE_HANDLER)
+
+PARSED_VHOSTS = []
+CERT_RENEWED = False
+
+
+def parse_vhost(file_obj):
+    """
+    Parses a given vhost file and extracts the main domain,
+    the certificate file and the TLS key file.
+
+    :param file_obj: file obj pointing to a vhost to parse
+    :return: list of tuples with domains and found certificates
+    :rtype: list
+    """
+    vhost_started = False
+    parsed_info = []
+    cert_path = ""
+    key_path = ""
+    main_domain = ""
+    for line in file_obj:
+        if "<VirtualHost" in line:
+            vhost_started = True
+        elif "</VirtualHost" in line and vhost_started:
+            vhost_started = False
+            if cert_path and key_path and main_domain:
+                parsed_info.append((main_domain, cert_path, key_path))
+                LOGGER.debug(
+                    "Found vhost with main domain %s, certificate %s and key "
+                    "file %s", main_domain, cert_path, key_path)
+                cert_path = ""
+                key_path = ""
+                main_domain = ""
+        elif "ServerName" in line:
+            main_domain = line.strip().rsplit()[1]
+        elif "SSLCertificateFile" in line:
+            cert_path = line.strip().rsplit()[1]
+        elif "SSLCertificateKeyFile" in line:
+            key_path = line.strip().rsplit()[1]
+    return parsed_info
+
+
+def parse_asn1_time(timestamp):
+    """
+    parses an ANS1 timestamp as returned by OpenSSL and turns it into a python
+    datetime object.
+
+    :param timestamp: ASN1 timestamp
+    :type timestamp: str
+    :return: timestamp as datetime object
+    :rtype: datetime
+    """
+    year = int(timestamp[:4])
+    month = int(timestamp[4:6])
+    day = int(timestamp[6:8])
+    date = datetime.datetime(year, month, day)
+    return date
+
+
+def copy_file(source, destination):
+    """
+    Copies a file from the given source file to
+    the given destionation and creates a copy of the
+    source file.
+
+    :param source: source file path
+    :type source: str
+    :param destination: destionation file path
+    :type destination: str
+    :return: success
+    :rtype: bool
+    """
+    backup_file = source + ".bak_%s" % datetime.datetime.now().strftime(
+        "%Y%m%d%H%M%S")
+    try:
+        shutil.copy(source, backup_file)
+    except IOError:
+        LOGGER.error("Creating of backup file for %s failed!", source)
+        return False
+    else:
+        try:
+            shutil.copy(source, destination)
+        except IOError:
+            LOGGER.error("Copying of file %s to %s failed!",
+                         source, destination)
+        else:
+            os.chmod(destination, 0o0644)
+            os.chown(destination, 0, 0)
+            return True
+
+for vhost in os.listdir(VHOSTS_DIRECTORY):
+    if vhost.endswith(".conf"):
+        vhost_absolute = os.path.join(VHOSTS_DIRECTORY, vhost)
+        with open(vhost_absolute, "r") as vhost_file:
+            PARSED_VHOSTS.extend(parse_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__():
+            acme_cert_path = os.path.join(ACME_STATE_DIR, "live", entry[0],
+                                          "cert")
+            try:
+                with open(acme_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 = parse_asn1_time(expiry_date)
+                if expiry_datetime < datetime.datetime.utcnow():
+                    LOGGER.warn("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])
+                    elif serial_acme_cert > serial_current_cert:
+                        if copy_file(acme_cert_path, entry[1]):
+                            acme_key_path = os.path.join(ACME_STATE_DIR,
+                                                         "live", entry[0],
+                                                         "privkey")
+                            if 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!")
+                        else:
+                            LOGGER.error("Renewal of cert for %s failed, "
+                                         "please clean up manually and "
+                                         "check the backup files!")
+
+if CERT_RENEWED:
+    LOGGER.debug("Checking apache configuration")
+    try:
+        subprocess.check_call(["apache2ctl", "-t"])
+    except subprocess.CalledProcessError:
+        LOGGER.error("Error in apache configuration, will not restart the "
+                     "web server")
+    else:
+        try:
+            subprocess.check_call(["service", "apache2", "restart"])
+        except subprocess.CalledProcessError:
+            LOGGER.error("Apache restart failed!")
+        else:
+            LOGGER.info("Apache restarted successfully")