#!/usr/bin/env python3
"""Static privacy/branding audit for the private Chatwoot fork.

The audit deliberately fails only on explicit privacy rules. Visible product
branding is reported separately until the branding phase is complete. No
network calls are made.
"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
from pathlib import Path


RULES = (
    ("hub-url", re.compile(r"\bhub\.2\.chatwoot\.com\b", re.IGNORECASE)),
    ("amplitude-sdk", re.compile(r"@amplitude/analytics-browser", re.IGNORECASE)),
    ("sentry-sdk", re.compile(r"@sentry/vue", re.IGNORECASE)),
    (
        "hub-method",
        re.compile(
            r"\bChatwootHub\.(?:sync_with_hub|register_instance|emit_event|send_push(?:_with_response)?)\b"
        ),
    ),
    ("cwctl-event-report", re.compile(r"\breport_event\b", re.IGNORECASE)),
)

VISIBLE_BRANDING = re.compile(r"\bChatwoot\b")
VISIBLE_ROOTS = ("app/views", "app/javascript", "config/locales", "public")
BUILT_ROOTS = ("public/assets", "public/packs", "public/vite")
SKIP_PARTS = {".git", "node_modules", ".pnpm-store", "tmp", "log", "coverage", "storage"}
DEFAULT_ALLOWED_PATHS = {
    "LICENSE",
    "script/privacy_audit",
    "script/privacy_audit_test.sh",
    "deployment/spec/setup_20.04_privacy_test.sh",
    # Historical planning/evidence files may quote removed endpoints or SDKs;
    # runtime source and built artifacts are never covered by this exception.
    ".hermes/plans/2026-08-15_092534-chatwoot-private-rebrand.md",
    ".hermes/plans/chatwoot-private/06-remove-amplitude.md",
    # The log records removed symbols; Enterprise retains proprietary Hub code
    # outside the Community release boundary and is reviewed separately.
    "engineering-log.md",
    "spec/enterprise/lib/chatwoot_hub_spec.rb",
}

# Enterprise keeps a proprietary Hub boundary outside the Community release.
# Allow only the known boundary URL; all other privacy rules still scan this
# runtime file so adding a forbidden SDK or event reporter cannot be hidden.
DEFAULT_ALLOWED_RULES = {
    "enterprise/lib/enterprise/chatwoot_hub.rb": {"hub-url"},
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--root",
        type=Path,
        default=Path(__file__).resolve().parent.parent,
        help="repository or fixture root to scan",
    )
    parser.add_argument(
        "--allowlist-file",
        type=Path,
        help="file containing exact repository-relative paths allowed to contain findings",
    )
    parser.add_argument(
        "--report-only",
        action="store_true",
        help="print findings but exit zero; useful before removal/branding phases",
    )
    return parser.parse_args()


def is_skipped(path: Path, root: Path) -> bool:
    relative_parts = path.relative_to(root).parts
    if any(part in SKIP_PARTS for part in relative_parts):
        return True
    return len(relative_parts) >= 2 and relative_parts[:2] == ("public", "uploads")


def is_safe_file(path: Path, root: Path) -> bool:
    if path.is_symlink() or not path.is_file():
        return False
    try:
        path.resolve(strict=True).relative_to(root)
    except (OSError, ValueError):
        return False
    return True


def files_under(root: Path) -> list[Path]:
    files: list[Path] = []
    for directory, directory_names, file_names in os.walk(root, topdown=True, followlinks=False):
        current = Path(directory)
        directory_names[:] = [
            name for name in directory_names if not (current / name).is_symlink()
        ]
        files.extend(
            current / name
            for name in file_names
            if is_safe_file(current / name, root)
        )
    return files


def tracked_files(root: Path) -> list[Path]:
    git_dir = root / ".git"
    if git_dir.exists():
        result = subprocess.run(
            ["git", "-C", str(root), "ls-files", "-z"],
            check=True,
            capture_output=True,
        )
        paths = [root / Path(raw.decode("utf-8")) for raw in result.stdout.split(b"\0") if raw]
    else:
        paths = files_under(root)

    for relative_root in BUILT_ROOTS:
        build_root = root / relative_root
        if build_root.exists() and not build_root.is_symlink():
            paths.extend(files_under(build_root))

    unique = {path for path in paths if is_safe_file(path, root)}
    return sorted(path for path in unique if not is_skipped(path, root))


def load_allowlist(root: Path, allowlist_file: Path | None) -> set[str]:
    entries = set(DEFAULT_ALLOWED_PATHS)
    if allowlist_file is None:
        return entries

    allowlist_path = allowlist_file.resolve()
    try:
        allowlist_relative = allowlist_path.relative_to(root).as_posix()
    except ValueError as error:
        raise ValueError("allowlist file must be inside --root") from error

    entries.add(allowlist_relative)
    for raw_line in allowlist_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        parts = line.split("/")
        if (
            Path(line).is_absolute()
            or line in {".", ".."}
            or line.startswith(("./", "../", "/"))
            or any(part in {"", ".", ".."} for part in parts)
        ):
            raise ValueError(
                f"allowlist entry must be an exact repository-relative path: {line}"
            )
        entries.add(line)
    return entries


def relative_path(path: Path, root: Path) -> str:
    return path.relative_to(root).as_posix()


def is_visible_path(relative: str) -> bool:
    return any(relative == root or relative.startswith(f"{root}/") for root in VISIBLE_ROOTS)


def read_text(path: Path) -> str | None:
    data = path.read_bytes()
    if b"\0" in data:
        return None
    return data.decode("utf-8", errors="replace")


def main() -> int:
    args = parse_args()
    root = args.root.resolve()
    if not root.is_dir():
        print(f"privacy_audit: root does not exist: {root}", file=sys.stderr)
        return 2

    try:
        allowlist = load_allowlist(root, args.allowlist_file)
        files = tracked_files(root)
    except (OSError, ValueError, subprocess.CalledProcessError) as error:
        print(f"privacy_audit: discovery failed: {error}", file=sys.stderr)
        return 2

    findings = 0
    visible_reports = 0
    for path in files:
        relative = relative_path(path, root)
        text = read_text(path)
        if text is None:
            continue
        allowed_path = relative in allowlist
        allowed_rules = DEFAULT_ALLOWED_RULES.get(relative, set())

        for line_number, line in enumerate(text.splitlines(), start=1):
            for category, pattern in RULES:
                if pattern.search(line):
                    if allowed_path or category in allowed_rules:
                        continue
                    prefix = "REPORT" if args.report_only else "FAIL"
                    print(f"{prefix} {relative}:{line_number}:{category}")
                    findings += 1
                    break

            if is_visible_path(relative) and VISIBLE_BRANDING.search(line):
                print(f"REPORT {relative}:{line_number}:visible-branding")
                visible_reports += 1

    if args.report_only:
        print(
            f"privacy_audit: REPORT-ONLY findings={findings} "
            f"visible_branding_reports={visible_reports} files={len(files)}"
        )
        return 0

    if findings:
        print(
            f"privacy_audit: FAIL findings={findings} "
            f"visible_branding_reports={visible_reports} files={len(files)}",
            file=sys.stderr,
        )
        return 1

    print(
        f"privacy_audit: PASS findings=0 "
        f"visible_branding_reports={visible_reports} files={len(files)}"
    )
    return 0


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