Optimized cloudflare implementation

library/cert_check_exists.py

@@ -0,0 +1,92 @@
+#!/usr/bin/python
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+DOCUMENTATION = r'''
+---
+module: cert_check_exists
+short_description: Check if a SSL certificate exists for a domain
+description:
+  - Checks if any certificate covers the given domain.
+options:
+  domain:
+    description:
+      - Domain name to check for in the certificates.
+    required: true
+    type: str
+  cert_base_path:
+    description:
+      - Path where certificates are stored.
+    required: false
+    type: str
+    default: /etc/letsencrypt/live
+  debug:
+    description:
+      - Enable verbose debug output.
+    required: false
+    type: bool
+    default: false
+author:
+  - Kevin Veen-Birkenbach
+'''
+
+EXAMPLES = r'''
+- name: Check if cert exists
+  cert_check_exists:
+    domain: "matomo.cymais.cloud"
+    cert_base_path: "/etc/letsencrypt/live"
+  register: result
+'''
+
+RETURN = r'''
+exists:
+  description: True if a certificate covering the domain exists, false otherwise.
+  type: bool
+  returned: always
+'''
+
+import os
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.cert_utils import CertUtils
+
+def cert_exists(domain, cert_files, debug=False):
+    for cert_path in cert_files:
+        cert_text = CertUtils.run_openssl(cert_path)
+        if not cert_text:
+            continue
+        sans = CertUtils.extract_sans(cert_text)
+        if debug:
+            print(f"Checking {cert_path}: {sans}")
+        for entry in sans:
+            if entry == domain or (entry.startswith('*.') and domain.endswith('.' + entry[2:])):
+                return True
+    return False
+
+def cert_check_exists(module):
+    domain = module.params['domain']
+    cert_base_path = module.params['cert_base_path']
+    debug = module.params['debug']
+
+    cert_files = CertUtils.list_cert_files(cert_base_path)
+
+    exists = cert_exists(domain, cert_files, debug)
+
+    module.exit_json(exists=exists)
+
+def main():
+    module_args = dict(
+        domain=dict(type='str', required=True),
+        cert_base_path=dict(type='str', required=False, default='/etc/letsencrypt/live'),
+        debug=dict(type='bool', required=False, default=False),
+    )
+
+    module = AnsibleModule(
+        argument_spec=module_args,
+        supports_check_mode=True
+    )
+
+    cert_check_exists(module)
+
+if __name__ == '__main__':
+    main()

library/cert_folder_find.py

@@ -5,7 +5,7 @@ __metaclass__ = type
 
 DOCUMENTATION = r'''
 ---
-module: find_cert_folder
+module: cert_folder_find
 short_description: Find SSL certificate folder covering a given domain
 description:
   - Searches through certificates to find a folder that covers the given domain.
@@ -38,7 +38,7 @@ author:
 
 EXAMPLES = r'''
 - name: Find cert folder for matomo.cymais.cloud
-  find_cert_folder:
+  cert_folder_find:
     domain: "matomo.cymais.cloud"
     certbot_flavor: "san"
     cert_base_path: "/etc/letsencrypt/live"
@@ -53,42 +53,18 @@ folder:
 '''
 
 import os
-import subprocess
 from ansible.module_utils.basic import AnsibleModule
-
-def run_openssl(cert_path):
-    try:
-        output = subprocess.check_output(
-            ['openssl', 'x509', '-in', cert_path, '-noout', '-text'],
-            universal_newlines=True
-        )
-        return output
-    except subprocess.CalledProcessError:
-        return ""
-
-def extract_sans(cert_text):
-    dns_entries = []
-    in_san = False
-    for line in cert_text.splitlines():
-        line = line.strip()
-        if 'X509v3 Subject Alternative Name:' in line:
-            in_san = True
-            continue
-        if in_san:
-            if not line:
-                break
-            dns_entries += [e.strip().replace('DNS:', '') for e in line.split(',') if e.strip()]
-    return dns_entries
+from ansible.module_utils.cert_utils import CertUtils  # IMPORT
 
 def find_matching_folders(domain, cert_files, flavor, debug):
     exact_matches = []
     wildcard_matches = []
 
     for cert_path in cert_files:
-        cert_text = run_openssl(cert_path)
+        cert_text = CertUtils.run_openssl(cert_path)
         if not cert_text:
             continue
-        sans = extract_sans(cert_text)
+        sans = CertUtils.extract_sans(cert_text)
         if debug:
             print(f"Checking {cert_path}: {sans}")
         for entry in sans:
@@ -106,16 +82,13 @@ def find_matching_folders(domain, cert_files, flavor, debug):
     else:
         return []
 
-def find_cert_folder(module):
+def cert_folder_find(module):
     domain = module.params['domain']
     certbot_flavor = module.params['certbot_flavor']
     cert_base_path = module.params['cert_base_path']
     debug = module.params['debug']
 
-    cert_files = []
-    for root, dirs, files in os.walk(cert_base_path):
-        if 'cert.pem' in files:
-            cert_files.append(os.path.join(root, 'cert.pem'))
+    cert_files = CertUtils.list_cert_files(cert_base_path)
 
     if debug:
         print(f"Found {len(cert_files)} cert.pem files under {cert_base_path}")
@@ -126,7 +99,7 @@ def find_cert_folder(module):
         if debug:
             print("Fallback: searching SAN matches without SAN structure parsing")
         for cert_path in cert_files:
-            cert_text = run_openssl(cert_path)
+            cert_text = CertUtils.run_openssl(cert_path)
             if f"DNS:{domain}" in cert_text:
                 preferred.append(os.path.dirname(cert_path))
 
@@ -151,7 +124,7 @@ def main():
         supports_check_mode=True
     )
 
-    find_cert_folder(module)
+    cert_folder_find(module)
 
 if __name__ == '__main__':
-    main()
+    main()