han-app/VM1_app/codebase/backend/tests/test_secret_hygiene.py

from __future__ import annotations

import os
import runpy
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
VALIDATOR = ROOT / "scripts/validate-env"


def runtime_values() -> dict[str, str]:
    symbols = runpy.run_path(str(VALIDATOR))
    values = {key: "a" * 32 for key in symbols["REQUIRED_RUNTIME"]}
    pg_tail = "?sslmode=verify-full&sslrootcert=/run/secrets/pg-ca.pem"
    values.update(
        {
            "DATABASE_URL": f"postgresql+asyncpg://han_app:password@pg:5433/han_chat{pg_tail}",
            "BITRIX_DATABASE_URL": f"postgresql://bitrix:password@pg:5433/han_chat{pg_tail}",
            "SMS_DATABASE_URL": f"postgresql+asyncpg://sms:password@pg:5433/han_chat{pg_tail}",
            "KEYCLOAK_DB_URL": f"jdbc:postgresql://pg:5433/han_chat{pg_tail}",
            "REDIS_API_PASSWORD": "r" * 32,
            "REDIS_HEALTH_PASSWORD": "h" * 32,
            "REDIS_URL": f"redis://api_backend:{'r' * 32}@redis:6379/0",
            "REDIS_REALTIME_URL": f"redis://api_backend:{'r' * 32}@redis:6379/1",
            "BITRIX_LOCAL_APP_INTERNAL_TOKEN": "b" * 32,
            "BITRIX_INTERNAL_API_TOKEN": "b" * 32,
            "BITRIX_API_FORWARD_TOKEN": "f" * 32,
            "BITRIX_API_INBOX_TOKEN": "f" * 32,
            "KEYCLOAK_SMS_SERVICE_TOKEN": "k" * 32,
            "SMS_SERVICE_TOKEN": "k" * 32,
            "KEYCLOAK_OTP_MOCK_CODE": "846271",
        }
    )
    return values


class SecretHygieneTests(unittest.TestCase):
    def run_validator(
        self, config: Path, *args: str, environment: dict[str, str] | None = None
    ) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            [sys.executable, str(VALIDATOR), str(config), *args],
            text=True,
            capture_output=True,
            check=False,
            env=environment,
        )

    def test_example_is_non_secret_config(self) -> None:
        example = (ROOT / ".env.example").read_text(encoding="utf-8")
        symbols = runpy.run_path(str(VALIDATOR))
        for key in symbols["FORBIDDEN_CONFIG_KEYS"]:
            self.assertNotRegex(example, rf"(?m)^{key}=")
        self.assertIn("SECRETS_SOURCE=file", example)

    def test_config_rejects_secret_key_and_credential_url(self) -> None:
        example = (ROOT / ".env.example").read_text(encoding="utf-8")
        with tempfile.TemporaryDirectory() as directory:
            config = Path(directory) / ".env"
            config.write_text(
                example
                + "\nSMS_SERVICE_TOKEN=not-for-config\n"
                + "EXTERNAL_URL=https://user:password@example.net/path\n",
                encoding="utf-8",
            )
            result = self.run_validator(config)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("SMS_SERVICE_TOKEN", result.stderr)
        self.assertIn("URL с credentials", result.stderr)
        self.assertNotIn("not-for-config", result.stderr)

    def test_runtime_environment_is_validated_without_value_disclosure(self) -> None:
        values = runtime_values()
        secret_canary = "canary-never-print-" + "x" * 20
        values["CURSOR_HMAC_SECRET"] = secret_canary
        environment = {**os.environ, **values}
        with tempfile.TemporaryDirectory() as directory:
            config = Path(directory) / ".env"
            config.write_text(
                (ROOT / ".env.example").read_text(encoding="utf-8").replace(
                    "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false",
                    "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true",
                ),
                encoding="utf-8",
            )
            result = self.run_validator(config, "--runtime-env", environment=environment)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertNotIn(secret_canary, result.stdout + result.stderr)

    def test_runtime_manifest_reads_protected_files_without_value_disclosure(self) -> None:
        values = runtime_values()
        secret_canary = "manifest-canary-" + "z" * 20
        values["CURSOR_HMAC_SECRET"] = secret_canary
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            config = root / ".env"
            config.write_text(
                (ROOT / ".env.example").read_text(encoding="utf-8").replace(
                    "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false",
                    "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true",
                ),
                encoding="utf-8",
            )
            entries = []
            for index, (key, value) in enumerate(sorted(values.items())):
                path = root / f"secret-{index}"
                path.write_text(value + "\n", encoding="utf-8")
                path.chmod(0o600)
                entries.append(f"{key}={path.resolve()}")
            manifest = root / "manifest"
            manifest.write_text("\n".join(entries) + "\n", encoding="utf-8")
            result = self.run_validator(config, "--runtime-manifest", str(manifest))
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertNotIn(secret_canary, result.stdout + result.stderr)

    def test_validator_enforces_remote_safety_tls_contract(self) -> None:
        example = (ROOT / ".env.example").read_text(encoding="utf-8")
        cases = (
            (
                "MESSAGE_SAFETY_URL=https://processing.internal:8443",
                "MESSAGE_SAFETY_URL=http://message-safety:8080",
                "remote TLS endpoint",
            ),
            (
                "MESSAGE_SAFETY_CA_HOST_PATH=/etc/han/ca/vm2-internal-ca.pem",
                "MESSAGE_SAFETY_CA_HOST_PATH=",
                "MESSAGE_SAFETY_CA_HOST_PATH",
            ),
            (
                "MESSAGE_SAFETY_EXTRA_HOST=processing.internal=192.168.0.4",
                "MESSAGE_SAFETY_EXTRA_HOST=processing.internal=8.8.8.8",
                "MESSAGE_SAFETY_EXTRA_HOST",
            ),
            (
                "MESSAGE_SAFETY_API_PREFIX=/internal/safety/v2",
                "MESSAGE_SAFETY_API_PREFIX=/internal/safety/v1",
                "MESSAGE_SAFETY_API_PREFIX",
            ),
        )
        for original, replacement, expected_error in cases:
            with self.subTest(replacement=replacement), tempfile.TemporaryDirectory() as directory:
                config = Path(directory) / ".env"
                config.write_text(
                    example.replace(original, replacement),
                    encoding="utf-8",
                )
                result = self.run_validator(config)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn(expected_error, result.stderr)


if __name__ == "__main__":
    unittest.main()