ai-exchange/scripts/run_secret_static_checks.py

from __future__ import annotations

from pathlib import Path
from typing import Iterable, Optional
import json
import os
import re
import subprocess


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

PLACEHOLDER_PATTERN = re.compile(
    r"^(<[^>]+>|\$\{[^}]+\}|your[_-].+|change[_-]?me|changeme|placeholder|example|dummy|test|xxx+)$",
    re.IGNORECASE,
)
KNOWN_SECRET_PREFIX_PATTERN = re.compile(
    r"(?<![A-Za-z0-9_])("
    r"AKIA[0-9A-Z]{16}|"
    r"gh[pousr]_[A-Za-z0-9_]{20,}|"
    r"github_pat_[A-Za-z0-9_]{20,}|"
    r"sk-[A-Za-z0-9]{20,}|"
    r"xox[baprs]-[A-Za-z0-9-]{10,}"
    r")"
)
PRIVATE_KEY_PATTERN = re.compile(r"-----BEGIN (RSA |OPENSSH |EC |DSA )?PRIVATE KEY-----")
WEBHOOK_SECRET_URL_PATTERN = re.compile(
    r"https?://[^\s\"']*(webhook|hooks\.slack|discord|qyapi)[^\s\"']*(key|token|secret)=[^\s\"']+",
    re.IGNORECASE,
)
EXPECT_PASSWORD_ASSIGNMENT_PATTERN = re.compile(r"\bset\s+password\s+\"([^\"]+)\"", re.IGNORECASE)
EXPECT_SEND_LITERAL_PATTERN = re.compile(r"\bsend\s+\"([^\"]+)\"", re.IGNORECASE)
INLINE_SECRET_ASSIGNMENT_PATTERN = re.compile(
    r"(?:^|\s)("
    r"AI_ICT_DEPLOY_PASSWORD|OKX_API_SECRET|OKX_API_KEY|OKX_API_PASSPHRASE|"
    r"DATABASE_PASSWORD|WEBHOOK_URL|WEBHOOK_SECRET|MARKET_TRUTH_PROXY_AUTH|BROKER_READ_PROXY_AUTH"
    r")\s*=\s*[\"']?([^\s#\"']+)",
    re.IGNORECASE,
)


def run_secret_static_checks(env=None, root: Path = ROOT, files: Optional[Iterable[Path]] = None) -> dict:
    env = os.environ if env is None else env
    root = Path(root)
    findings = []
    checked_files = list(files) if files is not None else _tracked_files(root)
    for path in checked_files:
        path = Path(path)
        absolute = path if path.is_absolute() else root / path
        if not absolute.is_file() or _is_binary_or_large(absolute):
            continue
        try:
            text = absolute.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue
        findings.extend(_scan_text(path=absolute, root=root, text=text))
    return {
        "passed": len(findings) == 0,
        "checked_file_count": len(checked_files),
        "findings": findings,
    }


def _scan_text(*, path: Path, root: Path, text: str) -> list[dict]:
    findings = []
    relative = _display_path(path, root)
    normalized = relative.replace("\\", "/")
    if _is_allowed_example_file(normalized):
        return findings
    if _is_tracked_runtime_env(normalized):
        findings.append(_finding(relative, 1, "tracked_runtime_env", "runtime_env_files_must_not_be_tracked"))

    for line_number, line in enumerate(text.splitlines(), start=1):
        rule = _line_rule(normalized, line)
        if rule:
            findings.append(_finding(relative, line_number, rule[0], rule[1]))
    return findings


def _line_rule(normalized_path: str, line: str) -> Optional[tuple[str, str]]:
    stripped = line.strip()
    if not stripped or stripped.startswith("#"):
        return None
    if PRIVATE_KEY_PATTERN.search(line):
        return ("private_key_material", "private_key_material_must_not_be_committed")
    if KNOWN_SECRET_PREFIX_PATTERN.search(line):
        return ("known_secret_prefix", "known_secret_prefix_must_not_be_committed")
    if WEBHOOK_SECRET_URL_PATTERN.search(line) and not _contains_placeholder(line):
        return ("webhook_secret_url", "webhook_secret_url_must_use_environment_variable")
    if normalized_path.endswith(".exp"):
        password_match = EXPECT_PASSWORD_ASSIGNMENT_PATTERN.search(line)
        if password_match and not _is_placeholder(password_match.group(1)):
            return ("expect_password_literal", "expect_scripts_must_not_embed_passwords")
        send_match = EXPECT_SEND_LITERAL_PATTERN.search(line)
        if send_match and _looks_like_expect_secret_send(send_match.group(1)):
            return ("expect_send_secret_literal", "expect_scripts_must_not_send_secret_literals")
    assignment_match = INLINE_SECRET_ASSIGNMENT_PATTERN.search(line)
    if (
        assignment_match
        and not normalized_path.startswith("tests/")
        and not _is_placeholder(assignment_match.group(2))
        and not _is_dynamic_reference(assignment_match.group(2))
    ):
        return ("inline_secret_assignment", "secrets_must_be_supplied_by_runtime_environment")
    return None


def _tracked_files(root: Path) -> list[Path]:
    process = subprocess.run(
        ["git", "ls-files", "-z"],
        cwd=root,
        capture_output=True,
        check=False,
    )
    if process.returncode != 0:
        return [path.relative_to(root) for path in root.rglob("*") if path.is_file()]
    raw = process.stdout.decode("utf-8", errors="replace")
    return [Path(item) for item in raw.split("\0") if item]


def _is_binary_or_large(path: Path) -> bool:
    try:
        return path.stat().st_size > 2_000_000
    except OSError:
        return True


def _is_allowed_example_file(normalized_path: str) -> bool:
    name = normalized_path.rsplit("/", 1)[-1].lower()
    return name.endswith(".example") or name.endswith(".sample") or name.endswith(".template")


def _is_tracked_runtime_env(normalized_path: str) -> bool:
    name = normalized_path.rsplit("/", 1)[-1].lower()
    return name == ".env" or (name.startswith(".env.") and not _is_allowed_example_file(normalized_path))


def _looks_like_expect_secret_send(value: str) -> bool:
    if value in {"yes\\r", "y\\r", "$password\\r", "$::password\\r"}:
        return False
    if value.startswith("$") or _is_placeholder(value):
        return False
    return value.endswith("\\r") and bool(re.search(r"[A-Za-z0-9]", value))


def _contains_placeholder(value: str) -> bool:
    return any(token in value.lower() for token in ("<", "your_", "changeme", "placeholder", "example"))


def _is_dynamic_reference(value: str) -> bool:
    cleaned = value.strip().strip("\"'")
    return (
        cleaned.startswith("$")
        or cleaned.startswith("os.")
        or cleaned.startswith("env.")
        or cleaned.startswith("env[")
        or cleaned.startswith("settings.")
        or "." in cleaned
    )


def _is_placeholder(value: str) -> bool:
    cleaned = value.strip().strip("\"'").removesuffix("\\r")
    return bool(PLACEHOLDER_PATTERN.match(cleaned))


def _finding(file_path: str, line: int, pattern: str, reason: str) -> dict:
    return {
        "file": file_path,
        "line": line,
        "pattern": pattern,
        "reason": reason,
    }


def _display_path(path: Path, root: Path) -> str:
    try:
        return str(path.relative_to(root))
    except ValueError:
        return path.name


def main(env=None) -> int:
    env = os.environ if env is None else env
    result = run_secret_static_checks(env=env)
    if _get_bool(env, "SECRET_CHECKS_OUTPUT_JSON", False):
        print(json.dumps(result, ensure_ascii=False, sort_keys=True))
    else:
        print(f"Secret static checks passed: {result['passed']}")
        print(f"Files checked: {result['checked_file_count']}")
        for finding in result["findings"]:
            print(
                f"{finding['file']}:{finding['line']} "
                f"{finding['reason']} pattern={finding['pattern']}"
            )
    return 0 if result["passed"] else 2


def _get_bool(env, key: str, default: bool) -> bool:
    value = env.get(key)
    if value is None or value == "":
        return default
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


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