trading-system-docs/red-queen-architecture/control-plane/cp2a2_2b/test_worker_credential_zero.py

# CP2A2.2B — L1 Red Queen Worker: Credential-Zero-Test
# STRICTLY NON-AUTONOMOUS / NON-PRODUCTIVE.
#
# Verifiziert, dass der Worker KEINE mutierenden Credentials besitzt.
# Report NUR PRESENT/ABSENT — NIEMALS Secret-Werte ausgeben.
#
# Geprueft werden:
#   - ENV-Variablen (mutierende Credential-Namen)
#   - config.yaml (worker-config.yaml)
#   - Home-Verzeichnis (keine .netrc, .git-credentials, SSH-Keys)
#   - git config (keine credential helpers)
#   - mounted secrets

import os
import re
import sys
import subprocess

# Mutierende Credential-Namen (nur Namen, keine Werte)
MUTATING_CREDENTIAL_NAMES = [
    "FORGEJO_WRITE_TOKEN",
    "NOTION_API_KEY",
    "TELEGRAM_BOT_TOKEN",
    "SAVE_TOKEN",
    "DELETE_TOKEN",
    "TOLARIA_SAVE_TOKEN",
    "TOLARIA_DELETE_TOKEN",
    "GITHUB_TOKEN",
    "GITLAB_TOKEN",
    "AWS_ACCESS_KEY_ID",
    "AWS_SECRET_ACCESS_KEY",
    "AZURE_CLIENT_SECRET",
    "GCP_SA_KEY",
    "DOCKER_AUTH_CONFIG",
    "DOCKER_CONFIG",
    "IG_API_KEY",
    "IG_ACCESS_TOKEN",
    "BROKER_API_KEY",
    "BROKER_TOKEN",
    "TRADING_TOKEN",
    "WEBHOOK_SECRET",
    "SERVICE_ACCOUNT_KEY",
    "DB_PASSWORD",
    "POSTGRES_PASSWORD",
    "MYSQL_PASSWORD",
    "REDIS_PASSWORD",
    "MONGO_PASSWORD",
]

# Dateien, die Credentials enthalten koennten (im Home/Config)
CREDENTIAL_FILE_PATTERNS = [
    ".netrc",
    ".git-credentials",
    ".pgpass",
    ".my.cnf",
    ".aws/credentials",
    ".config/gcloud/application_default_credentials.json",
    ".docker/config.json",
    "id_rsa",
    "id_ed25519",
    "id_dsa",
    "id_ecdsa",
]


def check_env() -> list[str]:
    """Prueft ENV auf mutierende Credential-Namen. Gibt gefundene Namen zurueck."""
    found = []
    for name in MUTATING_CREDENTIAL_NAMES:
        if name in os.environ:
            found.append(name)
    return found


def check_config(config_path: str) -> list[str]:
    """Prueft die Worker-Config auf mutierende Credential-Namen.

    Ignoriert Kommentare (Zeilen mit #) — Kommentare sind Dokumentation,
    keine tatsaechlichen Credentials. Prueft nur echte YAML-Keys/Werte.
    """
    found = []
    if not os.path.exists(config_path):
        return found
    try:
        with open(config_path, "r", encoding="utf-8") as f:
            lines = f.readlines()
        # Nur Nicht-Kommentar-Zeilen pruefen
        content = "\n".join(
            line for line in lines if not line.lstrip().startswith("#")
        )
        for name in MUTATING_CREDENTIAL_NAMES:
            if name in content:
                found.append(name)
        # Generische Secret-Schluessel in YAML (nur echte Keys, nicht Kommentare)
        for key in ["token", "api_key", "secret", "password", "bearer"]:
            if re.search(rf"^\s*{key}\s*:", content, re.IGNORECASE | re.MULTILINE):
                found.append(f"config-key:{key}")
    except Exception:
        pass
    return found


def check_home(home: str) -> list[str]:
    """Prueft Home-Verzeichnis auf Credential-Dateien/SSH-Keys."""
    found = []
    for pattern in CREDENTIAL_FILE_PATTERNS:
        path = os.path.join(home, pattern)
        if os.path.exists(path):
            found.append(pattern)
    # SSH-Verzeichnis
    ssh_dir = os.path.join(home, ".ssh")
    if os.path.isdir(ssh_dir):
        for entry in os.listdir(ssh_dir):
            if entry not in ("known_hosts", "config"):
                found.append(f".ssh/{entry}")
    return found


def check_git_config(home: str) -> list[str]:
    """Prueft git config auf credential helpers."""
    found = []
    git_config = os.path.join(home, ".gitconfig")
    if os.path.exists(git_config):
        try:
            with open(git_config, "r", encoding="utf-8") as f:
                content = f.read()
            if "credential" in content.lower() and "helper" in content.lower():
                found.append("git-credential-helper")
        except Exception:
            pass
    return found


def main() -> int:
    home = os.environ.get("HOME", "/opt/data")
    config_path = os.environ.get("WORKER_CONFIG", "/opt/l1-worker/worker-config.yaml")

    env_found = check_env()
    config_found = check_config(config_path)
    home_found = check_home(home)
    git_found = check_git_config(home)

    all_found = sorted(set(env_found + config_found + home_found + git_found))

    print("=== CP2A2.2B CREDENTIAL-ZERO TEST ===")
    print(f"ENV mutating credentials: {env_found if env_found else 'ABSENT'}")
    print(f"Config mutating credentials: {config_found if config_found else 'ABSENT'}")
    print(f"Home credential files: {home_found if home_found else 'ABSENT'}")
    print(f"Git credential helpers: {git_found if git_found else 'ABSENT'}")
    print(f"TOTAL MUTATING CREDENTIALS: {len(all_found)}")
    print(f"RESULT: {'PASS' if not all_found else 'FAIL'}")

    return 0 if not all_found else 1


if __name__ == "__main__":
    sys.exit(main())