Added Smoke Test (Blackbox not reviewed) for create_credentials

cli/create_credentials.py

@@ -12,7 +12,9 @@ from yaml.dumper import SafeDumper
 
 def ask_for_confirmation(key: str) -> bool:
     """Prompt the user for confirmation to overwrite an existing value."""
-    confirmation = input(f"Are you sure you want to overwrite the value for '{key}'? (y/n): ").strip().lower()
+    confirmation = input(
+        f"Are you sure you want to overwrite the value for '{key}'? (y/n): "
+    ).strip().lower()
     return confirmation == 'y'
 
 
@@ -20,17 +22,31 @@ def main():
     parser = argparse.ArgumentParser(
         description="Selectively vault credentials + become-password in your inventory."
     )
-    parser.add_argument("--role-path", required=True, help="Path to your role")
-    parser.add_argument("--inventory-file", required=True, help="Host vars file to update")
-    parser.add_argument("--vault-password-file", required=True, help="Vault password file")
-    parser.add_argument("--set", nargs="*", default=[], help="Override values key.subkey=VALUE")
-    parser.add_argument("-f", "--force", action="store_true", help="Force overwrite without confirmation")
+    parser.add_argument(
+        "--role-path", required=True, help="Path to your role"
+    )
+    parser.add_argument(
+        "--inventory-file", required=True, help="Host vars file to update"
+    )
+    parser.add_argument(
+        "--vault-password-file", required=True, help="Vault password file"
+    )
+    parser.add_argument(
+        "--set", nargs="*", default=[], help="Override values key.subkey=VALUE"
+    )
+    parser.add_argument(
+        "-f", "--force", action="store_true",
+        help="Force overwrite without confirmation"
+    )
     args = parser.parse_args()
 
-    # Parsing overrides
-    overrides = {k.strip(): v.strip() for pair in args.set for k, v in [pair.split("=", 1)]}
+    # Parse overrides
+    overrides = {
+        k.strip(): v.strip()
+        for pair in args.set for k, v in [pair.split("=", 1)]
+    }
 
-    # Initialize the Inventory Manager
+    # Initialize inventory manager
     manager = InventoryManager(
         role_path=Path(args.role_path),
         inventory_path=Path(args.inventory_file),
@@ -38,34 +54,57 @@ def main():
         overrides=overrides
     )
 
-    # 1) Apply schema and update inventory
+    # Load existing credentials to preserve
+    existing_apps = manager.inventory.get("applications", {})
+    existing_creds = {}
+    if manager.app_id in existing_apps:
+        existing_creds = existing_apps[manager.app_id].get("credentials", {}).copy()
+
+    # Apply schema (may generate defaults)
     updated_inventory = manager.apply_schema()
 
-    # 2) Apply vault encryption ONLY to 'credentials' fields (we no longer apply it globally)
-    credentials = updated_inventory.get("applications", {}).get(manager.app_id, {}).get("credentials", {})
-    for key, value in credentials.items():
-        if not value.lstrip().startswith("$ANSIBLE_VAULT"):  # Only apply encryption if the value is not already vaulted
-            if key in credentials and not args.force:
-                if not ask_for_confirmation(key):  # Ask for confirmation before overwriting
-                    print(f"Skipping overwrite of '{key}'.")
-                    continue
-            encrypted_value = manager.vault_handler.encrypt_string(value, key)
-            lines = encrypted_value.splitlines()
-            indent = len(lines[1]) - len(lines[1].lstrip())
-            body = "\n".join(line[indent:] for line in lines[1:])
-            credentials[key] = VaultScalar(body)  # Store encrypted value as VaultScalar
+    # Restore existing database_password if present
+    apps = updated_inventory.setdefault("applications", {})
+    app_block = apps.setdefault(manager.app_id, {})
+    creds = app_block.setdefault("credentials", {})
+    if "database_password" in existing_creds:
+        creds["database_password"] = existing_creds["database_password"]
 
-    # 3) Vault top-level ansible_become_password if present
+    # Store original plaintext values
+    original_plain = {key: str(val) for key, val in creds.items()}
+
+    for key, raw_val in list(creds.items()):
+        # Skip if already vaulted
+        if isinstance(raw_val, VaultScalar) or str(raw_val).lstrip().startswith("$ANSIBLE_VAULT"):
+            continue
+
+        # Determine plaintext
+        plain = original_plain.get(key, "")
+        if key in overrides and (args.force or ask_for_confirmation(key)):
+            plain = overrides[key]
+
+        # Encrypt the plaintext
+        encrypted = manager.vault_handler.encrypt_string(plain, key)
+        lines = encrypted.splitlines()
+        indent = len(lines[1]) - len(lines[1].lstrip())
+        body = "\n".join(line[indent:] for line in lines[1:])
+        creds[key] = VaultScalar(body)
+
+    # Vault top-level become password if present
     if "ansible_become_password" in updated_inventory:
         val = str(updated_inventory["ansible_become_password"])
-        if not val.lstrip().startswith("$ANSIBLE_VAULT"):
-            snippet = manager.vault_handler.encrypt_string(val, "ansible_become_password")
+        if val.lstrip().startswith("$ANSIBLE_VAULT"):
+            updated_inventory["ansible_become_password"] = VaultScalar(val)
+        else:
+            snippet = manager.vault_handler.encrypt_string(
+                val, "ansible_become_password"
+            )
             lines = snippet.splitlines()
             indent = len(lines[1]) - len(lines[1].lstrip())
             body = "\n".join(line[indent:] for line in lines[1:])
             updated_inventory["ansible_become_password"] = VaultScalar(body)
 
-    # 4) Save the updated inventory to file
+    # Write back to file
     with open(args.inventory_file, "w", encoding="utf-8") as f:
         yaml.dump(updated_inventory, f, sort_keys=False, Dumper=SafeDumper)
 

roles/docker-gitea/tasks/main.yml

@@ -62,4 +62,14 @@
 
 - name: Execute Cleanup Routines
   include_tasks: cleanup.yml 
-  when: mode_cleanup
+  when: mode_cleanup
+
+- name: Include DNS role to register Gitea domain(s)
+  include_role:
+    name: dns-records-cloudflare
+  vars:
+    cloudflare_api_token:   "{{ certbot_dns_api_token }}"
+    cloudflare_domains:     "{{ [ domains | get_domain(application_id) ] }}"
+    cloudflare_target_ip:   "{{ networks.internet.ip4 }}"
+    cloudflare_proxied:     false
+  when: dns_provider == 'cloudflare'

roles/docker-nextcloud/Administration.md

@@ -1,5 +0,0 @@
-# Administration
-
-## Logs
-
-The logs you will find here on the host: **/var/lib/docker/volumes/nextcloud_data/_data/data/nextcloud.log**

roles/docker-nextcloud/docs/Administration.md

@@ -14,3 +14,7 @@ Inside the container, install a text editor and edit the config:
 ```bash
 apk add --no-cache nano && nano config/config.php
 ```
+
+## Logs
+
+The logs you will find here on the host: **/var/lib/docker/volumes/nextcloud_data/_data/data/nextcloud.log**

roles/docker-nextcloud/docs/Architecture.md

@@ -1,4 +0,0 @@
-# Administration
-
-## Other Resources
-- [Nextcloud Docker Example with Nginx Proxy, MariaDB, and FPM](https://github.com/nextcloud/docker/blob/master/.examples/docker-compose/with-nginx-proxy/mariadb/fpm/docker-compose.yml)

roles/docker-nextcloud/docs/LDAP.md

@@ -1,7 +1,3 @@
-Natürlich, hier ist der aktualisierte Abschnitt inklusive des allgemeinen LDAP-Synchronisationsbefehls:
-
----
-
 ## Add LDAP Users Manually for Immediate Sharing
 
 In a default Nextcloud + LDAP setup, user accounts are only created in the internal Nextcloud database **after their first login**. This means that even if a user exists in LDAP, they **cannot receive shared files or folders** until they have logged in at least once—or are manually synchronized.
@@ -42,8 +38,4 @@ To synchronize account data (display name, mail address, group memberships, etc.
 docker exec -u www-data nextcloud-application php occ user:sync-account-data
 ```
 
-This step is especially useful after modifying LDAP attributes or group memberships, ensuring up-to-date data in the Nextcloud UI and permission system.
-
----
-
-Let me know if you'd like a similar section for OIDC or automated sync in Ansible.
+This step is especially useful after modifying LDAP attributes or group memberships, ensuring up-to-date data in the Nextcloud UI and permission system.

roles/docker-nextcloud/meta/main.yml

@@ -32,3 +32,4 @@ galaxy_info:
     - docker-collabora
     - docker-keycloak
     - docker-mastodon
+    - docker-mariadb

roles/docker-nextcloud/vars/plugins/user_ldap.yml

@@ -163,7 +163,7 @@ plugin_configuration:
   -
     appid: "user_ldap"
     configkey: "s01ldap_userlist_filter"
-    configvalue: "{{ ldap.filters.users.login }}"
+    configvalue: "{{ ldap.filters.users.all }}"
   -
     appid: "user_ldap"
     configkey: "s01use_memberof_to_detect_membership"

tests/unit/test_create_credentials.py

@@ -0,0 +1,101 @@
+import os
+import sys
+import unittest
+from unittest import mock
+
+# Ensure cli module is importable
+dir_path = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), '../../cli')
+)
+sys.path.insert(0, dir_path)
+
+# Import functions and classes to test
+from create_credentials import ask_for_confirmation, main
+from utils.handler.vault import VaultHandler, VaultScalar
+import subprocess
+import tempfile
+import yaml
+
+class TestCreateCredentials(unittest.TestCase):
+    def test_ask_for_confirmation_yes(self):
+        with mock.patch('builtins.input', return_value='y'):
+            self.assertTrue(ask_for_confirmation('test_key'))
+
+    def test_ask_for_confirmation_no(self):
+        with mock.patch('builtins.input', return_value='n'):
+            self.assertFalse(ask_for_confirmation('test_key'))
+
+    def test_vault_encrypt_string_success(self):
+        handler = VaultHandler('dummy_pw_file')
+        # Mock subprocess.run to simulate successful vault encryption
+        fake_output = 'Encrypted data'
+        completed = subprocess.CompletedProcess(
+            args=['ansible-vault'], returncode=0, stdout=fake_output, stderr=''
+        )
+        with mock.patch('subprocess.run', return_value=completed) as proc_run:
+            result = handler.encrypt_string('plain_val', 'name')
+            proc_run.assert_called_once()
+            self.assertEqual(result, fake_output)
+
+    def test_vault_encrypt_string_failure(self):
+        handler = VaultHandler('dummy_pw_file')
+        # Mock subprocess.run to simulate failure
+        completed = subprocess.CompletedProcess(
+            args=['ansible-vault'], returncode=1, stdout='', stderr='error')
+        with mock.patch('subprocess.run', return_value=completed):
+            with self.assertRaises(RuntimeError):
+                handler.encrypt_string('plain_val', 'name')
+
+    def test_main_overrides_and_file_writing(self):
+        # Setup temporary files for role-path vars and inventory
+        with tempfile.TemporaryDirectory() as tmpdir:
+            role_path = os.path.join(tmpdir, 'role')
+            os.makedirs(os.path.join(role_path, 'vars'))
+            os.makedirs(os.path.join(role_path, 'meta'))
+            # Create vars/main.yml with application_id
+            main_vars = {'application_id': 'app_test'}
+            with open(os.path.join(role_path, 'vars', 'main.yml'), 'w') as f:
+                yaml.dump(main_vars, f)
+            # Create vars/configuration.yml with features disabled
+            config = {'features': {'central_database': False}}
+            with open(os.path.join(role_path, 'vars', 'configuration.yml'), 'w') as f:
+                yaml.dump(config, f)
+            # Create schema.yml defining plain credential
+            schema = {'credentials': {'api_key': {'description': 'API key', 'algorithm': 'plain', 'validation': {}}}}
+            with open(os.path.join(role_path, 'meta', 'schema.yml'), 'w') as f:
+                yaml.dump(schema, f)
+            # Prepare inventory file
+            inventory_file = os.path.join(tmpdir, 'inventory.yml')
+            with open(inventory_file, 'w') as f:
+                yaml.dump({}, f)
+            vault_pw_file = os.path.join(tmpdir, 'pw.txt')
+            with open(vault_pw_file, 'w') as f:
+                f.write('pw')
+
+            # Simulate ansible-vault encrypt_string output for api_key
+            fake_snippet = "!vault |\n  $ANSIBLE_VAULT;1.1;AES256\n    ENCRYPTEDVALUE"
+            completed = subprocess.CompletedProcess(
+                args=['ansible-vault'], returncode=0, stdout=fake_snippet, stderr=''
+            )
+            with mock.patch('subprocess.run', return_value=completed):
+                # Run main with override for credentials.api_key and force to skip prompt
+                sys.argv = [
+                    'create_credentials.py',
+                    '--role-path', role_path,
+                    '--inventory-file', inventory_file,
+                    '--vault-password-file', vault_pw_file,
+                    '--set', 'credentials.api_key=SECRET',
+                    '--force'
+                ]
+                # Should complete without error
+                main()
+                # Verify inventory file updated with vaulted api_key
+                data = yaml.safe_load(open(inventory_file))
+                creds = data['applications']['app_test']['credentials']
+                self.assertIn('api_key', creds)
+                # VaultScalar serializes to a vault block, safe_load returns a string containing the vault header
+                self.assertIsInstance(creds['api_key'], str)
+                self.assertTrue(creds['api_key'].lstrip().startswith('$ANSIBLE_VAULT'))
+
+if __name__ == '__main__':
+    unittest.main()

tests/unit/test_generate_vaulted_credentials.py

@@ -1,77 +0,0 @@
-import pytest
-import sys, os
-from pathlib import Path
-
-sys.path.insert(
-    0,
-    os.path.abspath(
-        os.path.join(os.path.dirname(__file__), "../../cli")
-    ),
-)
-
-# 2) Import from the cli package
-import cli.create_credentials as gvc
-
-class DummyProc:
-    def __init__(self, returncode, stdout, stderr=''):
-        self.returncode = returncode
-        self.stdout = stdout
-        self.stderr = stderr
-
-# Monkeypatch subprocess.run for encrypt_with_vault
-@pytest.fixture(autouse=True)
-def mock_subprocess_run(monkeypatch):
-    def fake_run(cmd, capture_output, text):
-        name = None
-        # find --name=<key> in args
-        for arg in cmd:
-            if arg.startswith("--name="):
-                name = arg.split("=",1)[1]
-        val = cmd[ cmd.index(name) - 1 ] if name else "key"
-        # simulate Ansible output
-        snippet = f"{name or 'key'}: !vault |\n    encrypted_{val}"
-        return DummyProc(0, snippet)
-    monkeypatch.setattr(gvc.subprocess, 'run', fake_run)
-
-def test_wrap_existing_vaults():
-    data = {
-        'a': '$ANSIBLE_VAULT;1.1;AES256...blob',
-        'b': {'c': 'normal', 'd': '$ANSIBLE_VAULT;1.1;AES256...other'},
-        'e': ['x', '$ANSIBLE_VAULT;1.1;AES256...list']
-    }
-    wrapped = gvc.wrap_existing_vaults(data)
-    assert isinstance(wrapped['a'], gvc.VaultScalar)
-    assert isinstance(wrapped['b']['d'], gvc.VaultScalar)
-    assert isinstance(wrapped['e'][1], gvc.VaultScalar)
-    assert wrapped['b']['c'] == 'normal'
-    assert wrapped['e'][0] == 'x'
-
-@pytest.mark.parametrize("pairs,expected", [
-    (['k=v'], {'k': 'v'}),
-    (['a.b=1', 'c=two'], {'a.b': '1', 'c': 'two'}),
-    (['noeq'], {}),
-])
-def test_parse_overrides(pairs, expected):
-    assert gvc.parse_overrides(pairs) == expected
-
-def test_apply_schema_and_vault(tmp_path):
-    schema = {
-        'cred': {'description':'d','algorithm':'plain','validation':{}},
-        'nested': {'inner': {'description':'d2','algorithm':'plain','validation':{}}}
-    }
-    inv = {}
-    updated = gvc.apply_schema(schema, inv, 'app', {}, 'pwfile')
-    apps = updated['applications']['app']
-    assert isinstance(apps['cred'], gvc.VaultScalar)
-    assert isinstance(apps['nested']['inner'], gvc.VaultScalar)
-
-def test_encrypt_leaves_and_credentials():
-    branch = {'p':'v','nested':{'q':'u'}}
-    gvc.encrypt_leaves(branch, 'pwfile')
-    assert isinstance(branch['p'], gvc.VaultScalar)
-    assert isinstance(branch['nested']['q'], gvc.VaultScalar)
-
-    inv = {'credentials':{'a':'b'}, 'x':{'credentials':{'c':'d'}}}
-    gvc.encrypt_credentials_branch(inv, 'pwfile')
-    assert isinstance(inv['credentials']['a'], gvc.VaultScalar)
-    assert isinstance(inv['x']['credentials']['c'], gvc.VaultScalar)