cli/TODO.md

@@ -1,3 +1,2 @@
 # Todo
 - Test this script. It's just a draft. Checkout https://chatgpt.com/c/681d9e2b-7b28-800f-aef8-4f1427e9021d
-- Solve bugs in show_vault_variables.py

cli/create_credentials.py

@@ -1,76 +1,205 @@
+#!/usr/bin/env python3
 import argparse
+import secrets
+import hashlib
+import bcrypt
 import subprocess
 import sys
 from pathlib import Path
+
 import yaml
-from typing import Dict, Any
-from utils.manager.inventory import InventoryManager
-from utils.handler.vault import VaultHandler, VaultScalar
-from utils.handler.yaml import YamlHandler
+from yaml.loader import SafeLoader
 from yaml.dumper import SafeDumper
 
+# ─────────────────────────────────────────────────────────────────────────────
+# On load: treat any !vault tag as plain text
+def _vault_constructor(loader, node):
+    return node.value
+SafeLoader.add_constructor('!vault', _vault_constructor)
 
-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()
-    return confirmation == 'y'
+# A str subclass so PyYAML emits !vault literal blocks on dump
+class VaultScalar(str):
+    pass
 
+def _vault_representer(dumper, data):
+    return dumper.represent_scalar('!vault', data, style='|')
+
+SafeDumper.add_representer(VaultScalar, _vault_representer)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def generate_value(algorithm: str) -> str:
+    if algorithm == "random_hex":
+        return secrets.token_hex(64)
+    if algorithm == "sha256":
+        return hashlib.sha256(secrets.token_bytes(32)).hexdigest()
+    if algorithm == "sha1":
+        return hashlib.sha1(secrets.token_bytes(20)).hexdigest()
+    if algorithm == "bcrypt":
+        pw = secrets.token_urlsafe(16).encode()
+        return bcrypt.hashpw(pw, bcrypt.gensalt()).decode()
+    # we should never auto-generate for "plain"
+    return "undefined"
+
+def wrap_existing_vaults(node):
+    """
+    Recursively wrap any str that begins with '$ANSIBLE_VAULT'
+    in a VaultScalar so it dumps as a literal block.
+    """
+    if isinstance(node, dict):
+        return {k: wrap_existing_vaults(v) for k, v in node.items()}
+    if isinstance(node, list):
+        return [wrap_existing_vaults(v) for v in node]
+    if isinstance(node, str) and node.lstrip().startswith("$ANSIBLE_VAULT"):
+        return VaultScalar(node)
+    return node
+
+def load_yaml_plain(path: Path) -> dict:
+    """
+    Load any YAML (vaulted or not) via SafeLoader + our !vault constructor,
+    then wrap existing vault‐blocks for correct literal dumping.
+    """
+    text = path.read_text()
+    data = yaml.load(text, Loader=SafeLoader) or {}
+    return wrap_existing_vaults(data)
+
+def encrypt_with_vault(value: str, name: str, vault_password_file: str) -> str:
+    cmd = [
+        "ansible-vault", "encrypt_string",
+        value, f"--name={name}",
+        "--vault-password-file", vault_password_file
+    ]
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise RuntimeError(f"ansible-vault encrypt_string failed:\n{proc.stderr}")
+    return proc.stdout
+
+def parse_overrides(pairs: list[str]) -> dict:
+    out = {}
+    for p in pairs:
+        if "=" in p:
+            k, v = p.split("=", 1)
+            out[k.strip()] = v.strip()
+    return out
+
+def load_application_id(role_path: Path) -> str:
+    vars_file = role_path / "vars" / "main.yml"
+    data = load_yaml_plain(vars_file)
+    app_id = data.get("application_id")
+    if not app_id:
+        print(f"ERROR: 'application_id' missing in {vars_file}", file=sys.stderr)
+        sys.exit(1)
+    return app_id
+
+def apply_schema(schema: dict,
+                 inventory: dict,
+                 app_id: str,
+                 overrides: dict,
+                 vault_pw: str) -> dict:
+    apps = inventory.setdefault("applications", {})
+    target = apps.setdefault(app_id, {})
+
+    def recurse(branch: dict, dest: dict, prefix: str = ""):
+        for key, meta in branch.items():
+            full_key = f"{prefix}.{key}" if prefix else key
+
+            # leaf node spec
+            if isinstance(meta, dict) and all(k in meta for k in ("description","algorithm","validation")):
+                alg = meta["algorithm"]
+                if alg == "plain":
+                    # must be supplied via --set
+                    if full_key not in overrides:
+                        print(f"ERROR: Plain algorithm for '{full_key}' requires override via --set {full_key}=<value>", file=sys.stderr)
+                        sys.exit(1)
+                    plain = overrides[full_key]
+                else:
+                    # generate or override
+                    plain = overrides.get(full_key, generate_value(alg))
+
+                snippet = encrypt_with_vault(plain, key, vault_pw)
+                lines = snippet.splitlines()
+                indent = len(lines[1]) - len(lines[1].lstrip())
+                body = "\n".join(line[indent:] for line in lines[1:])
+                dest[key] = VaultScalar(body)
+
+            # nested mapping
+            elif isinstance(meta, dict):
+                sub = dest.setdefault(key, {})
+                recurse(meta, sub, full_key)
+
+            # literal passthrough
+            else:
+                dest[key] = meta
+
+    recurse(schema, target)
+    return inventory
+
+def encrypt_leaves(branch: dict, vault_pw: str):
+    for k, v in list(branch.items()):
+        if isinstance(v, dict):
+            encrypt_leaves(v, vault_pw)
+        else:
+            plain = str(v)
+            # skip if already vaulted
+            if plain.lstrip().startswith("$ANSIBLE_VAULT"):
+                continue
+            snippet = encrypt_with_vault(plain, k, vault_pw)
+            lines = snippet.splitlines()
+            indent = len(lines[1]) - len(lines[1].lstrip())
+            body = "\n".join(line[indent:] for line in lines[1:])
+            branch[k] = VaultScalar(body)
+
+def encrypt_credentials_branch(node, vault_pw: str):
+    if isinstance(node, dict):
+        for key, val in node.items():
+            if key == "credentials" and isinstance(val, dict):
+                encrypt_leaves(val, vault_pw)
+            else:
+                encrypt_credentials_branch(val, vault_pw)
+    elif isinstance(node, list):
+        for item in node:
+            encrypt_credentials_branch(item, vault_pw)
 
 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("--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)]}
+    role_path = Path(args.role_path)
+    inv_file   = Path(args.inventory_file)
+    vault_pw   = args.vault_password_file
+    overrides  = parse_overrides(args.set)
 
-    # Initialize the Inventory Manager
-    manager = InventoryManager(
-        role_path=Path(args.role_path),
-        inventory_path=Path(args.inventory_file),
-        vault_pw=args.vault_password_file,
-        overrides=overrides
-    )
+    # 1) Load & wrap any existing vault blocks
+    inventory = load_yaml_plain(inv_file)
 
-    # 1) Apply schema and update inventory
-    updated_inventory = manager.apply_schema()
+    # 2) Merge schema-driven credentials (plain ones must be overridden)
+    schema = load_yaml_plain(role_path / "meta" / "schema.yml")
+    app_id = load_application_id(role_path)
+    inventory = apply_schema(schema, inventory, app_id, overrides, vault_pw)
 
-    # 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
+    # 3) Vault any leaves under 'credentials:' mappings
+    encrypt_credentials_branch(inventory, vault_pw)
 
-    # 3) Vault top-level ansible_become_password if present
-    if "ansible_become_password" in updated_inventory:
-        val = str(updated_inventory["ansible_become_password"])
+    # 4) Vault top-level ansible_become_password if present
+    if "ansible_become_password" in inventory:
+        val = str(inventory["ansible_become_password"])
         if not val.lstrip().startswith("$ANSIBLE_VAULT"):
-            snippet = manager.vault_handler.encrypt_string(val, "ansible_become_password")
+            snippet = encrypt_with_vault(val, "ansible_become_password", vault_pw)
             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)
+            inventory["ansible_become_password"] = VaultScalar(body)
 
-    # 4) Save the updated inventory to file
-    with open(args.inventory_file, "w", encoding="utf-8") as f:
-        yaml.dump(updated_inventory, f, sort_keys=False, Dumper=SafeDumper)
-
-    print(f"✅ Inventory selectively vaulted → {args.inventory_file}")
+    # 5) Overwrite file with proper !vault literal blocks only where needed
+    with open(inv_file, "w", encoding="utf-8") as f:
+        yaml.dump(inventory, f, sort_keys=False, Dumper=SafeDumper)
 
+    print(f"✅ Inventory selectively vaulted → {inv_file}")
 
 if __name__ == "__main__":
     main()

cli/encrypt_inventory.py

@@ -1,66 +0,0 @@
-import argparse
-import subprocess
-import sys
-from pathlib import Path
-import yaml
-from typing import Dict, Any
-from utils.handler.vault import VaultHandler, VaultScalar
-from utils.handler.yaml import YamlHandler
-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"Do you want to encrypt the value for '{key}'? (y/n): ").strip().lower()
-    return confirmation == 'y'
-
-
-def encrypt_recursively(data: Any, vault_handler: VaultHandler, ask_confirmation: bool = True, prefix: str = "") -> Any:
-    """Recursively encrypt values in the data."""
-    if isinstance(data, dict):
-        for key, value in data.items():
-            new_prefix = f"{prefix}.{key}" if prefix else key
-            data[key] = encrypt_recursively(value, vault_handler, ask_confirmation, new_prefix)
-    elif isinstance(data, list):
-        for i, item in enumerate(data):
-            data[i] = encrypt_recursively(item, vault_handler, ask_confirmation, prefix)
-    elif isinstance(data, str):
-        # Only encrypt if it's not already vaulted
-        if not data.lstrip().startswith("$ANSIBLE_VAULT"):
-            if ask_confirmation:
-                # Ask for confirmation before encrypting if not `--all`
-                if not ask_for_confirmation(prefix):
-                    print(f"Skipping encryption for '{prefix}'.")
-                    return data
-            encrypted_value = vault_handler.encrypt_string(data, prefix)
-            lines = encrypted_value.splitlines()
-            indent = len(lines[1]) - len(lines[1].lstrip())
-            body = "\n".join(line[indent:] for line in lines[1:])
-            return VaultScalar(body)  # Store encrypted value as VaultScalar
-    return data
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Encrypt all fields, ask for confirmation unless --all is specified."
-    )
-    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("--all", action="store_true", help="Encrypt all fields without confirmation")
-    args = parser.parse_args()
-
-    # Initialize the VaultHandler and load the inventory
-    vault_handler = VaultHandler(vault_password_file=args.vault_password_file)
-    updated_inventory = YamlHandler.load_yaml(Path(args.inventory_file))
-
-    # 1) Encrypt all fields recursively
-    updated_inventory = encrypt_recursively(updated_inventory, vault_handler, ask_confirmation=not args.all)
-
-    # 2) Save the updated inventory to file
-    with open(args.inventory_file, "w", encoding="utf-8") as f:
-        yaml.dump(updated_inventory, f, sort_keys=False, Dumper=SafeDumper)
-
-    print(f"✅ Inventory selectively vaulted → {args.inventory_file}")
-
-
-if __name__ == "__main__":
-    main()

cli/show_vault_variables_draft.py

@@ -1,105 +0,0 @@
-import argparse
-import subprocess
-from ansible.parsing.vault import VaultLib, VaultSecret
-import sys
-import yaml
-import re
-from utils.handler.vault import VaultScalar
-from yaml.loader import SafeLoader
-from yaml.dumper import SafeDumper
-
-# Register the custom constructor and representer for VaultScalar in PyYAML
-SafeLoader.add_constructor('!vault', lambda loader, node: VaultScalar(node.value))
-SafeDumper.add_representer(VaultScalar, lambda dumper, data: dumper.represent_scalar('!vault', data))
-
-def is_vault_encrypted_data(data: str) -> bool:
-    """Check if the given data is encrypted with Ansible Vault by looking for the vault header."""
-    return data.lstrip().startswith('$ANSIBLE_VAULT')
-
-def decrypt_vault_data(encrypted_data: str, vault_secret: VaultSecret) -> str:
-    """
-    Decrypt the given encrypted data using the provided vault_secret.
-    :param encrypted_data: Encrypted string to be decrypted
-    :param vault_secret: The VaultSecret instance used to decrypt the data
-    :return: Decrypted data as a string
-    """
-    vault = VaultLib()
-    decrypted_data = vault.decrypt(encrypted_data, vault_secret)
-    return decrypted_data
-
-def decrypt_vault_file(vault_file: str, vault_password_file: str):
-    """
-    Decrypt the Ansible Vault file and return its contents.
-    :param vault_file: Path to the encrypted Ansible Vault file
-    :param vault_password_file: Path to the file containing the Vault password
-    :return: Decrypted contents of the Vault file
-    """
-    # Read the vault password
-    with open(vault_password_file, 'r') as f:
-        vault_password = f.read().strip()
-
-    # Create a VaultSecret instance from the password
-    vault_secret = VaultSecret(vault_password.encode())
-
-    # Read the encrypted file
-    with open(vault_file, 'r') as f:
-        file_content = f.read()
-
-    # If the file is partially encrypted, we'll decrypt only the encrypted values
-    decrypted_data = file_content  # Start with the unmodified content
-
-    # Find all vault-encrypted values (i.e., values starting with $ANSIBLE_VAULT)
-    encrypted_values = re.findall(r'^\s*([\w\.\-_]+):\s*["\']?\$ANSIBLE_VAULT[^\n]+', file_content, flags=re.MULTILINE)
-
-    # If there are encrypted values, decrypt them
-    for value in encrypted_values:
-        # Extract the encrypted value and decrypt it
-        encrypted_value = re.search(r'(["\']?\$ANSIBLE_VAULT[^\n]+)', value)
-        if encrypted_value:
-            # Remove any newlines or extra spaces from the encrypted value
-            encrypted_value = encrypted_value.group(0).replace('\n', '').replace('\r', '')
-            decrypted_value = decrypt_vault_data(encrypted_value, vault_secret)
-            # Replace the encrypted value with the decrypted value in the content
-            decrypted_data = decrypted_data.replace(encrypted_value, decrypted_value.strip())
-
-    return decrypted_data
-
-def decrypt_and_display(vault_file: str, vault_password_file: str):
-    """
-    Decrypts the Ansible Vault file and its values, then display the result.
-    Supports both full file and partial value encryption.
-    :param vault_file: Path to the encrypted Ansible Vault file
-    :param vault_password_file: Path to the file containing the Vault password
-    """
-    decrypted_data = decrypt_vault_file(vault_file, vault_password_file)
-
-    # Convert the decrypted data to a string format (YAML or JSON)
-    output_data = yaml.dump(yaml.safe_load(decrypted_data), default_flow_style=False)
-
-    # Use subprocess to call `less` for paginated, scrollable output
-    subprocess.run(["less"], input=output_data, text=True)
-
-def main():
-    # Set up the argument parser
-    parser = argparse.ArgumentParser(description="Decrypt and display variables from an Ansible Vault file.")
-    
-    # Add arguments for the vault file and vault password file
-    parser.add_argument(
-        'vault_file', 
-        type=str, 
-        help="Path to the encrypted Ansible Vault file"
-    )
-    parser.add_argument(
-        'vault_password_file', 
-        type=str, 
-        help="Path to the file containing the Vault password"
-    )
-    
-    # Parse the arguments
-    args = parser.parse_args()
-
-    # Display vault variables in a scrollable manner
-    decrypt_and_display(args.vault_file, args.vault_password_file)
-
-if __name__ == "__main__":
-    main()

cli/utils/handler/vault.py

@@ -1,50 +0,0 @@
-import subprocess
-from typing import Any, Dict
-
-from yaml.loader import SafeLoader
-from yaml.dumper import SafeDumper
-
-class VaultScalar(str):
-    """A subclass of str to represent vault-encrypted strings."""
-    pass
-
-def _vault_constructor(loader, node):
-    """Custom constructor to handle !vault tag as plain text."""
-    return node.value
-
-def _vault_representer(dumper, data):
-    """Custom representer to dump VaultScalar as literal blocks."""
-    return dumper.represent_scalar('!vault', data, style='|')
-
-SafeLoader.add_constructor('!vault', _vault_constructor)
-SafeDumper.add_representer(VaultScalar, _vault_representer)
-
-class VaultHandler:
-    def __init__(self, vault_password_file: str):
-        self.vault_password_file = vault_password_file
-
-    def encrypt_string(self, value: str, name: str) -> str:
-        """Encrypt a string using ansible-vault."""
-        cmd = [
-            "ansible-vault", "encrypt_string",
-            value, f"--name={name}",
-            "--vault-password-file", self.vault_password_file
-        ]
-        proc = subprocess.run(cmd, capture_output=True, text=True)
-        if proc.returncode != 0:
-            raise RuntimeError(f"ansible-vault encrypt_string failed:\n{proc.stderr}")
-        return proc.stdout
-
-    def encrypt_leaves(self, branch: Dict[str, Any], vault_pw: str):
-        """Recursively encrypt all leaves (plain text values) under the credentials section."""
-        for key, value in branch.items():
-            if isinstance(value, dict):
-                self.encrypt_leaves(value, vault_pw)  # Recurse into nested dictionaries
-            else:
-                # Skip if already vaulted (i.e., starts with $ANSIBLE_VAULT)
-                if isinstance(value, str) and not value.lstrip().startswith("$ANSIBLE_VAULT"):
-                    snippet = self.encrypt_string(value, key)
-                    lines = snippet.splitlines()
-                    indent = len(lines[1]) - len(lines[1].lstrip())
-                    body = "\n".join(line[indent:] for line in lines[1:])
-                    branch[key] = VaultScalar(body)  # Store encrypted value as VaultScalar

cli/utils/handler/yaml.py

@@ -1,23 +0,0 @@
-import yaml
-from yaml.loader import SafeLoader
-from typing import Any, Dict
-from utils.handler.vault import VaultScalar
-
-class YamlHandler:
-    @staticmethod
-    def load_yaml(path) -> Dict:
-        """Load the YAML file and wrap existing !vault entries."""
-        text = path.read_text()
-        data = yaml.load(text, Loader=SafeLoader) or {}
-        return YamlHandler.wrap_existing_vaults(data)
-
-    @staticmethod
-    def wrap_existing_vaults(node: Any) -> Any:
-        """Recursively wrap any str that begins with '$ANSIBLE_VAULT' in a VaultScalar so it dumps as a literal block."""
-        if isinstance(node, dict):
-            return {k: YamlHandler.wrap_existing_vaults(v) for k, v in node.items()}
-        if isinstance(node, list):
-            return [YamlHandler.wrap_existing_vaults(v) for v in node]
-        if isinstance(node, str) and node.lstrip().startswith("$ANSIBLE_VAULT"):
-            return VaultScalar(node)
-        return node

cli/utils/manager/inventory.py

@@ -1,115 +0,0 @@
-import secrets
-import hashlib
-import bcrypt
-from pathlib import Path
-from typing import Dict
-from utils.handler.yaml import YamlHandler
-from utils.handler.vault import VaultHandler, VaultScalar
-import string
-
-class InventoryManager:
-    def __init__(self, role_path: Path, inventory_path: Path, vault_pw: str, overrides: Dict[str, str]):
-        """Initialize the Inventory Manager."""
-        self.role_path = role_path
-        self.inventory_path = inventory_path
-        self.vault_pw = vault_pw
-        self.overrides = overrides
-        self.inventory = YamlHandler.load_yaml(inventory_path)
-        self.schema = YamlHandler.load_yaml(role_path / "meta" / "schema.yml")
-        self.app_id = self.load_application_id(role_path)
-
-        self.vault_handler = VaultHandler(vault_pw)
-
-    def load_application_id(self, role_path: Path) -> str:
-        """Load the application ID from the role's vars/main.yml file."""
-        vars_file = role_path / "vars" / "main.yml"
-        data = YamlHandler.load_yaml(vars_file)
-        app_id = data.get("application_id")
-        if not app_id:
-            print(f"ERROR: 'application_id' missing in {vars_file}", file=sys.stderr)
-            sys.exit(1)
-        return app_id
-
-    def apply_schema(self) -> Dict:
-        """Apply the schema and return the updated inventory."""
-        apps = self.inventory.setdefault("applications", {})
-        target = apps.setdefault(self.app_id, {})
-
-        # Load the data from vars/main.yml
-        vars_file = self.role_path / "vars" / "configuration.yml"
-        data = YamlHandler.load_yaml(vars_file)
-
-        # Check if 'central-database' is enabled in the features section of data
-        if "features" in data and \
-        "central_database" in data["features"] and \
-        data["features"]["central_database"]:
-            # Add 'central_database' value (password) to credentials
-            target.setdefault("credentials", {})["database_password"] = self.generate_value("alphanumeric")
-
-        # Apply recursion only for the `credentials` section
-        self.recurse_credentials(self.schema, target)
-        return self.inventory
-
-    def recurse_credentials(self, branch: dict, dest: dict, prefix: str = ""):
-        """Recursively process only the 'credentials' section and generate values."""
-        for key, meta in branch.items():
-            full_key = f"{prefix}.{key}" if prefix else key
-        
-            # Only process 'credentials' section for encryption
-            if prefix == "credentials" and isinstance(meta, dict) and all(k in meta for k in ("description", "algorithm", "validation")):
-                alg = meta["algorithm"]
-                if alg == "plain":
-                    # Must be supplied via --set
-                    if full_key not in self.overrides:
-                        print(f"ERROR: Plain algorithm for '{full_key}' requires override via --set {full_key}=<value>", file=sys.stderr)
-                        sys.exit(1)
-                    plain = self.overrides[full_key]
-                else:
-                    plain = self.overrides.get(full_key, self.generate_value(alg))
-        
-                # Check if the value is already vaulted or if it's a dictionary
-                existing_value = dest.get(key)
-                
-                # If existing_value is a dictionary, print a warning and skip encryption
-                if isinstance(existing_value, dict):
-                    print(f"Skipping encryption for '{key}', as it is a dictionary.")
-                    continue
-                
-                # Check if the value is a VaultScalar and already vaulted
-                if existing_value and isinstance(existing_value, VaultScalar):
-                    print(f"Skipping encryption for '{key}', as it is already vaulted.")
-                    continue
-                
-                # Encrypt only if it's not already vaulted
-                snippet = self.vault_handler.encrypt_string(plain, key)
-                lines = snippet.splitlines()
-                indent = len(lines[1]) - len(lines[1].lstrip())
-                body = "\n".join(line[indent:] for line in lines[1:])
-                dest[key] = VaultScalar(body)
-        
-            elif isinstance(meta, dict):
-                sub = dest.setdefault(key, {})
-                self.recurse_credentials(meta, sub, full_key)
-            else:
-                dest[key] = meta
-
-
-    def generate_secure_alphanumeric(self, length: int) -> str:
-        """Generate a cryptographically secure random alphanumeric string of the given length."""
-        characters = string.ascii_letters + string.digits  # a-zA-Z0-9
-        return ''.join(secrets.choice(characters) for _ in range(length))
-
-    def generate_value(self, algorithm: str) -> str:
-        """Generate a value based on the provided algorithm."""
-        if algorithm == "random_hex":
-            return secrets.token_hex(64)
-        if algorithm == "sha256":
-            return hashlib.sha256(secrets.token_bytes(32)).hexdigest()
-        if algorithm == "sha1":
-            return hashlib.sha1(secrets.token_bytes(20)).hexdigest()
-        if algorithm == "bcrypt":
-            pw = secrets.token_urlsafe(16).encode()
-            return bcrypt.hashpw(pw, bcrypt.gensalt()).decode()
-        if algorithm == "alphanumeric":
-            return self.generate_secure_alphanumeric(64)
-        return "undefined"

filter_plugins/csp_filters.py

@@ -110,7 +110,7 @@ class FilterModule(object):
                     self.is_feature_enabled(applications, matomo_feature_name, application_id)
                     and directive in ['script-src', 'connect-src']
                 ):
-                    matomo_domain = domains.get('matomo')[0]
+                    matomo_domain = domains.get('matomo')
                     if matomo_domain:
                         tokens.append(f"{web_protocol}://{matomo_domain}")
 

group_vars/all/15_about.yml

@@ -9,12 +9,12 @@ defaults_service_provider:
       city:         "Cybertown"
       postal_code:  "00001"
       country:      "Nexusland"
-    logo:           "{{ applications['assets-server'].url ~ '/img/logo.png' }}"
+    logo:           "{{ applications['assets-server'].url | safe_var | safe_join('img/logo.png') }}"
   platform:    
     titel:          "CyMaIS Demo"
     subtitel:       "The Future of Self-Hosted Infrastructure. Secure. Automated. Sovereign."
-    logo:           "{{ applications['assets-server'].url ~ '/img/logo.png' }}"
-    favicon:        "{{ applications['assets-server'].url ~ '/img/favicon.ico' }}"
+    logo:           "{{ applications['assets-server'].url | safe_var | safe_join('img/logo.png') }}"
+    favicon:        "{{ applications['assets-server'].url | safe_var | safe_join('img/favicon.ico') }}"
   contact:
     bluesky: >-
       {{ ('@' ~ users.administrator.username ~ '.' ~ domains.bluesky.api)
@@ -30,4 +30,4 @@ defaults_service_provider:
   legal:
     editorial_responsible:  "Johannes Gutenberg"
     source_code:            "https://github.com/kevinveenbirkenbach/cymais"
-    imprint: "{{web_protocol}}://{{ domains | get_domain('html-server') }}/imprint.html"
+    imprint: "{{web_protocol}}://{{domains['html-server']}}/imprint.html"

roles/docker-akaunting/meta/schema.yml

@@ -1,4 +1,8 @@
 credentials:
+  database_password:
+    description: "Database password for MariaDB"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
   setup_admin_password:
     description: "Initial admin user password for Akaunting"
     algorithm: "sha256"

roles/docker-akaunting/vars/configuration.yml

@@ -8,6 +8,8 @@ features:
   portfolio_iframe:       false
   central_database:       true
 credentials:
+#  database_password:     Needs to be defined in inventory file 
+#  setup_admin_password:  Needs to be defined in inventory file
 domains:
   canonical:
     - "accounting.{{ primary_domain }}"

roles/docker-attendize/meta/schema.yml

@@ -1 +1,5 @@
 credentials:
+  database_password:
+    description: "Database password for MariaDB used by Attendize"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"

roles/docker-baserow/meta/schema.yml

@@ -0,0 +1,5 @@
+credentials:
+  database_password:
+    description: "Password for the PostgreSQL database used by Baserow"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"

roles/docker-discourse/meta/schema.yml

@@ -0,0 +1,5 @@
+credentials:
+  database_password:
+    description: "Password for the Discourse PostgreSQL database"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"

roles/docker-friendica/meta/schema.yml

@@ -0,0 +1,5 @@
+credentials:
+  database_password:
+    description: "Password for the Friendica database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"

roles/docker-funkwhale/meta/schema.yml

@@ -1,4 +1,8 @@
 credentials:
+  database_password:
+    description: "Password for the Funkwhale PostgreSQL database"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
   django_secret:
     description: "Django SECRET_KEY used for cryptographic signing"
     algorithm: "sha256"

roles/docker-funkwhale/vars/configuration.yml

@@ -2,7 +2,7 @@ version:              "1.4.0"
 features:
   matomo:             true
   css:                true
-  portfolio_iframe:   true
+  portfolio_iframe: true
   ldap:               true
   central_database:   true
 credentials:

roles/docker-gitea/meta/schema.yml

@@ -1 +1,5 @@
-
+credentials:
+  database_password:
+    description: "Password for the Gitea database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"

roles/docker-gitlab/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the GitLab PostgreSQL database"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   initial_root_password:
     description: "Initial password for the GitLab root user"
     algorithm: "sha256"

roles/docker-joomla/meta/schema.yml

@@ -0,0 +1,5 @@
+credentials:
+  database_password:
+    description: "Password for the Joomla database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"

roles/docker-keycloak/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the Keycloak PostgreSQL database"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   administrator_password:
     description: "Password for the Keycloak administrator user (used in bootstrap and CLI access)"
     algorithm: "sha256"

roles/docker-listmonk/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the Listmonk PostgreSQL database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   administrator_password:
     description: "Initial password for the Listmonk administrator account"
     algorithm: "sha256"

roles/docker-mastodon/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the Mastodon PostgreSQL database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   secret_key_base:
     description: "Main secret key used to verify the integrity of signed cookies and tokens"
     algorithm: "sha256"

roles/docker-matomo/meta/schema.yml

@@ -1,4 +1,8 @@
 credentials:
+  database_password:
+    description: "Password for the Matomo database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
 
   auth_token:
     description: "Authentication token for the Matomo HTTP API (used for automation and integrations)"

roles/docker-moodle/meta/schema.yml

@@ -1,4 +1,8 @@
 credentials:
+  database_password:
+    description: "Password for the Moodle database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
   user_password:
     description: "Initial password for the Moodle admin user"
     algorithm: "sha256"

roles/docker-nextcloud/meta/schema.yml

@@ -1,4 +1,8 @@
 credentials:
+  database_password:
+    description: "Password for the Nextcloud database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
 
   administrator_password:
     description: "Initial password for the Nextcloud administrator (change immediately and enable 2FA)"

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

@@ -4,4 +4,4 @@ plugin_configuration:
     configvalue:  "{{ applications.bigbluebutton.credentials.shared_secret }}"
   - appid:        "bbb"
     configkey:    "api.url"
-    configvalue:  "{{ web_protocol }}://{{domains | get_domain('bigbluebutton')}}{{applications.bigbluebutton.api_suffix}}"
+    configvalue:  "{{ web_protocol }}://{{domains | get_domain(''bigbluebutton'')}}{{applications.bigbluebutton.api_suffix}}"

roles/docker-openproject/meta/schema.yml

@@ -1,4 +1,8 @@
 credentials:
+  database_password:
+    description: "Password for the OpenProject PostgreSQL database"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
 
   oauth2_proxy_cookie_secret:
     description: "Secret used to encrypt cookies for the OAuth2 proxy (hex-encoded, 16 bytes)"

roles/docker-peertube/Setup.md

@@ -1,6 +0,0 @@
-# Setup Peertube
-
-## Change Root Administrator Password
-```bash
-docker exec -it -u peertube peertube npm run reset-password -- -u root
-```

roles/docker-peertube/Todo.md

@@ -1,3 +0,0 @@
-# Todos
-- [DKIM activate](https://docs.joinpeertube.org/install/docker)
-- The plugin needs to be manually activated in the admin interface. would be nice if this is automatized as well

roles/docker-peertube/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the PeerTube PostgreSQL database"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   secret:
     description: "PeerTube secret used for session signing and CSRF protection"
     algorithm: "sha256"

roles/docker-peertube/tasks/enable-oidc.yml

@@ -1,7 +1,3 @@
-- name: "Load OIDC Settings vor Peertube"
-  include_vars: vars/oidc-settings.yml
-  changed_when: false
-
 - name: "Install auth-openid-connect plugin for Peertube"
   command: >
     docker exec {{ container_name }} \
@@ -15,8 +11,20 @@
     login_host: "127.0.0.1"
     login_port: "{{ database_port }}"
     query: |
-      UPDATE public.plugin
-      SET settings = '{{ oidc_settings | to_json }}',
-      enabled = TRUE,
-      uninstalled = False,
+      UPDATE plugins
+      SET settings = '{
+        "scope": "openid email profile",
+        "client-id": "{{ oidc.client.id }}",
+        "discover-url": "{{ oidc.client.discovery_document }}",
+        "client-secret": "{{ oidc.client.secret }}",
+        "mail-property": "email",
+        "auth-display-name": "{{ oidc.button_text }}",
+        "username-property": "{{ oidc.attributes.username }}",
+        "signature-algorithm": "RS256",
+        "display-name-property": "{{ oidc.attributes.username }}"
+      }',
+      enabled = TRUE
       WHERE name = 'auth-openid-connect';
+  when: applications | is_feature_enabled('oidc', application_id)
+  become: true
+  become_user: "{{ container_name }}"

roles/docker-peertube/templates/docker-compose.yml.j2

@@ -5,8 +5,7 @@ services:
 {% include 'roles/docker-central-database/templates/services/' + database_type + '.yml.j2' %}
 
   application:
-    image: chocobozzz/peertube:production-{{ applications[application_id].version }}
-    container_name: {{ container_name }}
+    image: chocobozzz/peertube:production-{{applications.peertube.version}}
 {% include 'roles/docker-compose/templates/services/base.yml.j2' %}
     ports:
      - "1935:1935"  # @todo Add to ports

roles/docker-peertube/vars/configuration.yml

@@ -4,7 +4,7 @@ features:
   css:                false
   portfolio_iframe:   false
   central_database:   true
-  oidc:               true
+  oidc:               false
 csp:
   flags:
     script-src:

roles/docker-peertube/vars/oidc-settings.yml

@@ -1,12 +0,0 @@
-oidc_settings: |
-  {
-    "scope": "openid email profile",
-    "client-id": "{{ oidc.client.id }}",
-    "discover-url": "{{ oidc.client.discovery_document }}",
-    "client-secret": "{{ oidc.client.secret }}",
-    "mail-property": "email",
-    "auth-display-name": "{{ oidc.button_text }}",
-    "username-property": "{{ oidc.attributes.username }}",
-    "signature-algorithm": "RS256",
-    "display-name-property": "{{ oidc.attributes.username }}"
-  }

roles/docker-pixelfed/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the Pixelfed database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   app_key:
     description: "Application key used for encryption in Pixelfed (.env APP_KEY)"
     algorithm: "plain"

roles/docker-snipe-it/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the Snipe-IT database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   app_key:
     description: "Application encryption key for Snipe-IT (.env APP_KEY)"
     algorithm: "plain"

roles/docker-taiga/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the Taiga PostgreSQL database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   secret_key:
     description: "Django SECRET_KEY used for cryptographic signing in Taiga"
     algorithm: "sha256"

roles/docker-wordpress/meta/schema.yml

@@ -1,4 +1,9 @@
 credentials:
+  database_password:
+    description: "Password for the WordPress database user"
+    algorithm: "bcrypt"
+    validation: "^\\$2[aby]\\$.{56}$"
+
   administrator_password:
     description: "Initial password for the WordPress admin account"
     algorithm: "sha256"

roles/nginx-serve-assets/vars/configuration.yml

@@ -1,2 +1,3 @@
 source_directory: "{{ playbook_dir }}/assets"
-url: "{{ web_protocol ~ '://' ~ 'files.' ~ primary_domain ~ '/assets' }}"
+url: "{{ web_protocol ~ '://' ~ 'file-server' 
+   | load_configuration('domains.canonical[0]') ~ '/assets' }}"

roles/nginx-serve-files/tasks/main.yml

@@ -3,11 +3,11 @@
   include_role:
     name: nginx-https-get-cert-modify-all
   vars:
-    domain:     "{{ domains | get_domain(application_id) }}"
+    domain:     "{{domains | get_domain(application_id)}}"
     http_port:  "{{ ports.localhost.http[application_id] }}"
 
 - name: "generate {{domains | get_domain(application_id)}}.conf"
   template: 
     src:  "nginx.conf.j2" 
-    dest: "{{ nginx.directories.http.servers }}{{ domains | get_domain(application_id) }}.conf"
+    dest: "{{nginx.directories.http.servers}}{{domains | get_domain(application_id)}}.conf"
   notify: restart nginx

tests/unit/test_csp_filters.py

@@ -49,7 +49,7 @@ class TestCspFilters(unittest.TestCase):
             'app2': {}
         }
         self.domains = {
-            'matomo': ['matomo.example.org']
+            'matomo': 'matomo.example.org'
         }
 
     def test_get_csp_whitelist_list(self):

tests/unit/test_generate_vaulted_credentials.py

@@ -2,12 +2,9 @@ import pytest
 import sys, os
 from pathlib import Path
 
-sys.path.insert(
-    0,
-    os.path.abspath(
-        os.path.join(os.path.dirname(__file__), "../../cli")
-    ),
-)
+# 1) Add project root (two levels up) so 'cli' is on the path
+PROJECT_ROOT = Path(__file__).parent.parent.parent.resolve()
+sys.path.insert(0, str(PROJECT_ROOT))
 
 # 2) Import from the cli package
 import cli.create_credentials as gvc