crank/scripts/check-community-scope.py

#!/usr/bin/env python3
import argparse
import io
import os
import re
import subprocess
import sys
from pathlib import Path, PurePosixPath


DEFAULT_EXCLUDED_PATHS = {
    "LICENSE",
    "Cargo.lock",
    "apps/ui/package-lock.json",
    "scripts/check-community-scope.py",
    "tests/unit/test_check_community_scope.py",
}

DEFAULT_EXCLUDED_PREFIXES = {
    "target/",
    "apps/ui/node_modules/",
    "apps/ui/dist/",
}

ALLOW_DIRECTIVE = re.compile(
    r"community-scope:\s*allow=([a-z0-9-]+(?:,[a-z0-9-]+)*)",
    re.IGNORECASE,
)

FORBIDDEN_PATTERNS = [
    (
        "multi-workspace",
        re.compile(r"\bmulti[-_ ]workspace\b", re.IGNORECASE),
    ),
    (
        "enterprise",
        re.compile(r"\benterprise\b|\benterprise_rbac\b", re.IGNORECASE),
    ),
    ("cloud", re.compile(r"\bcloud\b", re.IGNORECASE)),
    ("commercial", re.compile(r"\bcommercial\b|коммер", re.IGNORECASE)),
    ("cloud-russian", re.compile(r"облач", re.IGNORECASE)),
    ("graphql", re.compile(r"\bgraphql\b", re.IGNORECASE)),
    ("grpc", re.compile(r"\bgrpc\b", re.IGNORECASE)),
    ("soap", re.compile(r"\bsoap\b", re.IGNORECASE)),
    ("websocket", re.compile(r"\bwebsocket\b", re.IGNORECASE)),
    ("short-lived-token", re.compile(r"short[-_ ]?lived", re.IGNORECASE)),
    ("one-time-token", re.compile(r"one[-_ ]?time", re.IGNORECASE)),
    ("sso", re.compile(r"\bsso\b", re.IGNORECASE)),
    ("two-factor", re.compile(r"two[-_ ]?factor|\b2fa\b|\bmfa\b", re.IGNORECASE)),
    ("billing", re.compile(r"\bbilling\b", re.IGNORECASE)),
    ("tenant", re.compile(r"\btenant\b", re.IGNORECASE)),
    ("stream-session", re.compile(r"stream session", re.IGNORECASE)),
    ("async-job", re.compile(r"async job", re.IGNORECASE)),
    ("mcp-auth", re.compile(r"mcp-auth", re.IGNORECASE)),
    ("protocol-capabilities", re.compile(r"protocol-capabilities", re.IGNORECASE)),
    ("machine-token", re.compile(r"machine token", re.IGNORECASE)),
    ("token-issuer", re.compile(r"token issuer", re.IGNORECASE)),
    ("request-variables", re.compile(r"request\.variables", re.IGNORECASE)),
    (
        "non-rest-upstream",
        re.compile(r"\bnon[-_ ]rest[-_ ]upstream\b", re.IGNORECASE),
    ),
    (
        "distributed-load-targets",
        re.compile(r"\barbitrary[-_ ]distributed[-_ ]load[-_ ]targets\b", re.IGNORECASE),
    ),
]

MAX_PATH_LENGTH = 1024
MAX_FINDINGS = 1000
MAX_REPORT_BYTES = 65_536
BINARY_PROBE_BYTES = 4096
MAX_LINE_CHARACTERS = 65_536
MAX_TEXT_FILE_BYTES = 16 * 1024 * 1024
MAX_TOTAL_TEXT_BYTES = 256 * 1024 * 1024
MAX_FILES = 100_000
MAX_FILE_LIST_BYTES = 16 * 1024 * 1024


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Fail if Community repository contains out-of-scope product markers."
    )
    parser.add_argument(
        "--root",
        type=Path,
        default=Path(__file__).resolve().parents[1],
        help="Repository root. Defaults to parent of scripts/.",
    )
    parser.add_argument(
        "--files",
        nargs="*",
        help="Optional explicit file list, relative to root. Defaults to git ls-files.",
    )
    return parser.parse_args()


def git_tracked_files(root: Path) -> list[str]:
    process = subprocess.Popen(
        ["git", "ls-files", "--cached", "-z"],
        cwd=root,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
    )
    if process.stdout is None:
        process.kill()
        process.wait()
        raise OSError("git file discovery has no output stream")
    output = process.stdout.read(MAX_FILE_LIST_BYTES + 1)
    if len(output) > MAX_FILE_LIST_BYTES:
        process.kill()
        process.wait()
        raise ValueError("tracked file list exceeds limit")
    return_code = process.wait()
    if return_code != 0:
        raise subprocess.CalledProcessError(return_code, process.args)
    discovered = [path.decode("utf-8") for path in output.split(b"\0") if path]
    # A tracked deletion has no content to inspect and is a valid worktree state.
    # Existing and broken symlink entries remain in the list so path validation
    # can reject them fail-closed.
    return [
        path
        for path in discovered
        if (root / path).exists() or (root / path).is_symlink()
    ]


def should_skip_path(path: str) -> bool:
    if path in DEFAULT_EXCLUDED_PATHS:
        return True
    return any(path.startswith(prefix) for prefix in DEFAULT_EXCLUDED_PREFIXES)


def is_binary(data: bytes) -> bool:
    return b"\0" in data[:BINARY_PROBE_BYTES]


def path_has_symlink(root: Path, logical: PurePosixPath) -> bool:
    current = root
    for part in logical.parts:
        current = current / part
        if current.is_symlink():
            return True
    return False


def resolve_scannable_file(root: Path, relative_path: str) -> Path | None:
    if (
        not relative_path
        or len(relative_path) > MAX_PATH_LENGTH
        or "\\" in relative_path
        or any(ord(character) < 32 or ord(character) == 127 for character in relative_path)
    ):
        return None
    logical = PurePosixPath(relative_path)
    if logical.is_absolute() or any(part in {"", ".", ".."} for part in logical.parts):
        return None
    if str(logical) != relative_path or path_has_symlink(root, logical):
        return None
    candidate = root.joinpath(*logical.parts)
    try:
        resolved = candidate.resolve(strict=True)
        resolved.relative_to(root)
    except (FileNotFoundError, OSError, RuntimeError, ValueError):
        return None
    return resolved if resolved.is_file() else None


def normalized_identifier_text(line: str) -> str:
    with_word_boundaries = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", line)
    return re.sub(r"(?<=[A-Z])(?=[A-Z][a-z])", "_", with_word_boundaries)


def scan_file(
    root: Path,
    relative_path: str,
    remaining_text_bytes: int = MAX_TOTAL_TEXT_BYTES,
) -> tuple[list[str], bool, bool, int]:
    path = resolve_scannable_file(root, relative_path)
    if path is None:
        return [], False, False, 0

    if should_skip_path(relative_path):
        return [], True, False, 0

    findings: list[str] = []
    scanned_bytes = 0
    try:
        with path.open("rb") as binary_file:
            if is_binary(binary_file.read(BINARY_PROBE_BYTES)):
                return [], True, False, 0
            file_size = os.fstat(binary_file.fileno()).st_size
            if file_size > MAX_TEXT_FILE_BYTES or file_size > remaining_text_bytes:
                return [], False, False, 0
            binary_file.seek(0)
            with io.TextIOWrapper(binary_file, encoding="utf-8", errors="strict") as text_file:
                line_no = 0
                while True:
                    line = text_file.readline(MAX_LINE_CHARACTERS + 1)
                    if not line:
                        break
                    line_no += 1
                    if len(line) > MAX_LINE_CHARACTERS:
                        return findings, False, False, scanned_bytes
                    if len(line) == MAX_LINE_CHARACTERS and not line.endswith("\n"):
                        if text_file.read(1):
                            return findings, False, False, scanned_bytes

                    scanned_bytes += len(line.encode("utf-8"))
                    if (
                        scanned_bytes > MAX_TEXT_FILE_BYTES
                        or scanned_bytes > remaining_text_bytes
                    ):
                        return findings, False, False, scanned_bytes

                    directive = ALLOW_DIRECTIVE.search(line)
                    allowed_markers = (
                        {label.lower() for label in directive.group(1).split(",")}
                        if directive
                        else set()
                    )
                    normalized_line = normalized_identifier_text(line)
                    for label, pattern in FORBIDDEN_PATTERNS:
                        if label in allowed_markers:
                            continue
                        if pattern.search(line) or pattern.search(normalized_line):
                            if len(findings) >= MAX_FINDINGS:
                                return findings, True, True, scanned_bytes
                            findings.append(
                                f"{relative_path}:{line_no}: forbidden marker `{label}`"
                            )
    except (OSError, UnicodeDecodeError):
        return findings, False, False, scanned_bytes
    return findings, True, False, scanned_bytes


def render_failure(findings: list[str], invalid_count: int, input_truncated: bool) -> str:
    lines = ["Community scope check failed:"]
    selected_findings = findings[:MAX_FINDINGS]
    remaining = MAX_FINDINGS - len(selected_findings)
    selected_invalid = min(invalid_count, remaining)
    lines.extend(f"error: {finding}" for finding in selected_findings)
    lines.extend("error: invalid explicit file path" for _ in range(selected_invalid))
    if len(findings) + invalid_count > MAX_FINDINGS or input_truncated:
        lines.append("error: findings truncated")
    marker = "error: report truncated\n"
    report = "\n".join(lines) + "\n"
    if len(report.encode("utf-8")) <= MAX_REPORT_BYTES:
        return report
    budget = MAX_REPORT_BYTES - len(marker.encode("utf-8"))
    kept: list[str] = []
    used = 0
    for line in lines:
        encoded = (line + "\n").encode("utf-8")
        if used + len(encoded) > budget:
            break
        kept.append(line)
        used += len(encoded)
    return "\n".join(kept) + ("\n" if kept else "") + marker


def main() -> int:
    args = parse_args()
    root = args.root.resolve()
    try:
        files = args.files if args.files is not None else git_tracked_files(root)
    except (OSError, subprocess.SubprocessError, UnicodeDecodeError, ValueError):
        sys.stderr.write(render_failure([], 1, True))
        return 1
    findings: list[str] = []
    invalid_count = 0
    input_truncated = False
    scanned_text_bytes = 0

    if len(files) > MAX_FILES:
        sys.stderr.write(render_failure([], 1, True))
        return 1

    sorted_files = sorted(files)
    for index, relative_path in enumerate(sorted_files):
        file_findings, valid, file_truncated, file_bytes = scan_file(
            root,
            relative_path,
            MAX_TOTAL_TEXT_BYTES - scanned_text_bytes,
        )
        scanned_text_bytes += file_bytes
        if not valid:
            invalid_count += 1
        findings.extend(file_findings)
        if file_truncated:
            input_truncated = True
            break
        if len(findings) + invalid_count >= MAX_FINDINGS:
            input_truncated = index + 1 < len(sorted_files)
            break

    if findings or invalid_count:
        sys.stderr.write(render_failure(findings, invalid_count, input_truncated))
        return 1

    print(f"Community scope check passed ({len(files)} files scanned)")
    return 0


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