PlantAI-rag/tools/audit_open_source_readiness.py

#!/usr/bin/env python3
"""Read-only audit for the PlantAI-rag open-source cleanup surface."""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable, Sequence


PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from tools.runtime_surface import classify_path as classify_runtime_path  # noqa: E402

TEXT_SUFFIXES = {
    ".cfg",
    ".css",
    ".env",
    ".html",
    ".ini",
    ".js",
    ".json",
    ".md",
    ".py",
    ".sh",
    ".toml",
    ".ts",
    ".txt",
    ".vue",
    ".yaml",
    ".yml",
}

ABSOLUTE_PATH_RE = re.compile(
    r"(?P<value>/(?:home/[A-Za-z0-9._-]+|data/(?:gene-family-mas-v2|wutx|wutx_old))"
    r"[A-Za-z0-9._~+/@:%,-]*)"
)

SECRET_ASSIGNMENT_RE = re.compile(
    r"\b(?P<name>[A-Z][A-Z0-9_]*(?:API[_-]?KEY|PASSWORD|SECRET|"
    r"(?:ACCESS|AUTH|BEARER|REFRESH|ID|API)[_-]?TOKEN)[A-Z0-9_]*)"
    r"\b\s*[:=]\s*[\"']?(?P<value>[^\"'\s#,)}]+)"
)

DANGEROUS_DEFAULT_RE = re.compile(
    r"(?i)(?:os\.environ\.get|os\.getenv|env)\([^)]*(?:PASSWORD|TOKEN|SECRET|API[_-]?KEY)"
    r"[^)]*,\s*[\"'](?P<value>password|admin|changeme|123456)[\"']"
)

MANUAL_ENV_FILE_HANDLING_RE = re.compile(
    r"\bload_dotenv\s*\("
    r"|Path\.home\(\)\s*/\s*[\"']\.env[\"']"
    r"|open\([^)]*[\"'][^\"']*\.env[\"']"
    r"|\.startswith\([\"'][A-Z0-9_]*(?:API[_-]?KEY|PASSWORD|SECRET|TOKEN)="
)

MANUAL_ENV_FILE_HANDLING_ALLOWED = {
    "src/rag/project_config.py",
    "tools/audit_open_source_readiness.py",
}

PLACEHOLDER_VALUES = {
    "",
    "<empty>",
    "<placeholder>",
    "changeme",
    "env",
    "none",
    "null",
    "os.environ.get(",
    "os.getenv(",
    "your_api_key",
    "your-api-key",
}

AUDIT_FIXTURE_IGNORE_MARKER = "open-source-audit-fixture"


@dataclass(frozen=True)
class AuditIssue:
    path: str
    line: int
    category: str
    severity: str
    kind: str
    value: str
    message: str


def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Scan tracked files for open-source cleanup issues without editing them."
    )
    parser.add_argument("--root", type=Path, default=PROJECT_ROOT, help="Repository root.")
    parser.add_argument("--json", action="store_true", help="Emit JSON instead of text.")
    parser.add_argument(
        "--no-fail",
        action="store_true",
        help="Always exit 0 after reporting issues.",
    )
    parser.add_argument(
        "--max-by-category",
        type=int,
        default=40,
        help="Maximum text issues to print per category.",
    )
    return parser.parse_args(list(argv) if argv is not None else None)


def classify_path(path: str) -> str:
    return classify_runtime_path(path)


def issue_severity(category: str, kind: str) -> str:
    if category == "current_runtime":
        return "high" if kind in {
            "secret_assignment",
            "dangerous_default",
            "manual_env_file_handling",
        } else "medium"
    if category == "test_fixture":
        return "info"
    if kind in {"secret_assignment", "dangerous_default"}:
        return "medium"
    return "info"


def is_probably_text(path: Path) -> bool:
    if path.suffix.lower() in TEXT_SUFFIXES:
        return True
    return path.name in {".env.example", "Dockerfile", "Makefile"}


def read_text(path: Path) -> str | None:
    if not is_probably_text(path):
        return None
    try:
        raw = path.read_bytes()
    except OSError:
        return None
    if b"\0" in raw:
        return None
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        return raw.decode("utf-8", errors="replace")


def is_secret_placeholder(value: str) -> bool:
    normalized = value.strip().strip("\"'").lower()
    if normalized in PLACEHOLDER_VALUES:
        return True
    if normalized.startswith(("os.environ", "os.getenv", "env(")):
        return True
    if normalized.startswith("${"):
        return True
    return False


def is_shell_expansion_reference(line: str, name: str) -> bool:
    return f"${{{name}" in line


def is_explicit_test_fixture_line(category: str, line: str) -> bool:
    return category == "test_fixture" and AUDIT_FIXTURE_IGNORE_MARKER in line


def allows_manual_env_file_handling(path: str) -> bool:
    return path in MANUAL_ENV_FILE_HANDLING_ALLOWED


def scan_text(path: str, text: str) -> list[AuditIssue]:
    category = classify_path(path)
    issues: list[AuditIssue] = []
    for line_number, line in enumerate(text.splitlines(), start=1):
        if is_explicit_test_fixture_line(category, line):
            continue
        manual_env_match = MANUAL_ENV_FILE_HANDLING_RE.search(line)
        if manual_env_match and not allows_manual_env_file_handling(path):
            issues.append(
                AuditIssue(
                    path=path,
                    line=line_number,
                    category=category,
                    severity=issue_severity(category, "manual_env_file_handling"),
                    kind="manual_env_file_handling",
                    value=manual_env_match.group(0),
                    message="Manual .env loading/parsing should be centralized in src/rag/project_config.py.",
                )
            )

        for match in ABSOLUTE_PATH_RE.finditer(line):
            value = match.group("value")
            issues.append(
                AuditIssue(
                    path=path,
                    line=line_number,
                    category=category,
                    severity=issue_severity(category, "personal_absolute_path"),
                    kind="personal_absolute_path",
                    value=value,
                    message="Personal or project-specific absolute path should be env-driven or archived.",
                )
            )

        default_match = DANGEROUS_DEFAULT_RE.search(line)
        if default_match:
            value = default_match.group("value")
            issues.append(
                AuditIssue(
                    path=path,
                    line=line_number,
                    category=category,
                    severity=issue_severity(category, "dangerous_default"),
                    kind="dangerous_default",
                    value=value,
                    message="Credential-like setting has a dangerous hard-coded default.",
                )
            )

        secret_match = SECRET_ASSIGNMENT_RE.search(line)
        if secret_match:
            name = secret_match.group("name")
            value = secret_match.group("value")
            if is_secret_placeholder(value) or is_shell_expansion_reference(line, name):
                continue
            issues.append(
                AuditIssue(
                    path=path,
                    line=line_number,
                    category=category,
                    severity=issue_severity(category, "secret_assignment"),
                    kind="secret_assignment",
                    value=redact_value(value),
                    message="Credential-like assignment should come from env without a real default.",
                )
            )
    return issues


def redact_value(value: str) -> str:
    if len(value) <= 4:
        return "<redacted>"
    return f"{value[:2]}...{value[-2:]}"


def git_tracked_files(root: Path) -> list[str]:
    result = subprocess.run(
        ["git", "-C", str(root), "ls-files", "-z"],
        check=True,
        capture_output=True,
    )
    return [item.decode("utf-8") for item in result.stdout.split(b"\0") if item]


def audit_files(root: Path, files: Sequence[str]) -> list[AuditIssue]:
    issues: list[AuditIssue] = []
    for rel_path in files:
        text = read_text(root / rel_path)
        if text is None:
            continue
        issues.extend(scan_text(rel_path, text))
    return issues


def summarize_issues(issues: Sequence[AuditIssue]) -> dict[str, object]:
    by_category: dict[str, int] = {}
    by_severity: dict[str, int] = {}
    by_kind: dict[str, int] = {}
    for issue in issues:
        by_category[issue.category] = by_category.get(issue.category, 0) + 1
        by_severity[issue.severity] = by_severity.get(issue.severity, 0) + 1
        by_kind[issue.kind] = by_kind.get(issue.kind, 0) + 1
    blocking = [
        issue
        for issue in issues
        if issue.category == "current_runtime" and issue.severity in {"high", "medium"}
    ]
    return {
        "total": len(issues),
        "blocking_current_runtime": len(blocking),
        "by_category": by_category,
        "by_severity": by_severity,
        "by_kind": by_kind,
    }


def build_report(issues: Sequence[AuditIssue]) -> dict[str, object]:
    return {
        "summary": summarize_issues(issues),
        "issues": [asdict(issue) for issue in issues],
    }


def print_text_report(report: dict[str, object], max_by_category: int) -> None:
    summary = report["summary"]
    issues = [AuditIssue(**item) for item in report["issues"]]  # type: ignore[index]
    print("Open-source readiness audit")
    print(json.dumps(summary, ensure_ascii=False, indent=2))

    grouped: dict[str, list[AuditIssue]] = {}
    for issue in issues:
        grouped.setdefault(issue.category, []).append(issue)

    for category in ("current_runtime", "other_tracked", "legacy_script", "test_fixture", "docs_history"):
        category_issues = grouped.get(category, [])
        if not category_issues:
            continue
        print(f"\n[{category}] {len(category_issues)} issue(s)")
        for issue in category_issues[:max_by_category]:
            print(
                f"- {issue.severity} {issue.kind} {issue.path}:{issue.line} "
                f"{issue.value} :: {issue.message}"
            )
        remaining = len(category_issues) - max_by_category
        if remaining > 0:
            print(f"- ... {remaining} more issue(s)")


def main(argv: Iterable[str] | None = None) -> int:
    args = parse_args(argv)
    root = args.root.resolve()
    tracked = git_tracked_files(root)
    issues = audit_files(root, tracked)
    report = build_report(issues)

    if args.json:
        print(json.dumps(report, ensure_ascii=False, indent=2))
    else:
        print_text_report(report, max_by_category=args.max_by_category)

    blocking = int(report["summary"]["blocking_current_runtime"])  # type: ignore[index]
    if blocking and not args.no_fail:
        return 2
    return 0


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