Optimized nextcloud plugin tasks

library/cert_check_exists.py

@@ -1,79 +1,7 @@
-#!/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),
@@ -82,11 +10,17 @@ def main():
     )
 
     module = AnsibleModule(
-        argument_spec=module_args,
-        supports_check_mode=True
+        argument_spec=module_args
     )
 
-    cert_check_exists(module)
+    domain = module.params['domain']
+    cert_base_path = module.params['cert_base_path']
+    debug = module.params['debug']
+
+    folder = CertUtils.find_cert_for_domain(domain, cert_base_path, debug)
+    exists = folder is not None
+
+    module.exit_json(exists=exists)
 
 if __name__ == '__main__':
     main()

library/cert_folder_find.py

@@ -1,130 +1,27 @@
-#!/usr/bin/python
-
-from __future__ import absolute_import, division, print_function
-__metaclass__ = type
-
-DOCUMENTATION = r'''
----
-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.
-options:
-  domain:
-    description:
-      - Domain name to search for in the certificates.
-    required: true
-    type: str
-  certbot_flavor:
-    description:
-      - Certificate type. Either 'san', 'wildcard', or 'dedicated'.
-    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: Find cert folder for matomo.cymais.cloud
-  cert_folder_find:
-    domain: "matomo.cymais.cloud"
-    certbot_flavor: "san"
-    cert_base_path: "/etc/letsencrypt/live"
-  register: result
-'''
-
-RETURN = r'''
-folder:
-  description: The name of the folder covering the domain.
-  type: str
-  returned: always
-'''
-
-import os
 from ansible.module_utils.basic import AnsibleModule
-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 = 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:
-                exact_matches.append(os.path.dirname(cert_path))
-            elif entry.startswith('*.'):
-                base = entry[2:]
-                if domain.endswith('.' + base):
-                    wildcard_matches.append(os.path.dirname(cert_path))
-
-    if flavor in ('san', 'dedicated'):
-        return exact_matches or wildcard_matches
-    elif flavor == 'wildcard':
-        return wildcard_matches or exact_matches
-    else:
-        return []
-
-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 = CertUtils.list_cert_files(cert_base_path)
-
-    if debug:
-        print(f"Found {len(cert_files)} cert.pem files under {cert_base_path}")
-
-    preferred = find_matching_folders(domain, cert_files, certbot_flavor, debug)
-
-    if not preferred and certbot_flavor == 'san':
-        if debug:
-            print("Fallback: searching SAN matches without SAN structure parsing")
-        for cert_path in cert_files:
-            cert_text = CertUtils.run_openssl(cert_path)
-            if f"DNS:{domain}" in cert_text:
-                preferred.append(os.path.dirname(cert_path))
-
-    if not preferred:
-        module.fail_json(msg=f"No certificate covering domain {domain} found.")
-
-    preferred = sorted(preferred, key=lambda p: (p.count('-'), len(p)))
-    folder = os.path.basename(preferred[0])
-
-    module.exit_json(folder=folder)
+from ansible.module_utils.cert_utils import CertUtils
 
 def main():
     module_args = dict(
         domain=dict(type='str', required=True),
-        certbot_flavor=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
+        argument_spec=module_args
     )
 
-    cert_folder_find(module)
+    domain = module.params['domain']
+    cert_base_path = module.params['cert_base_path']
+    debug = module.params['debug']
+
+    folder = CertUtils.find_cert_for_domain(domain, cert_base_path, debug)
+
+    if folder is None:
+        module.fail_json(msg=f"No certificate covering domain {domain} found.")
+    else:
+        module.exit_json(folder=folder)
 
 if __name__ == '__main__':
     main()

main.py

@@ -9,7 +9,14 @@ def run_ansible_vault(action, filename, password_file):
     cmd = ["ansible-vault", action, filename, "--vault-password-file", password_file]
     subprocess.run(cmd, check=True) 
 
-def run_ansible_playbook(inventory: str, playbook: str, modes: dict, limit: str = None, password_file: str = None, verbose: int = 0):
+def run_ansible_playbook(inventory: str, playbook: str, modes: dict, limit: str = None, password_file: str = None, verbose: int = 0, skip_tests: bool = False):
+    print("\n🛠️  Building project (make build)...\n")
+    subprocess.run(["make", "build"], check=True)
+
+    if not skip_tests:
+        print("\n🧪 Running tests (make test)...\n")
+        subprocess.run(["make", "test"], check=True)
+    
     """Execute an ansible-playbook command with optional parameters."""
     cmd = ["ansible-playbook", "-i", inventory, playbook]
     
@@ -18,7 +25,6 @@ def run_ansible_playbook(inventory: str, playbook: str, modes: dict, limit: str
     
     if modes:
         for key, value in modes.items():
-            # Convert boolean values to lowercase strings
             arg_value = f"{str(value).lower()}" if isinstance(value, bool) else f"{value}"
             cmd.extend(["-e", f"{key}={arg_value}"])
     
@@ -28,9 +34,9 @@ def run_ansible_playbook(inventory: str, playbook: str, modes: dict, limit: str
         cmd.extend(["--ask-vault-pass"])
     
     if verbose:
-        # Append a single flag with multiple "v"s (e.g. -vvv)
         cmd.append("-" + "v" * verbose)
-    subprocess.run(['make','build'], check=True)
+    
+    print("\n🚀 Launching Ansible Playbook...\n")
     subprocess.run(cmd, check=True)
 
 def main():
@@ -60,6 +66,7 @@ def main():
     playbook_parser.add_argument("--cleanup", action="store_true", help="Enable cleanup mode")
     playbook_parser.add_argument("--debug", action="store_true", help="Enable debugging output")
     playbook_parser.add_argument("--password-file", help="Path to the Vault password file")
+    playbook_parser.add_argument("--skip-tests", action="store_true", help="Skip running make test before executing the playbook")
     playbook_parser.add_argument("-v", "--verbose", action="count", default=0,
                                  help=("Increase verbosity. This option can be specified multiple times "
                                        "to increase the verbosity level (e.g., -vvv for more detailed debug output)."))
@@ -79,8 +86,15 @@ def main():
             "host_type": args.host_type
         }
 
-        # Use a fixed playbook file "playbook.yml"
-        run_ansible_playbook(args.inventory, f"{script_dir}/playbook.yml", modes, args.limit, args.password_file, args.verbose)
+        run_ansible_playbook(
+            inventory=args.inventory,
+            playbook=f"{script_dir}/playbook.yml",
+            modes=modes,
+            limit=args.limit,
+            password_file=args.password_file,
+            verbose=args.verbose,
+            skip_tests=args.skip_tests
+        )
 
 if __name__ == "__main__":
     main()

module_utils/cert_utils.py

@@ -1,9 +1,16 @@
 #!/usr/bin/python
 
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
 import os
 import subprocess
+import time
 
 class CertUtils:
+    _domain_cert_mapping = None
+    _cert_snapshot = None
+
     @staticmethod
     def run_openssl(cert_path):
         try:
@@ -37,3 +44,92 @@ class CertUtils:
             if 'cert.pem' in files:
                 cert_files.append(os.path.join(root, 'cert.pem'))
         return cert_files
+
+    @staticmethod
+    def matches(domain, san):
+        """RFC compliant SAN matching."""
+        if san.startswith('*.'):
+            base = san[2:]
+            # Wildcard matches ONLY one additional label
+            if domain == base:
+                return False
+            if domain.endswith('.' + base) and domain.count('.') == base.count('.') + 1:
+                return True
+            return False
+        else:
+            return domain == san
+
+
+    @classmethod
+    def build_snapshot(cls, cert_base_path):
+        snapshot = []
+        for cert_file in cls.list_cert_files(cert_base_path):
+            try:
+                stat = os.stat(cert_file)
+                snapshot.append((cert_file, stat.st_mtime, stat.st_size))
+            except FileNotFoundError:
+                continue
+        snapshot.sort()
+        return snapshot
+
+    @classmethod
+    def snapshot_changed(cls, cert_base_path):
+        current_snapshot = cls.build_snapshot(cert_base_path)
+        if cls._cert_snapshot != current_snapshot:
+            cls._cert_snapshot = current_snapshot
+            return True
+        return False
+
+    @classmethod
+    def refresh_cert_mapping(cls, cert_base_path, debug=False):
+        cert_files = cls.list_cert_files(cert_base_path)
+        mapping = {}
+        for cert_path in cert_files:
+            cert_text = cls.run_openssl(cert_path)
+            if not cert_text:
+                continue
+            sans = cls.extract_sans(cert_text)
+            folder = os.path.basename(os.path.dirname(cert_path))
+            for san in sans:
+                if san not in mapping:
+                    mapping[san] = folder
+        cls._domain_cert_mapping = mapping
+        if debug:
+            print(f"[DEBUG] Refreshed domain-to-cert mapping: {mapping}")
+
+    @classmethod
+    def ensure_cert_mapping(cls, cert_base_path, debug=False):
+        if cls._domain_cert_mapping is None or cls.snapshot_changed(cert_base_path):
+            cls.refresh_cert_mapping(cert_base_path, debug)
+
+    @classmethod
+    def find_cert_for_domain(cls, domain, cert_base_path, debug=False):
+        cls.ensure_cert_mapping(cert_base_path, debug)
+
+        exact_match = None
+        wildcard_match = None
+
+        for san, folder in cls._domain_cert_mapping.items():
+            if san == domain:
+                exact_match = folder
+                break
+            if san.startswith('*.'):
+                base = san[2:]
+                if domain.count('.') == base.count('.') + 1 and domain.endswith('.' + base):
+                    wildcard_match = folder
+
+        if exact_match:
+            if debug:
+                print(f"[DEBUG] Exact match for {domain} found in {exact_match}")
+            return exact_match
+
+        if wildcard_match:
+            if debug:
+                print(f"[DEBUG] Wildcard match for {domain} found in {wildcard_match}")
+            return wildcard_match
+
+        if debug:
+            print(f"[DEBUG] No certificate folder found for {domain}")
+
+        return None
+

roles/docker-nextcloud/tasks/plugin.yml

@@ -35,17 +35,18 @@
   become: false
   register: plugin_vars_file
 
-- name: Apply configuration to {{ plugin_key }} 
-  block:
-  - name: Load {{ plugin_key }} configuration variables
+
+- name: "Load {{ plugin_key }} configuration variables"
   include_vars:
     file: "{{nextcloud_control_node_plugin_vars_directory}}{{ plugin_key }}.yml"
+  when: plugin_vars_file.stat.exists
     
 - name: "Set {{ item.configkey }} for {{ item.appid }}"
   loop: "{{ plugin_configuration }}"
   command: >
     {{ nextcloud_docker_exec_occ }} config:app:set {{ item.appid }} {{ item.configkey }} --value '{{ item.configvalue | to_json if item.configvalue is mapping else item.configvalue }}'
-
+  register: config_set_result
+  changed_when: (config_set_result.stdout is defined) and ("Config value were not updated" not in config_set_result.stdout)
   when: plugin_vars_file.stat.exists
 
 - name: Check if {{nextcloud_control_node_plugin_tasks_directory}}{{ plugin_key }}.yml exists

roles/nginx-https-get-cert/tasks/flavors/san.yml

@@ -20,6 +20,7 @@
     {% endif %}
     {{ '--mode-test' if mode_test | bool else '' }}
   register: certbundle_result
+  changed_when: "'Certificate not yet due for renewal' not in certbundle_result.stdout"
   when: run_once_san_certs is not defined
 
 - name: run the san tasks once

roles/nginx-https-get-cert/tasks/main.yml

@@ -20,7 +20,6 @@
 - name: Find SSL cert folder for domain
   cert_folder_find:
     domain: "{{ domain }}"
-    certbot_flavor: "{{ certbot_flavor }}"
     cert_base_path: "{{ certbot_cert_path }}"
     debug: "{{ enable_debug | default(false) }}"
   register: cert_folder_result
@@ -29,3 +28,8 @@
 - name: Set fact
   set_fact:
     ssl_cert_folder: "{{ cert_folder_result.folder }}"
+
+- name: Ensure ssl_cert_folder is set
+  fail:
+    msg: "No certificate folder found for domain {{ domain }}"
+  when: ssl_cert_folder is undefined or ssl_cert_folder is none

roles/nginx-redirect-domain/tasks/main.yml

@@ -1,21 +0,0 @@
----
-- name: "include task receive certbot certificate"
-  include_role: 
-    name: nginx-https-get-cert
-  vars:
-    domain: "{{item.source}}"
-  loop: "{{domain_mappings}}"
-
-- name: The domains for which a www. redirect will be implemented
-  debug:
-    var: domain_mappings
-  when: enable_debug | bool
-
-- name: configure nginx redirect configurations
-  vars:
-    item: "{{item}}"
-  template: 
-    src:  redirect.domain.nginx.conf.j2
-    dest: "{{nginx.directories.http.servers}}{{item.source}}.conf"
-  loop:   "{{domain_mappings}}"
-  notify: restart nginx

roles/nginx-redirect-domain/templates/redirect.domain.nginx.conf.j2

@@ -1,8 +0,0 @@
-server {
-  {% set domain = item.source %}
-  {% set target = item.target %}
-  server_name {{domain}};
-  {% include 'roles/letsencrypt/templates/ssl_header.j2' %}
-
-  return 301 https://{{target}}$request_uri;
-}

roles/nginx-redirect-domains/tasks/main.yml

@@ -0,0 +1,7 @@
+---
+- name: "Include domains redirects"
+  include_tasks: redirect-domain.yml
+  vars:
+    domain: "{{ item.source }}"
+    target: "{{ item.target }}"
+  loop: "{{domain_mappings}}"

roles/nginx-redirect-domains/tasks/redirect-domain.yml

@@ -0,0 +1,9 @@
+- name: "include task receive certbot certificate"
+  include_role: 
+    name: nginx-https-get-cert
+
+- name: configure nginx redirect configurations
+  template:
+    src:  redirect.domain.nginx.conf.j2
+    dest: "{{ nginx.directories.http.servers }}{{ domain }}.conf"
+  notify: restart nginx

roles/nginx-redirect-domains/templates/redirect.domain.nginx.conf.j2

@@ -0,0 +1,6 @@
+server {
+  server_name {{ domain }};
+  {% include 'roles/letsencrypt/templates/ssl_header.j2' %}
+
+  return 301 https://{{ target }}$request_uri;
+}

roles/nginx-redirect-www/README.md

@@ -7,7 +7,7 @@ Automates the creation of Nginx server blocks that redirect all `www.` subdomain
 This role will:
 - **Discover** existing `*.conf` vhosts in your Nginx servers directory  
 - **Filter** domains with or without your `primary_domain`  
-- **Generate** redirect rules via the `nginx-redirect-domain` role  
+- **Generate** redirect rules via the `nginx-redirect-domains` role  
 - **Optionally** include a wildcard redirect template (experimental) ⭐️  
 - **Clean up** leftover configs when running in cleanup mode 🧹  
 

roles/nginx-redirect-www/tasks/main.yml

@@ -1,16 +1,11 @@
 ---
-- name: "Debug: all_domains"
-  debug:
-    var: all_domains
-  when: enable_debug
-
 - name: Filter www-prefixed domains from all_domains
   set_fact:
     www_domains: "{{ all_domains | select('match', '^www\\.') | list }}"
 
-- name: Include nginx-redirect-domain role for www-to-bare redirects
+- name: Include nginx-redirect-domains role for www-to-bare redirects
   include_role:
-    name: nginx-redirect-domain
+    name: nginx-redirect-domains
   vars:
     domain_mappings: "{{ www_domains
        | map('regex_replace',

tasks/constructor.yml

@@ -121,8 +121,6 @@
   include_role:
     name: client-wireguard
 
-
-
 ## backup setup
 - name: setup replica backup hosts
   when: ("backup_remote_to_local" in group_names)

tasks/server.yml

@@ -237,7 +237,7 @@
 - name: setup redirect hosts
   when: ("redirect" in group_names)
   include_role:
-    name: nginx-redirect-domain
+    name: nginx-redirect-domains
   vars:
     domain_mappings: "{{redirect_domain_mappings}}"
 

tests/unit/test_cert_utils.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+
+import os
+import sys
+
+# Add module_utils/ to the import path
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..", "module_utils")))
+
+from module_utils.cert_utils import CertUtils
+
+def test_matches():
+    tests = [
+        # Exact matches
+        ("example.com", "example.com", True),
+        ("www.example.com", "www.example.com", True),
+        ("api.example.com", "api.example.com", True),
+
+        # Wildcard matches
+        ("sub.example.com", "*.example.com", True),
+        ("www.example.com", "*.example.com", True),
+
+        # Wildcard non-matches
+        ("example.com", "*.example.com", False),  # base domain is not covered
+        ("deep.sub.example.com", "*.example.com", False),  # too deep
+        ("sub.deep.example.com", "*.deep.example.com", True),  # correct: one level below
+
+        # Special cases
+        ("deep.api.example.com", "*.api.example.com", True),
+        ("api.example.com", "*.api.example.com", False),  # base not covered by wildcard
+
+        # Completely different domains
+        ("test.other.com", "*.example.com", False),
+    ]
+
+    passed = 0
+    failed = 0
+
+    for domain, san, expected in tests:
+        result = CertUtils.matches(domain, san)
+        if result == expected:
+            print(f"✅ PASS: {domain} vs {san} -> {result}")
+            passed += 1
+        else:
+            print(f"❌ FAIL: {domain} vs {san} -> {result} (expected {expected})")
+            failed += 1
+
+    print(f"\nSummary: {passed} passed, {failed} failed")
+
+if __name__ == "__main__":
+    test_matches()