#!/usr/bin/env python3
"""Repository safety checks for JARVIS Command Dashboard.

Stdlib only. No runtime/domain access. The script scans the repository worktree
for forbidden files, obvious secret values, unsafe runtime/path leaks and JSON
contract example validity.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]

FORBIDDEN_SUFFIXES = (
    ".db",
    ".sqlite",
    ".sqlite3",
    ".sqlite-wal",
    ".sqlite-shm",
    ".csv",
    ".tsv",
    ".xlsx",
    ".xls",
    ".pdf",
    ".gpg",
    ".zip",
    ".tar",
    ".tar.gz",
    ".7z",
)
FORBIDDEN_FILENAMES = {".env", "credentials.json"}
FORBIDDEN_DIRS = {
    "node_modules",
    "dist",
    "build",
    ".venv",
    "venv",
    "runtime",
    "jarvis_runtime",
    "data",
    "reports",
    "exports",
    "imports",
    "backups",
    "logs",
    "playwright-report",
    "test-results",
    "inbox",
    "processed",
    "archiv",
    "archive",
}
ALLOWED_FILENAMES = {".env.example", ".env.demo.example"}
SKIP_DIRS = {".git", "__pycache__", ".pytest_cache", ".ruff_cache", ".mypy_cache", "node_modules", "dist", "build", "coverage", ".tmp", "tmp", "playwright-report", "test-results"}

# Docs that intentionally document risky concepts or old inventory paths.
RUNTIME_LEAK_ALLOWLIST = (
    "docs/inventory/",
    "docs/architecture/runtime-config.md",
    "docs/security/forbidden-fields.md",
    "docs/architecture/redaction-policy.md",
    "docs/security/repo-safety-rules.md",
    "docs/security/git-safety.md",
    "scripts/verify-repo-safety.py",
    # Targeted exceptions for redaction implementation/tests that intentionally
    # contain unsafe sample strings to prove they are blocked. No broad apps/** allowlist.
    "apps/api-gateway/jarvis_gateway/redaction.py",
    "apps/api-gateway/tests/test_redaction.py",
    "apps/dashboard/src/tests/safety-ui.test.ts",
    "apps/dashboard/src/tests/api-live-mock.test.ts",
    "apps/dashboard/src/tests/store-integration.test.ts",
    "apps/dashboard/e2e/safety.spec.ts",
    "apps/dashboard/e2e/ux-acceptance.spec.ts",
    "apps/dashboard/src/api/errors.ts",
    "apps/api-gateway/jarvis_gateway/adapters/finance_sanitizer.py",
    "apps/api-gateway/tests/test_finance_sanitizer.py",
    "apps/api-gateway/tests/test_finance_live_readonly_adapter.py",
    "apps/api-gateway/tests/test_finance_live_redaction.py",
    "apps/api-gateway/tests/test_finance_live_contract_mapping.py",
    "apps/api-gateway/tests/test_finance_live_smoke_contract.py",
    "apps/api-gateway/tests/fixtures/finance_live_synthetic/runtime_status.json",
    "apps/api-gateway/tests/fixtures/finance_live_synthetic/dirty_sensitive_response.json",
    "scripts/smoke-finance-live-readonly.sh",
    "apps/api-gateway/jarvis_gateway/adapters/health_sanitizer.py",
    "apps/api-gateway/tests/test_health_redaction.py",
    "apps/api-gateway/tests/test_health_overview_resilience.py",
    "apps/api-gateway/tests/test_health_forbidden_surface.py",
    "apps/api-gateway/tests/fixtures/health_probe_synthetic/dirty_sensitive_health_response.json",
    "scripts/smoke-health-local-probe.sh",
    "apps/api-gateway/tests/test_legacy_links.py",
)
SECRET_SCAN_ALLOWLIST = (
    "docs/security/forbidden-fields.md",
    "docs/security/git-safety.md",
    "docs/security/repo-safety-rules.md",
    "scripts/verify-repo-safety.py",
    "apps/api-gateway/jarvis_gateway/redaction.py",
    "apps/api-gateway/tests/test_redaction.py",
    "apps/dashboard/src/api/errors.ts",
    "apps/dashboard/src/tests/safety-ui.test.ts",
    "apps/dashboard/src/tests/api-live-mock.test.ts",
    "apps/dashboard/src/tests/store-integration.test.ts",
    "apps/dashboard/e2e/safety.spec.ts",
    "apps/dashboard/e2e/ux-acceptance.spec.ts",
    "apps/api-gateway/jarvis_gateway/adapters/finance_sanitizer.py",
    "apps/api-gateway/tests/test_finance_sanitizer.py",
    "apps/api-gateway/tests/test_finance_live_readonly_adapter.py",
    "apps/api-gateway/tests/test_finance_live_redaction.py",
    "apps/api-gateway/tests/test_finance_live_contract_mapping.py",
    "apps/api-gateway/tests/test_finance_live_smoke_contract.py",
    "apps/api-gateway/tests/fixtures/finance_live_synthetic/runtime_status.json",
    "apps/api-gateway/tests/fixtures/finance_live_synthetic/dirty_sensitive_response.json",
    "scripts/smoke-finance-live-readonly.sh",
    "apps/api-gateway/jarvis_gateway/adapters/health_sanitizer.py",
    "apps/api-gateway/tests/test_health_redaction.py",
    "apps/api-gateway/tests/test_health_overview_resilience.py",
    "apps/api-gateway/tests/test_health_forbidden_surface.py",
    "apps/api-gateway/tests/fixtures/health_probe_synthetic/dirty_sensitive_health_response.json",
    "scripts/smoke-health-local-probe.sh",
    "apps/api-gateway/tests/test_legacy_links.py",
)

SECRET_PATTERNS = [
    # High-signal token prefixes / environment names.
    re.compile(r"github_pat_", re.IGNORECASE),
    re.compile(r"ghp_", re.IGNORECASE),
    re.compile(r"GITHUB_TOKEN\s*=", re.IGNORECASE),
    re.compile(r"OPENAI_API_KEY\s*=", re.IGNORECASE),
    re.compile(r"ANTHROPIC_API_KEY\s*=", re.IGNORECASE),
    # Assignment-style secret values. Generic documentation may mention these words;
    # committed key/value assignments are the dangerous case this repo-level scan blocks.
    re.compile(r"API_KEY\s*=", re.IGNORECASE),
    re.compile(r"SECRET\s*=", re.IGNORECASE),
    re.compile(r"TOKEN\s*=", re.IGNORECASE),
    re.compile(r"PASSWORD\s*=", re.IGNORECASE),
    re.compile(r"passphrase\s*=", re.IGNORECASE),
    re.compile(r"private_key\s*=", re.IGNORECASE),
    re.compile(r"oauth[^\n]{0,40}=", re.IGNORECASE),
]
RUNTIME_LEAK_PATTERNS = [
    "/home/agent/.hermes/assets/Gesundheit/",
    "/home/agent/jarvis_runtime/",
    "~/jarvis_runtime/",
    "health_data.db",
    "finance.sqlite",
    "finance.sqlite3",
    "drive_web_url",
    "drive_file_id",
    "local_original_path",
    "extrahierte_inhalte",
]
TEXT_SUFFIXES = {
    "",
    ".md",
    ".txt",
    ".json",
    ".yaml",
    ".yml",
    ".toml",
    ".py",
    ".sh",
    ".ts",
    ".vue",
    ".css",
    ".html",
    ".cjs",
    ".example",
    ".gitignore",
    ".mk",
}


def rel(path: Path) -> str:
    return path.relative_to(ROOT).as_posix()


def is_under_any(rel_path: str, prefixes: tuple[str, ...]) -> bool:
    return any(rel_path == prefix.rstrip("/") or rel_path.startswith(prefix) for prefix in prefixes)


def iter_files() -> list[Path]:
    files: list[Path] = []
    for current, dirs, names in os.walk(ROOT):
        current_path = Path(current)
        dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
        for name in names:
            path = current_path / name
            if ".git" in path.parts:
                continue
            files.append(path)
    return files


def is_text_file(path: Path) -> bool:
    if path.name in {"Makefile", ".gitignore"}:
        return True
    return path.suffix.lower() in TEXT_SUFFIXES


def load_forbidden_fields() -> set[str]:
    denylist = ROOT / "docs/security/forbidden-fields.md"
    fields: set[str] = set()
    if not denylist.exists():
        return fields
    for line in denylist.read_text(encoding="utf-8", errors="ignore").splitlines():
        stripped = line.strip()
        if not stripped.startswith("-"):
            continue
        value = stripped[1:].strip().strip("`").strip()
        if value and re.fullmatch(r"[A-Za-z0-9_]+", value):
            fields.add(value)
    return fields


def scan_forbidden_files(files: list[Path]) -> tuple[list[str], list[str]]:
    file_hits: list[str] = []
    dir_hits: set[str] = set()
    for path in files:
        r = rel(path)
        if path.name in ALLOWED_FILENAMES:
            continue
        if path.name in FORBIDDEN_FILENAMES:
            file_hits.append(r)
        lower = path.name.lower()
        if any(lower.endswith(s) for s in FORBIDDEN_SUFFIXES):
            file_hits.append(r)
        for part in path.relative_to(ROOT).parts[:-1]:
            if part in FORBIDDEN_DIRS:
                dir_hits.add(str(Path(*path.relative_to(ROOT).parts[: path.relative_to(ROOT).parts.index(part) + 1])))
    return sorted(set(file_hits)), sorted(dir_hits)


def validate_json_examples() -> tuple[int, list[str], list[str]]:
    example_dir = ROOT / "docs/contracts/examples"
    errors: list[str] = []
    forbidden_hits: list[str] = []
    forbidden = load_forbidden_fields()
    count = 0

    def walk(obj: Any, path: str, file: str) -> None:
        if isinstance(obj, dict):
            for key, value in obj.items():
                if key in forbidden:
                    forbidden_hits.append(f"{file}:{path}.{key}")
                if isinstance(value, str):
                    lower = value.lower()
                    for field in forbidden:
                        if field.lower() in lower:
                            forbidden_hits.append(f"{file}:{path}.{key} contains {field}")
                walk(value, f"{path}.{key}", file)
        elif isinstance(obj, list):
            for idx, value in enumerate(obj):
                walk(value, f"{path}[{idx}]", file)

    for path in sorted(example_dir.glob("*.json")):
        count += 1
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except Exception as exc:  # noqa: BLE001 - reporting only
            errors.append(f"{rel(path)}: {exc}")
            continue
        walk(data, "$", rel(path))
    return count, errors, forbidden_hits


def scan_text_patterns(files: list[Path]) -> tuple[list[str], list[str]]:
    secret_hits: list[str] = []
    runtime_hits: list[str] = []
    for path in files:
        r = rel(path)
        if not is_text_file(path):
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="ignore")
        except Exception:
            continue
        if not is_under_any(r, SECRET_SCAN_ALLOWLIST):
            for idx, line in enumerate(text.splitlines(), 1):
                for pattern in SECRET_PATTERNS:
                    if pattern.search(line):
                        secret_hits.append(f"{r}:{idx}: {pattern.pattern}")
                        break
        if not is_under_any(r, RUNTIME_LEAK_ALLOWLIST):
            for idx, line in enumerate(text.splitlines(), 1):
                for pattern in RUNTIME_LEAK_PATTERNS:
                    if pattern in line:
                        runtime_hits.append(f"{r}:{idx}: {pattern}")
                        break
    return secret_hits, runtime_hits


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--docs-only", action="store_true", help="Run the same safe checks; reserved for Makefile docs-check.")
    _args = parser.parse_args()

    files = iter_files()
    forbidden_file_hits, forbidden_dir_hits = scan_forbidden_files(files)
    checked_json_examples, json_errors, forbidden_json_hits = validate_json_examples()
    secret_hits, runtime_hits = scan_text_patterns(files)

    forbidden_pattern_hits = secret_hits + runtime_hits + forbidden_json_hits
    fail = bool(forbidden_file_hits or forbidden_dir_hits or json_errors or forbidden_pattern_hits)

    print("JARVIS repository safety check")
    print(f"checked_files={len(files)}")
    print(f"checked_json_examples={checked_json_examples}")
    print(f"forbidden_file_hits={len(forbidden_file_hits) + len(forbidden_dir_hits)}")
    for hit in forbidden_file_hits:
        print(f"  forbidden_file: {hit}")
    for hit in forbidden_dir_hits:
        print(f"  forbidden_dir: {hit}")
    print(f"forbidden_pattern_hits={len(forbidden_pattern_hits)}")
    for hit in forbidden_pattern_hits[:100]:
        print(f"  forbidden_pattern: {hit}")
    if len(forbidden_pattern_hits) > 100:
        print(f"  ... {len(forbidden_pattern_hits) - 100} more")
    print(f"json_errors={len(json_errors)}")
    for err in json_errors:
        print(f"  json_error: {err}")
    print(f"result={'FAIL' if fail else 'PASS'}")
    return 1 if fail else 0


if __name__ == "__main__":
    raise SystemExit(main())
