Merge pull request 'Create separate files for domain configs' (#8) from 7-create-separate-files-for-domain-configs into master

Reviewed-on: #8

defaults/main.yml

@@ -1,27 +1,36 @@
-acme_config_dir: /etc/ssl
+acme_ssl_dir: /etc/ssl
-acme_config_file: "{{ acme_config_dir }}/acme.yml"
+acme_config_dir: "{{ acme_ssl_dir }}/acme.d"
-acme_keys_dir: "{{ acme_config_dir }}/private"
+acme_config_file: "{{ acme_ssl_dir }}/acme.yml"
-acme_csr_dir: "{{ acme_config_dir }}/csr"
+acme_keys_dir: "{{ acme_ssl_dir }}/private"
-acme_certs_dir: "{{ acme_config_dir }}/certs"
+acme_csr_dir: "{{ acme_ssl_dir }}/csr"
-acme_accounts_dir: "{{ acme_config_dir }}/accounts"
+acme_certs_dir: "{{ acme_ssl_dir }}/certs"
+acme_accounts_dir: "{{ acme_ssl_dir }}/accounts"
 acme_script_dir: /opt/acme
 acme_script_bin: /usr/local/bin/acme-renew-cert
 
 acme_ssl_group: ssl-cert
+
+acme_account_private_key: "{{ acme_accounts_dir }}/acme_account.key"
+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_accounts_dir }}/acme_account.key"
+    private_key: "{{ acme_account_private_key }}"
-    email: acme@example.com
+    email: "{{ acme_account_email }}"
-    owner: root
+    owner: "{{ acme_account_owner }}"
-    group: root
+    group: "{{ acme_account_group }}"
-  directory_url: https://acme-staging-v02.api.letsencrypt.org/directory
+  directory_url: "{{ acme_directory_url }}"
-  challenge_dir: /var/www/acme/.well-known/acme-challenge
+  challenge_dir: "{{ acme_challenge_dir }}"
-  domains:
-    example.com:
-      alt_names: 
-        - test.example.com
-      owner: root
-      group: "{{ acme_ssl_group }}"
-      remaining_days: 30
-      hooks:
-        - systemctl reload nginx

files/acme_renew_cert.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 
+import glob
 import grp
 import logging
 import os
@@ -28,6 +29,8 @@ try:
 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"
@@ -176,9 +179,12 @@ class Challenge:
 class Domain(PrivateKey):
     def __init__(self, name: str, owner: str, group: str,
                  alt_names: List[str] = None,
-                 remaining_days: int = CERT_REMAINING_DAYS,
+                 cert: str = None, chain_cert: str = None,
+                 fullchain_cert: str = None, private_key: str = None,
+                 csr: str = None, remaining_days: int = CERT_REMAINING_DAYS,
                  hooks: List[str] = None):
-        private_key = os.path.join(SSL_KEYS_DIR, name + '.key')
+        if private_key is None:
+            private_key = os.path.join(SSL_KEYS_DIR, name + '.key')
         super().__init__(private_key, owner, group)
 
         self.name = name
@@ -187,10 +193,23 @@ class Domain(PrivateKey):
         self.hooks = [] if hooks is None else hooks
 
         certs_dir = os.path.join(SSL_CERTS_DIR, name + '.d')
-        self.csr = os.path.join(SSL_CSR_DIR, name + '.csr')
+        if cert is None:
-        self.cert = os.path.join(certs_dir, 'cert.pem')
+            self.cert = os.path.join(certs_dir, 'cert.pem')
-        self.chain_cert = os.path.join(certs_dir, 'chain.pem')
+        else:
-        self.fullchain_cert = os.path.join(certs_dir, 'fullchain.pem')
+            self.cert = cert
+        if chain_cert is None:
+            self.chain_cert = os.path.join(certs_dir, 'chain.pem')
+        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:
@@ -315,14 +334,95 @@ class Domain(PrivateKey):
                 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 "
                     "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=SSL_CONFIG_FILE,
+    parser.add_argument("--config", "-c", default=ACME_CONFIG_FILE,
                         help="Path to the config file.")
+    parser.add_argument("--config-dir", "-d", default=ACME_CONFIG_DIR,
+                        help="Path to the config directory.")
     parser.add_argument("--quiet", "-q", action="store_true",
                         help="Quiet mode.")
     parser.add_argument("--verbose", "-v", action="store_true",
@@ -342,40 +442,22 @@ def main():
     logging.basicConfig(stream=sys.stdout, level=log_level,
                         format="%(levelname)s:%(message)s")
 
+    config = Config(args.config, args.config_dir)
     try:
-        with open(args.config, 'r') as ifd:
+        config.load()
-            config = yaml.safe_load(ifd)
     except IOError as e:
         logging.error(f"Unable to load config file: {e}")
         return 1
 
-    if isinstance(config["domains"], list):
+    try:
-        domains = {domain["name"]: domain for domain in config["domains"]}
+        domains = config.parse_domains()
-        if len(domains) != len(config["domains"]):
+    except RuntimeError as e:
-            domain_uniques = set()
+        logging.error(f"Unable to parse domains: {e}")
-            domain_duplicates = set()
+        return 2
-            for domain in config["domains"]:
-                domain_name = domain["name"]
-                if domain_name not in domain_uniques:
-                    domain_uniques.add(domain_name)
-                else:
-                    domain_duplicates.add(domain_name)
-            logging.error(f"Duplicate domain name(s) found: {domain_uniques}")
-            return 2
-        config["domains"] = domains
 
     try:
         domains_to_renew = []
-        for domain_name, domain_details in config["domains"].items():
+        for domain in domains:
-            remaining_days = domain_details.get("remaining_days",
-                                                CERT_REMAINING_DAYS)
-            domain = Domain(domain_name,
-                            owner=domain_details.get("owner", "root"),
-                            group=domain_details.get("group", SSL_GROUP),
-                            alt_names=domain_details.get("alt_names"),
-                            remaining_days=remaining_days,
-                            hooks=domain_details.get("hooks"))
-
             if not (args.force or domain.check_certificate_expiration_date()):
                 logging.info(f"{domain.name} certificate will not be renewed.")
                 continue
@@ -387,18 +469,9 @@ def main():
             logging.info("No domain to renew. Aborting.")
             return 0
 
-        account_config = config["account"]
+        account = config.parse_account()
-        account = Account(
+        acme_client = account.register(config.directory_url)
-            private_key=account_config.get("private_key", ACME_ACCOUNT_KEY),
+        challenge = Challenge(config.challenge_dir, acme_client)
-            email=account_config["email"],
-            owner=account_config.get("owner", "root"),
-            group=account_config.get("group", "root"))
-
-        directory_url = config.get("directory_url", LETSENCRYPT_DIRECTORY_URL)
-        acme_client = account.register(directory_url)
-
-        challenge_dir = config.get("challenge_dir", ACME_CHALLENGE_DIR)
-        challenge = Challenge(challenge_dir, acme_client)
 
         for domain in domains_to_renew:
             domain.renew_cert(challenge, force=args.force)

tasks/main.yml

@@ -21,6 +21,7 @@
     group: root
     mode: "755"
   loop:
+    - "{{ acme_ssl_dir }}"
     - "{{ acme_config_dir }}"
     - "{{ acme_certs_dir }}"
     - "{{ acme_csr_dir }}"
@@ -52,7 +53,22 @@
     dest: "{{ acme_config_file }}"
     owner: root
     group: root
-    mode: "600"
+    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 }}"
+  vars:
+    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
@@ -84,7 +100,7 @@
   tags: acme_install
 
 - name: Perform ACME challenge for each domain
-  command: acme-renew-cert -c {{ acme_config_file | quote }}
+  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
@@ -94,7 +110,7 @@
     user: root
     name: acme-renew-cert
     cron_file: acme-renew-cert
-    job: "{{ acme_script_bin }} -q -c {{ acme_config_file | quote }}"
+    job: "{{ acme_script_bin }} -q -c {{ acme_config_file | quote }} -d {{ acme_config_dir | quote }}"
     minute: "30"
     hour: "2"
     state: present