ai-contract-platform/yuqei-infra/scripts/run-source-secret-scan.py

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_BASELINE = REPO_ROOT / ".secrets.baseline"
DEFAULT_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-source-secret-scan"
SOURCE_PATHS = (
    ".github",
    "project-docs",
    "yuqei-ai-platform",
    "yuqei-ai-sdk-python",
    "yuqei-contract-platform",
    "yuqei-infra",
    "AGENTS.md",
    "README.md",
    "V2-README.md",
    "package.json",
    "pyproject.toml",
)
EXCLUDE_FILES = (
    r"(^|[\\/])(\.git|\.tmp-pytest|node_modules|\.next|\.venv[^\\/]*|\.runtime|output|"
    r"qa-artifacts|automation-artifacts|\.playwright-cli|\.codex-temp|ui-design-shotgun-2026-07-04)"
    r"([\\/]|$)|(^|[\\/])\.secrets\.baseline$|\.log$|package-lock\.json$|tsconfig\.tsbuildinfo$"
)


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Scan source files with detect-secrets and enforce the reviewed baseline.",
    )
    parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE)
    parser.add_argument("--artifact-dir", type=Path)
    parser.add_argument(
        "--include-untracked",
        action="store_true",
        help="Scan the current source tree, including untracked files that are not build artifacts.",
    )
    return parser.parse_args()


def _normalize_path(value: str) -> str:
    normalized = value.replace("\\", "/")
    while normalized.startswith("./"):
        normalized = normalized[2:]
    return normalized


def _finding_key(finding: dict[str, Any]) -> tuple[str, str, str]:
    return (
        _normalize_path(str(finding["filename"])),
        str(finding["type"]),
        str(finding["hashed_secret"]),
    )


def _flatten_results(payload: dict[str, Any]) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    for filename, items in payload.get("results", {}).items():
        for item in items:
            finding = dict(item)
            finding["filename"] = filename
            findings.append(finding)
    return findings


def _redact_finding(finding: dict[str, Any], baseline_status: str) -> dict[str, Any]:
    return {
        "filename": _normalize_path(str(finding["filename"])),
        "line_number": int(finding["line_number"]),
        "type": str(finding["type"]),
        "is_verified": bool(finding.get("is_verified", False)),
        "baseline_status": baseline_status,
    }


def _tracked_file_count() -> int:
    completed = subprocess.run(
        ["git", "ls-files", "-z"],
        cwd=REPO_ROOT,
        check=True,
        capture_output=True,
    )
    return len([item for item in completed.stdout.split(b"\0") if item])


def _run_detect_secrets(include_untracked: bool) -> dict[str, Any]:
    command = [
        sys.executable,
        "-m",
        "detect_secrets",
        "scan",
        "--no-verify",
        "--exclude-files",
        EXCLUDE_FILES,
    ]
    if include_untracked:
        command.append("--all-files")
        command.extend(path for path in SOURCE_PATHS if (REPO_ROOT / path).exists())

    completed = subprocess.run(
        command,
        cwd=REPO_ROOT,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        # detect-secrets opens source files with the process locale; keep Windows scans
        # equivalent to the UTF-8 Linux CI scan instead of silently skipping files.
        env={**os.environ, "PYTHONUTF8": "1"},
    )
    if completed.returncode != 0:
        raise RuntimeError(completed.stderr.strip() or "detect-secrets failed without diagnostic output")
    return json.loads(completed.stdout.lstrip("\ufeff"))


def main() -> int:
    args = _parse_args()
    baseline_path = args.baseline.resolve()
    baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
    scan = _run_detect_secrets(args.include_untracked)

    baseline_findings = _flatten_results(baseline)
    accepted_keys = {
        _finding_key(item)
        for item in baseline_findings
        if item.get("is_secret") is False
    }
    actual_findings = _flatten_results(scan)
    actual_keys = {_finding_key(item) for item in actual_findings}
    new_findings = [item for item in actual_findings if _finding_key(item) not in accepted_keys]
    stale_baseline = [item for item in baseline_findings if _finding_key(item) not in actual_keys]

    checked_at = datetime.now().astimezone()
    artifact_dir = args.artifact_dir
    if artifact_dir is None:
        artifact_dir = DEFAULT_ARTIFACT_ROOT / checked_at.strftime("%Y-%m-%dT%H-%M-%S")
    artifact_dir = artifact_dir.resolve()
    artifact_dir.mkdir(parents=True, exist_ok=True)

    new_keys = {_finding_key(item) for item in new_findings}
    redacted_findings = [
        _redact_finding(item, "new" if _finding_key(item) in new_keys else "reviewed_false_positive")
        for item in actual_findings
    ]
    summary = {
        "status": "pass" if not new_findings else "fail",
        "checked_at": checked_at.isoformat(),
        "scanner": "detect-secrets",
        "scanner_version": scan.get("version"),
        "scan_scope": "source_tree_including_untracked" if args.include_untracked else "git_tracked",
        "tracked_file_count": _tracked_file_count(),
        "detected_candidates": len(actual_findings),
        "reviewed_false_positives": len(actual_findings) - len(new_findings),
        "new_findings": len(new_findings),
        "stale_baseline_entries": len(stale_baseline),
        "baseline": _normalize_path(str(baseline_path.relative_to(REPO_ROOT))),
        "findings": redacted_findings,
        "stale_baseline": [
            _redact_finding(item, "stale") for item in stale_baseline
        ],
    }
    summary_path = artifact_dir / "source-secret-scan-summary.json"
    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    print(
        "SOURCE_SECRET_SCAN_"
        f"{'PASS' if not new_findings else 'FAIL'} "
        f"scope={summary['scan_scope']} candidates={len(actual_findings)} new={len(new_findings)} "
        f"artifact={summary_path}"
    )
    return 0 if not new_findings else 1


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