140 lines
5 KiB
Python
140 lines
5 KiB
Python
#!/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())
|