#!/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())