Merge branch '3-create-molecule-scenario' of dev.parley.be:PPBe/ansible-role-acme into 3-create-molecule-scenario

.gitignore

@@ -1 +0,0 @@
-.vscode

defaults/main.yml

@@ -1,36 +1,10 @@
-acme_ssl_dir: /etc/ssl
+acme_directory: https://acme-v02.api.letsencrypt.org/directory
-acme_config_dir: "{{ acme_ssl_dir }}/acme.d"
+acme_config_dir: /etc/ssl
-acme_config_file: "{{ acme_ssl_dir }}/acme.yml"
+acme_keys_dir: "{{ acme_config_dir }}/private"
-acme_keys_dir: "{{ acme_ssl_dir }}/private"
+acme_csr_dir: "{{ acme_config_dir }}/csr"
-acme_csr_dir: "{{ acme_ssl_dir }}/csr"
+acme_certs_dir: "{{ acme_config_dir }}/certs"
-acme_certs_dir: "{{ acme_ssl_dir }}/certs"
+acme_accounts_dir: "{{ acme_config_dir }}/accounts"
-acme_accounts_dir: "{{ acme_ssl_dir }}/accounts"
+acme_account_key: "acme_account.key"
-acme_script_dir: /opt/acme
-acme_script_bin: /usr/local/bin/acme-renew-cert
-
 acme_ssl_group: ssl-cert
-
+acme_challenge_dir: /var/www/acme
-acme_account_private_key: "{{ acme_accounts_dir }}/acme_account.key"
+acme_domains: []
-acme_account_email: acme@example.com
-acme_account_owner: root
-acme_account_group: "{{ acme_account_group }}"
-acme_directory_url: https://acme-staging-v02.api.letsencrypt.org/directory
-acme_root_dir: /var/www/acme
-acme_challenge_dir: "{{ acme_challenge_root_dir }}/.well-known/acme-challenge"
-acme_domains:
-  example.com:
-    alt_names: 
-      - test.example.com
-    owner: root
-    group: "{{ acme_ssl_group }}"
-    remaining_days: 30
-    hooks:
-      - systemctl reload nginx
-acme_config:
-  account:
-    private_key: "{{ acme_account_private_key }}"
-    email: "{{ acme_account_email }}"
-    owner: "{{ acme_account_owner }}"
-    group: "{{ acme_account_group }}"
-  directory_url: "{{ acme_directory_url }}"
-  challenge_dir: "{{ acme_challenge_dir }}"

files/acme_renew_cert.py

@@ -1,435 +1,217 @@
 #!/usr/bin/env python3
 
-import glob
-import grp
 import logging
 import os
-import pwd
-import subprocess
 import sys
 from datetime import datetime
-from typing import List, Dict
 
-import acme
 import josepy as jose
-import requests
+import acme
-import yaml
-from acme import client, messages, challenges, crypto_util
 from OpenSSL import crypto
+from acme import client, messages, challenges, crypto_util
 
-SSL_CONFIG_DIR = "/etc/ssl"
+SSL_CONFIG_DIR="/etc/ssl"
-SSL_CONFIG_FILE = os.path.join(SSL_CONFIG_DIR, "acme.yml")
-SSL_ACCOUNTS_DIR = os.path.join(SSL_CONFIG_DIR, "accounts")
-SSL_KEYS_DIR = os.path.join(SSL_CONFIG_DIR, "private")
-SSL_CSR_DIR = os.path.join(SSL_CONFIG_DIR, "csr")
-SSL_CERTS_DIR = os.path.join(SSL_CONFIG_DIR, "certs")
-
-try:
-    SSL_GROUP = grp.getgrnam("ssl-cert").gr_name
-except KeyError:
-    SSL_GROUP = "root"
-
-ACME_CONFIG_FILE = f"{SSL_CONFIG_DIR}/acme.yml"
-ACME_CONFIG_DIR = f"{SSL_CONFIG_DIR}/acme.d"
-ACME_ACCOUNT_KEY = os.path.join(SSL_ACCOUNTS_DIR, "acme_account.key")
-ACME_CHALLENGE_DIR = "/var/www/acme/.well-known/acme-challenge"
-LETSENCRYPT_DIRECTORY_URL = "https://acme-v02.api.letsencrypt.org/directory"
 OPENSSL_DATE_FORMAT = '%Y%m%d%H%M%SZ'
-CERT_REMAINING_DAYS = 30
+USER_AGENT="python-acme"
-USER_AGENT = "python-acme"
+
-
+def get_cert_expiration_days(cert_path: str):
-OS_OWNERS = {}
+    """Calculate remaining number of days before certificate's expiration.
-OS_GROUPS = {}
+
-
+    :param str cert_path: Path to a certificate file.
-
+
-def chown(path: str, owner: str, group: str, **kwargs) -> None:
+    :returns: Number of remaining days before certificate's expiration.
-    if owner not in OS_OWNERS:
+    :rtype: int
-        OS_OWNERS[owner] = pwd.getpwnam(owner).pw_uid
+    """
-    if group not in OS_GROUPS:
+    now = datetime.now()
-        OS_GROUPS[group] = grp.getgrnam(group).gr_gid
+    cert_expiration_date = now
-    os.chown(path, OS_OWNERS[owner], OS_GROUPS[group], **kwargs)
+
-
+    try:
-
+        with open(cert_path, 'rb') as pem_in:
-class PrivateKey:
+            fullchain_pem = crypto.load_certificate(crypto.FILETYPE_PEM, pem_in.read())
-    def __init__(self, private_key: str, owner: str, group: str):
+        cert_expiration_date = datetime.strptime(fullchain_pem.get_notAfter().decode(), OPENSSL_DATE_FORMAT)
-        self.private_key = private_key
+    except IOError as e:
-        self.owner = owner
+        logging.warning(f"Unable to load certificate: {e}")
-        self.group = group
+
-
+    cert_expiration_days = (cert_expiration_date - now).days
-    def load_or_create_private_key(self) -> crypto.PKey:
+    return cert_expiration_days
-        logging.info(f"Loading private key from {self.private_key}")
+
-        try:
+def load_or_create_csr(domain: str, csr_path: str, privkey_path: str):
-            with open(self.private_key, 'rb') as pem_in:
+    """Load or create a CSR for a given domain.
-                private_key_pem = crypto.load_privatekey(crypto.FILETYPE_PEM,
+
-                                                         pem_in.read())
+    :param str domain: A domain name.
-            return private_key_pem
+    :param str csr_path: Path to a CSR file.
-        except IOError as e:
+    :param str privkey_path: Path to a private key. It is used as fallback when the CSR doesn't exist.
-            logging.warning(f"Unable to load private key: {e}")
+
-
+    :returns: The CSR in PEM file format.
-            logging.info("Generating a new private key")
+    :rtype: str
-            private_key_pem = crypto.PKey()
+    """
-            private_key_pem.generate_key(crypto.TYPE_RSA, 4096)
+    logging.info(f"Loading {domain} CSR file from {csr_path}…")
-
+    try:
-            logging.info(f"Writing private key into {self.private_key}")
+      with open(csr_path, 'r') as pem_in:
-            private_key_dir = os.path.dirname(self.private_key)
+        csr_pem = pem_in.read()
-            os.makedirs(private_key_dir, mode=0o750, exist_ok=True)
+    except IOError as e:
-            chown(private_key_dir, self.owner, self.group)
+        logging.warning(f"Unable to load CSR file: {e}")
-            ofd = os.open(self.private_key, os.O_CREAT | os.O_WRONLY, 0o640)
+        logging.info(f"Loading {domain} private key from {privkey_path}…")
-            with open(ofd, 'wb') as pem_out:
+        with open(privkey_path, 'r') as pem_in:
-                pem_out.write(crypto.dump_privatekey(crypto.FILETYPE_PEM,
+            privkey_pem = pem_in.read()
-                                                     private_key_pem))
+        
-            chown(self.private_key, self.owner, self.group)
+        logging.info(f"Generating a new CSR…")
-            return private_key_pem
+        csr_pem = crypto_util.make_csr(privkey_pem, [domain])
-
+    return csr_pem
-
+
-class Account(PrivateKey):
+def select_http01_challenge(order: messages.OrderResource):
-    def __init__(self, private_key: str, email: str, owner: str, group: str):
+    """Select the HTTP-01 challenge from a given order
-        super().__init__(private_key, owner, group)
+
-        self.private_key_pem = self.load_or_create_private_key()
+    :param messages.OrderResource order: ACME order containing the challenges.
-        self.email = email
+
-
+    :returns: The HTTP-01 challenge.
-        self.jwk = jose.JWKRSA(key=self.private_key_pem.to_cryptography_key())
+    :rtype: challenges.Challenge
-        self.jwk_pub = self.jwk.public_key()
+    """
-
+    for auth in order.authorizations:
-    def register(self, directory_url: str) -> acme.client.ClientV2:
+        for challenge in auth.body.challenges:
-        client_network = client.ClientNetwork(self.jwk, user_agent=USER_AGENT)
+            if isinstance(challenge.chall, challenges.HTTP01):
-
+                return challenge
-        directory_json = client_network.get(directory_url).json()
+    raise Exception("HTTP-01 challenge was not offered by the CA server.")
-        directory = messages.Directory.from_json(directory_json)
+
-        acme_client = client.ClientV2(directory, net=client_network)
+def perform_http01_challenge(
-
+    client_acme: client.ClientV2,
-        try:
+    challenge: challenges.Challenge,
-            logging.info(f"Registering with a new ACME account")
+    order: messages.OrderResource,
-            registration = messages.NewRegistration.from_data(
+    challenge_dir: str):
-                email=self.email,
+    """Perform the HTTP-01 challenge in a given directory
-                terms_of_service_agreed=True)
+
-            acme_client.new_account(registration)
+    :param client.ClientV2 client_acme: A ACME v2 client
-        except acme.errors.ConflictError as e:
+    :param challenges.Challenge challenge: A ACME challenge
-            logging.warning(f"Unable to create a new ACME account: {e}")
+    :param messages.OrderResource order: An ACME order
-            logging.info("Registering with an existing ACME account")
+    :param str challenge_dir: The directory containing the challenge
-            registration_message = messages.NewRegistration(
+
-                key=self.jwk_pub,
+    :returns: The fullchain certificate in PEM file format
-                only_return_existing=True)
+    :rtype: str
-            registration_response = acme_client._post(directory["newAccount"],
+    """
-                                                      registration_message)
+    response, validation = challenge.response_and_validation(client_acme.net.key)
-            account = acme_client._regr_from_response(registration_response)
+    validation_filename, _ = validation.split('.')
-            acme_client.net.account = account
+
-        return acme_client
+    challenge_path = os.path.join(challenge_dir, validation_filename)
-
+    with open(challenge_path, 'w') as f_out:
-
+        f_out.write(validation)
-class Challenge:
+
-    def __init__(self, path: str, acme_client: acme.client.ClientV2):
+    client_acme.answer_challenge(challenge, response)
-        self.path = path
+
-        self.acme_client = acme_client
+    fullchain_pem = client_acme.poll_and_finalize(order).fullchain_pem
-
+
-    def run(self, csr_pem: str) -> str:
+    os.remove(challenge_path)
-        logging.info("Ordering ACME challenge")
+
-        order = self.acme_client.new_order(csr_pem)
+    return fullchain_pem
-
+
-        logging.info("Selecting HTTP-01 ACME challenge")
+def renew_cert(domain: str,
-        challenge = self.select_http01_challenge(order)
+        account_key_path: str,
-
+        privkey_path: str,
-        logging.info("Performing HTTP-01 ACME challenge")
+        csr_path: str,
-        fullchain_pem = self.perform_http01_challenge(challenge, order)
+        certs_dir: str,
-        return fullchain_pem
+        challenge_dir: str,
-
+        directory_url: str,
-    @staticmethod
+        days_before_renewal: int,
-    def select_http01_challenge(order: messages.OrderResource) \
+        force: bool = False):
-            -> challenges.Challenge:
+    logging.info(f"Checking {domain} certificate's expiration date…")
-        """Select the HTTP-01 challenge from a given order
+    fullchain_path = os.path.join(certs_dir, "fullchain.pem")
-
+
-        :param messages.OrderResource order: Order containing the challenges.
+    cert_expiration_days = get_cert_expiration_days(fullchain_path)
-
+    if not force and (cert_expiration_days >= days_before_renewal):
-        :returns: The HTTP-01 challenge.
+        logging.info(f"Certificate expires in {cert_expiration_days} days. Nothing to do.")
-        :rtype: challenges.Challenge
+        return
-        """
+
-        for auth in order.authorizations:
+    logging.info(f"Certificate expires in {cert_expiration_days} days! Renewing certificate…")
-            for challenge in auth.body.challenges:
+
-                if isinstance(challenge.chall, challenges.HTTP01):
+    logging.info(f"Loading account key from {account_key_path}…")
-                    return challenge
+    with open(account_key_path, 'rb') as pem_in:
-        raise Exception("HTTP-01 challenge was not offered by the CA server.")
+        account_pem = crypto.load_privatekey(crypto.FILETYPE_PEM, pem_in.read())
-
+    account_jwk = jose.JWKRSA(key=account_pem.to_cryptography_key())
-    def perform_http01_challenge(self, challenge: challenges.Challenge,
+    account_jwk_pub = account_jwk.public_key()
-                                 order: messages.OrderResource) -> str:
+
-        """Perform the HTTP-01 challenge in a given directory
+    csr_pem = load_or_create_csr(domain, csr_path, privkey_path)
-
+
-        :param challenges.Challenge challenge: A ACME challenge
+    client_network = client.ClientNetwork(account_jwk, user_agent=USER_AGENT)
-        :param messages.OrderResource order: An ACME order
+    directory = messages.Directory.from_json(client_network.get(directory_url).json())
-
+    client_acme = client.ClientV2(directory, net=client_network)
-        :returns: The fullchain certificate in PEM file format
+
-        :rtype: str
+    logging.info("Registering with ACME account…")
-        """
+    # Here we assume we already have an account
-        account_key = self.acme_client.net.key
+    registration_message = messages.NewRegistration(
-        response, validation = challenge.response_and_validation(account_key)
+        key=account_jwk_pub,
-        validation_filename, _ = validation.split('.')
+        only_return_existing=True)
-
+    registration_response = client_acme._post(directory["newAccount"], registration_message)
-        challenge_path = os.path.join(self.path, validation_filename)
+    account = client_acme._regr_from_response(registration_response)
-        logging.info(f"Writing challenge into {challenge_path} directory")
+    client_acme.net.account = account
-        os.makedirs(self.path, mode=0o755, exist_ok=True)
+
-        with open(challenge_path, 'w') as ofd:
+    logging.info("Ordering ACME challenge…")
-            ofd.write(validation)
+    order = client_acme.new_order(csr_pem)
-
+
-        self.acme_client.answer_challenge(challenge, response)
+    logging.info("Selecting HTTP-01 ACME challenge…")
-
+    challenge = select_http01_challenge(order)
-        fullchain_pem = self.acme_client.poll_and_finalize(order).fullchain_pem
+
-        os.remove(challenge_path)
+    logging.info("Performing HTTP-01 ACME challenge…")
-        return fullchain_pem
+    fullchain_pem = perform_http01_challenge(client_acme, challenge, order, challenge_dir)
-
+
-
+    logging.info(f"Writing {domain} certificates into {certs_dir}…")
-class Domain(PrivateKey):
+    certs_pem = []
-    def __init__(self, name: str, owner: str, group: str,
+    for line in fullchain_pem.split('\n'):
-                 alt_names: List[str] = None,
+        if 'BEGIN CERTIFICATE' in line:
-                 cert: str = None, chain_cert: str = None,
+            cert_pem = ''
-                 fullchain_cert: str = None, private_key: str = None,
+        
-                 csr: str = None, remaining_days: int = CERT_REMAINING_DAYS,
+        cert_pem += line + '\n'
-                 hooks: List[str] = None):
+
-        if private_key is None:
+        if 'END CERTIFICATE' in line:
-            private_key = os.path.join(SSL_KEYS_DIR, name + '.key')
+            certs_pem.append(cert_pem)
-        super().__init__(private_key, owner, group)
+    
-
+    cert_path = os.path.join(certs_dir, "cert.pem")
-        self.name = name
+    with open(cert_path, 'w') as pem_out:
-        self.alt_names = [] if alt_names is None else alt_names
+        pem_out.write(certs_pem[0])
-        self.remaining_days = remaining_days
+    
-        self.hooks = [] if hooks is None else hooks
+    chain_path = os.path.join(certs_dir, "chain.pem")
-
+    with open(chain_path, 'w') as pem_out:
-        certs_dir = os.path.join(SSL_CERTS_DIR, name + '.d')
+        pem_out.write(''.join(certs_pem[1:]))
-        if cert is None:
+
-            self.cert = os.path.join(certs_dir, 'cert.pem')
+    with open(fullchain_path, 'w') as pem_out:
-        else:
+        pem_out.write(fullchain_pem)
-            self.cert = cert
+
-        if chain_cert is None:
+if __name__ == '__main__':
-            self.chain_cert = os.path.join(certs_dir, 'chain.pem')
+    import argparse
-        else:
+
-            self.chain_cert = chain_cert
-        if fullchain_cert is None:
-            self.fullchain_cert = os.path.join(certs_dir, 'fullchain.pem')
-        else:
-            self.fullchain_cert = fullchain_cert
-        if csr is None:
-            self.csr = os.path.join(SSL_CSR_DIR, name + '.csr')
-        else:
-            self.csr = csr
-
-        self.cert_expiration_days = None
-
-    def check_certificate_expiration_date(self) -> bool:
-        """Indicate whether the certificate will expire soon or not.
-
-        :returns: True if the certificate expires soon.
-        :rtype: bool
-        """
-        if self.cert_expiration_days is None:
-            logging.debug(f"Checking '{self.cert}' certificate expiration "
-                          "date")
-            try:
-                with open(self.cert, 'rb') as pem_in:
-                    cert_pem = crypto.load_certificate(crypto.FILETYPE_PEM,
-                                                       pem_in.read())
-            except IOError as e:
-                logging.warning(f"Unable to load certificate: {e}")
-                self.cert_expiration_days = float("-inf")
-                return True
-
-            now = datetime.now()
-            cert_raw_expiration_date = cert_pem.get_notAfter().decode()
-            cert_expiration_date = datetime.strptime(cert_raw_expiration_date,
-                                                     OPENSSL_DATE_FORMAT)
-            self.cert_expiration_days = (cert_expiration_date - now).days
-            logging.info(f"Certificate expires in {self.cert_expiration_days} "
-                         "days.")
-        return self.cert_expiration_days <= self.remaining_days
-
-    def renew_cert(self, challenge: Challenge, force: bool = False) -> None:
-        if not (force or self.check_certificate_expiration_date()):
-            return
-        logging.info(f"Renewing {self.name} certificates")
-
-        csr_pem = self.load_or_create_csr()
-        fullchain_pem = challenge.run(csr_pem)
-
-        logging.info(f"Saving {self.name} certificates")
-        certs_pem = []
-        for line in fullchain_pem.split('\n'):
-            if 'BEGIN CERTIFICATE' in line:
-                cert_pem = ''
-
-            cert_pem += line + '\n'
-
-            if 'END CERTIFICATE' in line:
-                certs_pem.append(cert_pem)
-
-        logging.info(f"Writing '{self.cert}' certificate")
-        cert_dir = os.path.dirname(self.cert)
-        os.makedirs(cert_dir, mode=0o755, exist_ok=True)
-        with open(self.cert, 'w') as pem_out:
-            pem_out.write(certs_pem[0])
-
-        logging.info(f"Writing '{self.chain_cert}' chain certificate")
-        chain_cert_dir = os.path.dirname(self.chain_cert)
-        os.makedirs(chain_cert_dir, mode=0o755, exist_ok=True)
-        with open(self.chain_cert, 'w') as pem_out:
-            pem_out.write(''.join(certs_pem[1:]))
-
-        logging.info(f"Writing '{self.fullchain_cert}' full chain certificate")
-        fullchain_cert_dir = os.path.dirname(self.fullchain_cert)
-        os.makedirs(fullchain_cert_dir, mode=0o755, exist_ok=True)
-        with open(self.fullchain_cert, 'w') as pem_out:
-            pem_out.write(fullchain_pem)
-
-        self.run_hooks()
-
-    def load_or_create_csr(self) -> str:
-        """Load or create a CSR for the domain.
-
-        :returns: The CSR in PEM file format.
-        :rtype: str
-        """
-        logging.info(f"Loading {self.name} CSR file from '{self.csr}'")
-        try:
-            with open(self.csr, 'rb') as pem_in:
-                csr = pem_in.read()
-        except IOError as e:
-            logging.warning(f"Unable to load CSR file: {e}")
-            csr = self.create_csr()
-        csr_pem = crypto.load_certificate_request(crypto.FILETYPE_PEM, csr)
-
-        logging.info("Checking if CSR contains all the domains")
-        domain_names = set()
-        for ext in csr_pem.get_extensions():
-            if ext.get_short_name().decode() != "subjectAltName":
-                continue
-            subject_domains = str(ext).split(',')
-            for subject_domain in subject_domains:
-                _, domain = str(subject_domain).split(':')
-                domain_names.add(domain)
-        domain_names_diff = domain_names ^ set([self.name] + self.alt_names)
-        if len(domain_names_diff) > 0:
-            logging.warning(f"Differences found in CSR: {domain_names_diff}")
-            csr = self.create_csr()
-
-        return csr
-
-    def create_csr(self):
-        private_key_pem = self.load_or_create_private_key()
-        private_key = crypto.dump_privatekey(crypto.FILETYPE_PEM,
-                                             private_key_pem)
-
-        logging.info(f"Generating a new CSR")
-        csr = crypto_util.make_csr(private_key, [self.name] + self.alt_names)
-
-        logging.info(f"Writing CSR file into '{self.csr}'")
-        csr_dir = os.path.dirname(self.csr)
-        os.makedirs(csr_dir, mode=0o755, exist_ok=True)
-        with open(self.csr, 'wb') as pem_out:
-            pem_out.write(csr)
-        return csr
-
-    def run_hooks(self) -> None:
-        for hook in self.hooks:
-            logging.info(f"Running hook: '{hook}'")
-            hook_res = subprocess.run(hook.split(),
-                                      stdout=subprocess.DEVNULL,
-                                      stderr=subprocess.PIPE)
-            if hook_res.returncode != 0:
-                logging.error(f"Unable to run hook: {hook_res.stderr}")
-
-
-class Config:
-    def __init__(self, filename, directory):
-        self.filename = filename
-        self.directory = directory
-        self.config = {}
-        self.domain_configs = {}
-
-    def load(self) -> None:
-        with open(self.filename, 'r') as ifd:
-            self._data = yaml.safe_load(ifd)
-
-        for pattern in ('*.yml', '*.yaml'):
-            config_glob = os.path.join(self.directory, pattern)
-            for config_file in glob.glob(config_glob):
-                with open(config_file, 'r') as ifd:
-                    self.domain_configs[config_file] = yaml.safe_load(ifd)
-
-    def parse_domains(self) -> List[Domain]:
-        domains = self._data.get("domains", {})
-
-        domain_uniques = set()
-        domain_duplicates = set()
-        if isinstance(domains, list):
-            for domain in domains:
-                domain_name = domain["name"]
-                if domain_name not in domain_uniques:
-                    domain_uniques.add(domain_name)
-                else:
-                    domain_duplicates.add(domain_name)
-            domains = {domain["name"]: domain for domain in domains}
-
-        for config_file, domain_config in self.domain_configs.items():
-            domain_filename, _ = os.path.splitext(config_file)
-            domain_name = domain_config.get("name", domain_filename)
-            if domain_name not in domain_uniques:
-                domain_uniques.add(domain_name)
-            else:
-                domain_duplicates.add(domain_name)
-            domains[domain_name] = domain_config
-
-        if len(domain_duplicates) > 0:
-            raise RuntimeError(f"Duplicate domains found: {domain_duplicates}")
-
-        domains_parsed = []
-        for domain_name, domain_details in domains.items():
-            remaining_days = domain_details.get("remaining_days",
-                                                CERT_REMAINING_DAYS)
-            domain = Domain(
-                name=domain_name,
-                owner=domain_details.get("owner", "root"),
-                group=domain_details.get("group", SSL_GROUP),
-                alt_names=domain_details.get("alt_names"),
-                cert=domain_details.get("cert"),
-                chain_cert=domain_details.get("chain_cert"),
-                fullchain_cert=domain_details.get("fullchain_cert"),
-                private_key=domain_details.get("private_key"),
-                csr=domain_details.get("csr"),
-                remaining_days=remaining_days,
-                hooks=domain_details.get("hooks"))
-            domains_parsed.append(domain)
-        return domains_parsed
-
-    def parse_account(self) -> Account:
-        account_config = self._data["account"]
-        return Account(
-            private_key=account_config.get("private_key", ACME_ACCOUNT_KEY),
-            email=account_config["email"],
-            owner=account_config.get("owner", "root"),
-            group=account_config.get("group", "root"))
-
-    @property
-    def directory_url(self) -> str:
-        return self._data.get("directory_url", LETSENCRYPT_DIRECTORY_URL)
-
-    @property
-    def challenge_dir(self) -> str:
-        return self._data.get("challenge_dir", ACME_CHALLENGE_DIR)
-
-
-def main():
     parser = argparse.ArgumentParser(
-        description="Renew ACME certificates for a list of domains. This "
+        description="Renew ACME certificates for a list of domains. This script assumes you already have an account on the CA server and a private key for each certificate.",
-                    "script assumes you already have an account on the CA "
-                    "server and a private key for each certificate.",
         formatter_class=argparse.ArgumentDefaultsHelpFormatter)
-    parser.add_argument("--config", "-c", default=ACME_CONFIG_FILE,
+    parser.add_argument("domains", nargs='+',
-                        help="Path to the config file.")
+        help="List of domain names for which to renew the certificate.")
-    parser.add_argument("--config-dir", "-d", default=ACME_CONFIG_DIR,
+    parser.add_argument("--account_key", "-a",
-                        help="Path to the config directory.")
+        default=os.path.join(SSL_CONFIG_DIR, "accounts", "acme_account.key"),
-    parser.add_argument("--quiet", "-q", action="store_true",
+        help="Path to the account key.")
-                        help="Quiet mode.")
+    parser.add_argument("--privkey", "-p",
-    parser.add_argument("--verbose", "-v", action="store_true",
+        default=os.path.join(SSL_CONFIG_DIR, "private", "{domain}.pem"),
-                        help="Increase verbosity.")
+        help="Path to the private certificate.")
-    parser.add_argument("--force", "-f", action="store_true",
+    parser.add_argument("--csr", "-r",
-                        help="Force certificates renewal without checking "
+        default=os.path.join(SSL_CONFIG_DIR, "csr", "{domain}.pem"),
-                             "their expiration date.")
+        help="Path to the CSR file. If the file doesn't exist, it will be generated from the private key and the domain name.")
+    parser.add_argument("--certs", "-o",
+        default=os.path.join(SSL_CONFIG_DIR, "certs", "{domain}.d"),
+        help="Path to the certificates directory.")
+    parser.add_argument("--challenge", "-c",
+        default="/var/www/html/.well-known/acme-challenge",
+        help="Path to the challenge directory.")
+    parser.add_argument("--directory_url", "-u",
+        default="https://acme-v02.api.letsencrypt.org/directory",
+        help="Directory URL on which performing ACME challenges. Only ACME v2 is supported.")
+    parser.add_argument("--days", "-d", type=int,
+        default=30,
+        help="Days before attempting to renew the certificates.")
+    parser.add_argument("--quiet", "-q",
+        action="store_true",
+        help="Quiet mode.")
+    parser.add_argument("--verbose", "-v",
+        action="store_true",
+        help="Increase verbosity.")
+    parser.add_argument("--force", "-f",
+        action="store_true",
+        help="Force certificates renewal without checking their expiration date.")
     args = parser.parse_args()
     
     if args.quiet:
@@ -439,47 +221,21 @@ def main():
     else:
         log_level = logging.INFO
         
-    logging.basicConfig(stream=sys.stdout, level=log_level,
+    logging.basicConfig(stream=sys.stdout, level=log_level, format="%(levelname)s:%(message)s")
-                        format="%(levelname)s:%(message)s")
 
-    config = Config(args.config, args.config_dir)
+    for domain in args.domains:
-    try:
+        account_key = args.account_key.format(domain=domain)
-        config.load()
+        privkey = args.privkey.format(domain=domain)
-    except IOError as e:
+        csr = args.csr.format(domain=domain)
-        logging.error(f"Unable to load config file: {e}")
+        certs_dir = args.certs.format(domain=domain)
-        return 1
+        challenge_dir = args.challenge.format(domain=domain)
 
-    try:
+        renew_cert(domain, 
-        domains = config.parse_domains()
+            account_key_path=account_key,
-    except RuntimeError as e:
+            privkey_path=privkey,
-        logging.error(f"Unable to parse domains: {e}")
+            csr_path=csr,
-        return 2
+            certs_dir=certs_dir,
-
+            challenge_dir=challenge_dir,
-    try:
+            directory_url=args.directory_url,
-        domains_to_renew = []
+            days_before_renewal=args.days,
-        for domain in domains:
+            force=args.force)
-            if not (args.force or domain.check_certificate_expiration_date()):
-                logging.info(f"{domain.name} certificate will not be renewed.")
-                continue
-
-            logging.info(f"{domain.name} certificate will be renewed.")
-            domains_to_renew.append(domain)
-
-        if len(domains_to_renew) == 0:
-            logging.info("No domain to renew. Aborting.")
-            return 0
-
-        account = config.parse_account()
-        acme_client = account.register(config.directory_url)
-        challenge = Challenge(config.challenge_dir, acme_client)
-
-        for domain in domains_to_renew:
-            domain.renew_cert(challenge, force=args.force)
-    except Exception as e:
-        logging.exception(e)
-        return 3
-
-
-if __name__ == '__main__':
-    import argparse
-    exit(main())

molecule/default/prepare.yml

@@ -3,13 +3,8 @@
   gather_facts: no
 
   tasks:
-    - name: Update APT cache
+    - name: Update apt cache
       raw: apt update
 
-    - name: Install Python3 package
+    - name: Install python3 package
       raw: apt install -y --no-install-recommends python3
-
-    - name: Install cron package
-      package:
-        name: cron
-        state: present

tasks/acme_challenge.yml

@@ -0,0 +1,74 @@
+- name: Create {{ domain_name }} certificates directory
+  file:
+    path: "{{ acme_certs_dir }}/{{ domain_name }}.d"
+    state: directory
+    owner: root
+    group: "{{ acme_ssl_group }}"
+    mode: "755"
+  tags: acme_install
+
+- name: Generate Let's Encrypt account key
+  openssl_privatekey:
+    path: "{{ acme_accounts_dir }}/{{ acme_account_key }}"
+    owner: root
+    group: root
+    mode: "600"
+    type: RSA
+    size: 4096
+  tags: acme_account
+
+- name: Generate Let's Encrypt private key for {{ domain_name }}
+  openssl_privatekey:
+    path: "{{ acme_keys_dir }}/{{ domain_name }}.pem"
+    owner: root
+    group: "{{ acme_ssl_group }}"
+    mode: "640"
+    type: RSA
+    size: 4096
+
+- name: Generate Let's Encrypt CSR for {{ domain_name }}
+  openssl_csr:
+    path: "{{ acme_csr_dir }}/{{ domain_name }}.csr"
+    owner: root
+    group: "{{ acme_ssl_group }}"
+    mode: "644"
+    privatekey_path: "{{ acme_keys_dir }}/{{ domain_name }}.pem"
+    common_name: "{{ domain_name }}"
+
+- name: Begin Let's Encrypt challenges for {{ domain_name }}
+  acme_certificate:
+    acme_directory: "{{ acme_directory }}"
+    acme_version: "{{ acme_version }}"
+    account_key_src: "{{ acme_accounts_dir }}/{{ acme_account_key }}"
+    account_email: "{{ acme_email }}"
+    terms_agreed: yes
+    challenge: http-01
+    csr: "{{ acme_csr_dir }}/{{ domain_name }}.csr"
+    dest: "{{ acme_certs_dir }}/{{ domain_name }}.d/cert.pem"
+    fullchain_dest: "{{ acme_certs_dir }}/{{ domain_name }}.d/fullchain.pem"
+    remaining_days: 30
+  register: _acme_challenge
+
+- name: Implement and complete Let's Encrypt challenge for {{ domain_name }}
+  when: _acme_challenge is changed
+  block:
+  - name: Implement http-01 challenge files for {{ domain_name }}
+    copy:
+      content: "{{ _acme_challenge.challenge_data[domain_name]['http-01'].resource_value }}"
+      dest: "{{ acme_challenge_dir }}/{{ _acme_challenge.challenge_data[domain_name]['http-01'].resource }}"
+      owner: root
+      group: root
+      mode: "644"
+  
+  - name: Complete Let's Encrypt challenges for {{ domain_name }}
+    acme_certificate:
+      acme_directory: "{{ acme_directory }}"
+      acme_version: "{{ acme_version }}"
+      account_key_src: "{{ acme_accounts_dir }}/{{ acme_account_key }}"
+      account_email: "{{ acme_email }}"
+      challenge: http-01
+      csr: "{{ acme_csr_dir }}/{{ domain_name }}.csr"
+      dest: "{{ acme_certs_dir }}/{{ domain_name }}.d/cert.pem"
+      chain_dest: "{{ acme_certs_dir }}/{{ domain_name }}.d/chain.pem"
+      fullchain_dest: "{{ acme_certs_dir }}/{{ domain_name }}.d/fullchain.pem"
+      data: "{{ _acme_challenge }}"

tasks/main.yml

@@ -1,75 +1,48 @@
 - name: Install ACME dependencies
   package:
-    name: "{{ package }}"
+    name: python3-acme
     state: present
-  loop: "{{ acme_packages }}"
-  loop_control:
-    loop_var: package
   tags: acme_install
 
 - name: Install SSL dependencies
   package:
     name: ssl-cert
     state: present
-  tags: acme_install
 
-- name: Create ACME config directories
+- name: Create Let's Encrypt config directories
   file:
     path: "{{ config_dir }}"
     state: directory
     owner: root
-    group: root
+    group: "{{ acme_ssl_group }}"
-    mode: "755"
+    mode: "711"
   loop:
-    - "{{ acme_ssl_dir }}"
+  - "{{ acme_config_dir }}"
-    - "{{ acme_config_dir }}"
+  - "{{ acme_keys_dir }}"
-    - "{{ acme_certs_dir }}"
+  - "{{ acme_accounts_dir }}"
-    - "{{ acme_csr_dir }}"
+  - "{{ acme_csr_dir }}"
   loop_control:
     loop_var: config_dir
   tags: acme_install
 
-- name: Create ACME private keys directory
+- name: Create challenge directory
   file:
-    path: "{{ acme_keys_dir }}"
+    path: "{{ acme_challenge_dir }}/.well-known/acme-challenge"
-    state: directory
-    owner: root
-    group: "{{ acme_ssl_group }}"
-    mode: "750"
-  tags: acme_install
-
-- name: Create ACME accounts directory
-  file:
-    path: "{{ acme_accounts_dir }}"
     state: directory
     owner: root
     group: root
-    mode: "750"
+    mode: "755"
   tags: acme_install
 
-- name: Copy ACME config file
+- name: Perform ACME challenge for each domain
-  copy:
+  include_tasks:
-    content: "{{ acme_config | to_nice_yaml(indent=2) }}"
+    file: acme_challenge.yml
-    dest: "{{ acme_config_file }}"
+    apply:
-    owner: root
+      tags: acme_challenge
-    group: root
+  loop: "{{ acme_domains | unique }}"
-    mode: "640"
-  tags: [acme_install, acme_config]
-
-- name: Copy ACME domain config files
-  copy:
-    content: "{{ domain | to_nice_yaml(indent=2) }}"
-    dest: "{{ acme_config_dir }}/{{ domain_name }}.yml"
-    owner: root
-    group: root
-    mode: "640"
-  loop: "{{ (acme_domains.keys() | list) if acme_domains is mapping else acme_domains }}"
   loop_control:
-    label: "{{ domain_name }}"
+    loop_var: domain_name
-  vars:
+  tags: acme_challenge
-    domain_name: "{{ item if item is string else item.name }}"
-    domain: "{{ acme_domains[item] if item is string else item }}"
-  tags: [acme_install, acme_config]
 
 - name: Create directory for certificate renewal tool
   file:
@@ -78,40 +51,31 @@
     group: root
     mode: "755"
     state: directory
-  tags: acme_install
+  tags: acme_renew
 
 - name: Copy script to renew ACME certificates
   copy:
     src: acme_renew_cert.py
-    dest: "{{ acme_script_dir }}/acme_renew_cert.py"
+    dest: /opt/acme/acme_renew_cert.py
     owner: root
     group: root
     mode: "755"
-  tags: acme_install
+  tags: acme_renew
-
-- name: Create '{{ acme_script_bin }}' symlink for ACME renewal script
-  file:
-    src: "{{ acme_script_dir }}/acme_renew_cert.py"
-    dest: "{{ acme_script_bin }}"
-    state: link
-    owner: root
-    group: root
-    mode: "755"
-  tags: acme_install
-
-- name: Perform ACME challenge for each domain
-  command: acme-renew-cert -c {{ acme_config_file | quote }} -d {{ acme_config_dir | quote }}
-  changed_when: "'No domain to renew' not in _acme_challenge.stdout"
-  register: _acme_challenge
-  tags: acme_challenge
   
 - name: Setup cron job for ACME certificates renewal of {{ domain_name }}
   cron:
-    user: root
+    name: acme renew {{ domain_name }} cert
-    name: acme-renew-cert
+    job: >-
-    cron_file: acme-renew-cert
+      bash -c 'sleep $((RANDOM \% 3600))' && /opt/acme/acme_renew_cert.py {{ domain_name }} -q
-    job: "{{ acme_script_bin }} -q -c {{ acme_config_file | quote }} -d {{ acme_config_dir | quote }}"
+      -a {{ (acme_accounts_dir + '/' + acme_account_key) | quote }}
+      -p {{ acme_keys_dir | quote }}/{domain}.pem
+      -r {{ acme_csr_dir | quote }}/{domain}.csr
+      -o {{ acme_certs_dir | quote }}/{domain}.d
+      -c {{ acme_challenge_dir | quote }}/.well-known/acme-challenge
     minute: "30"
     hour: "2"
     state: present
-  tags: acme_cron
+  loop: "{{ acme_domains | unique }}"
+  loop_control:
+    loop_var: domain_name
+  tags: acme_renew

vars/main.yml

@@ -1,4 +1 @@
 acme_version: 2
-acme_packages:
-  - python3-acme
-  - python3-yaml