guanghu-ice-heart/server-tools/zhulan-remote-cell/runtime/zhulan_code_gate.py

#!/usr/bin/env python3
"""铸澜工作区代码门：在 commit / candidate push 前校验能力与实际 Git 变化。"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any


SECRET_PATTERNS = [
    re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    re.compile(rb"(?i)(?:api[_-]?key|secret|password|token)\s*[:=]\s*['\"][^'\"\s]{12,}"),
    re.compile(rb"AKID[A-Z0-9]{13,}"),
]


def run_git(workspace: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", "-C", str(workspace), *args],
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        timeout=30,
    )
    return result.stdout.strip()


def within(path: str, prefixes: list[str]) -> bool:
    return any(prefix == "." or path == prefix or path.startswith(prefix + "/") for prefix in prefixes)


def verify_remote(url: str, token: str, action: str) -> dict[str, Any]:
    raw = json.dumps({"capability": token, "action": action}).encode("utf-8")
    request = urllib.request.Request(
        url.rstrip("/") + "/api/v1/runtime/verify",
        data=raw,
        headers={"Content-Type": "application/json", "User-Agent": "zhulan-code-gate/1"},
    )
    try:
        with urllib.request.urlopen(request, timeout=8) as response:
            return json.loads(response.read(262144))
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, ValueError) as exc:
        raise RuntimeError("capability_verify_failed") from exc


def changed_files(workspace: Path, base_sha: str) -> list[str]:
    run_git(workspace, "merge-base", "--is-ancestor", base_sha, "HEAD")
    committed = run_git(workspace, "diff", "--name-only", f"{base_sha}...HEAD")
    pending = run_git(workspace, "diff", "--name-only", "HEAD")
    untracked = run_git(workspace, "ls-files", "--others", "--exclude-standard")
    return sorted({line for block in (committed, pending, untracked) for line in block.splitlines() if line})


def scan_file(workspace: Path, path: Path, max_bytes: int) -> list[str]:
    errors: list[str] = []
    if path.is_symlink():
        target = path.resolve()
        try:
            target.relative_to(workspace)
        except ValueError:
            errors.append("symlink_outside_workspace")
        return errors
    if not path.is_file():
        return errors
    size = path.stat().st_size
    if size > max_bytes:
        return [f"file_too_large:{size}"]
    if size > 2_000_000:
        return errors
    raw = path.read_bytes()
    for pattern in SECRET_PATTERNS:
        if pattern.search(raw):
            errors.append("possible_secret")
            break
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--workspace", required=True)
    parser.add_argument("--capability-file", required=True)
    parser.add_argument("--action", choices=["edit", "test", "commit", "push_candidate"], required=True)
    parser.add_argument("--verify-url", default=os.getenv("ZHULAN_VERIFY_URL", "http://127.0.0.1:17631"))
    parser.add_argument("--max-file-bytes", type=int, default=20 * 1024 * 1024)
    args = parser.parse_args()

    workspace = Path(args.workspace).resolve()
    token = Path(args.capability_file).read_text(encoding="utf-8").strip()
    verified = verify_remote(args.verify_url, token, args.action)
    claims = verified["claims"]
    errors: list[dict[str, str]] = []

    if run_git(workspace, "rev-parse", "--is-inside-work-tree") != "true":
        errors.append({"gate": "git", "error": "not_a_worktree"})
    branch = run_git(workspace, "branch", "--show-current")
    if branch != claims["branch"]:
        errors.append({"gate": "branch", "error": f"expected:{claims['branch']}:actual:{branch}"})

    try:
        files = changed_files(workspace, claims["base_sha"])
    except subprocess.CalledProcessError:
        files = []
        errors.append({"gate": "base_sha", "error": "base_not_ancestor"})

    for name in files:
        if not within(name, claims["paths"]):
            errors.append({"gate": "path", "error": name})
            continue
        for finding in scan_file(workspace, workspace / name, args.max_file_bytes):
            errors.append({"gate": finding, "error": name})

    result = {
        "schema": "guanghu.zhulan-code-gate-receipt/v1",
        "decision": "PASS" if not errors else "FAIL",
        "request_id": claims["request_id"],
        "development_id": claims["development_id"],
        "repository": claims["repository"],
        "branch": branch,
        "base_sha": claims["base_sha"],
        "changed_files": files,
        "errors": errors,
        "workspace_fingerprint": hashlib.sha256(str(workspace).encode()).hexdigest(),
        "capability_receipt": verified["receipt"],
    }
    print(json.dumps(result, ensure_ascii=False, sort_keys=True, indent=2))
    return 0 if not errors else 2


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