feat(provenance): add controlled Git writer with operation policy

2026-07-21 09:46:40 +08:00
parent 819ac94c9f
commit fbd880cf8c
1 changed files with 363 additions and 0 deletions
@@ -0,0 +1,363 @@
+"""受 Run Provenance 约束的 Git 写入器。
+
+该模块不接受任意 Shell 字符串。所有 Git 操作都使用参数数组执行，在执行前
+通过 ``git_operation`` 路径授权；改变代码树的操作还必须通过 ``code_change``
+路径授权。标准输出和错误输出会脱敏并截断后写入结果对象。
+"""
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import uuid
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+
+from .provenance import (
+    GitOperation,
+    ProvenanceControlledWriter,
+    ProvenanceDenied,
+    digest_payload,
+    utc_now,
+)
+
+_BRANCH_PATTERN = re.compile(r"^(?![-./])(?!.*\.\.)(?!.*//)[A-Za-z0-9._/-]{1,240}$")
+_SECRET_URL = re.compile(r"(?i)(https?://)([^/@\s]+)@")
+_TOKENISH = re.compile(r"(?i)(token|password|secret|authorization)([=:]\s*)([^\s]+)")
+
+
+class ProvenanceGitError(RuntimeError):
+    """Git 操作被拒绝或执行失败。"""
+
+
+@dataclass(frozen=True)
+class GitExecutionContext:
+    provenance_id: str
+    run_id: str
+    tenant_id: str
+    task_id: str
+    agent_id: str
+    principal_id: str
+
+
+@dataclass(frozen=True)
+class GitCommandResult:
+    operation_id: str
+    operation: str
+    authorization_ids: Sequence[str]
+    command: Sequence[str]
+    exit_code: int
+    stdout: str
+    stderr: str
+    head_before: str | None
+    head_after: str | None
+    started_at: str
+    completed_at: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+class ProvenanceGitWriter:
+    """执行经过 Provenance 授权的窄范围 Git 操作。"""
+
+    def __init__(
+        self,
+        repo: str | Path,
+        writer: ProvenanceControlledWriter,
+        *,
+        timeout_seconds: float = 120.0,
+        output_limit: int = 16000,
+    ) -> None:
+        self.repo = Path(repo).resolve()
+        self.writer = writer
+        self.timeout_seconds = float(timeout_seconds)
+        self.output_limit = int(output_limit)
+        if self.timeout_seconds <= 0 or self.output_limit < 1024:
+            raise ValueError("invalid Git writer limits")
+        self._run_raw(["rev-parse", "--is-inside-work-tree"], check=True)
+
+    @staticmethod
+    def _validate_branch(branch: str) -> str:
+        normalized = str(branch or "").strip()
+        if not _BRANCH_PATTERN.fullmatch(normalized) or normalized.endswith((".", "/")):
+            raise ProvenanceGitError("invalid Git branch name")
+        if normalized in {"main", "master"}:
+            raise ProvenanceGitError("direct main/master operation is prohibited")
+        return normalized
+
+    def _validate_path(self, value: str | Path) -> str:
+        raw = Path(value)
+        candidate = (self.repo / raw).resolve() if not raw.is_absolute() else raw.resolve()
+        try:
+            relative = candidate.relative_to(self.repo)
+        except ValueError as exc:
+            raise ProvenanceGitError("Git path escapes repository") from exc
+        if str(relative) in {"", "."}:
+            raise ProvenanceGitError("repository root is not a valid file path")
+        return relative.as_posix()
+
+    @staticmethod
+    def _redact(value: str, limit: int) -> str:
+        text = _SECRET_URL.sub(r"\1***@", str(value or ""))
+        text = _TOKENISH.sub(r"\1\2***", text)
+        if len(text) > limit:
+            return text[:limit] + "\n...[truncated]"
+        return text
+
+    def _head(self) -> str | None:
+        process = self._run_raw(["rev-parse", "HEAD"], check=False)
+        return process.stdout.strip() if process.returncode == 0 else None
+
+    def _run_raw(
+        self,
+        args: Sequence[str],
+        *,
+        check: bool,
+        env: Mapping[str, str] | None = None,
+    ) -> subprocess.CompletedProcess[str]:
+        command = ["git", "-C", str(self.repo), *[str(item) for item in args]]
+        safe_env = {
+            **os.environ,
+            "GIT_TERMINAL_PROMPT": "0",
+            "GIT_CONFIG_NOSYSTEM": os.environ.get("GIT_CONFIG_NOSYSTEM", "1"),
+        }
+        if env:
+            safe_env.update({str(key): str(value) for key, value in env.items()})
+        process = subprocess.run(
+            command,
+            text=True,
+            capture_output=True,
+            timeout=self.timeout_seconds,
+            env=safe_env,
+        )
+        if check and process.returncode != 0:
+            raise ProvenanceGitError(
+                self._redact(process.stderr or process.stdout, self.output_limit)
+                or f"Git command failed with exit code {process.returncode}"
+            )
+        return process
+
+    def _authorize(
+        self,
+        context: GitExecutionContext,
+        *,
+        operation: GitOperation,
+        request: Mapping[str, Any],
+        code_change: bool,
+    ) -> list[str]:
+        request_digest = digest_payload(
+            {
+                "operation": operation.value,
+                "repo": self.repo.name,
+                "request": dict(request),
+                "head": self._head(),
+            }
+        )
+        ids: list[str] = []
+        try:
+            if code_change:
+                code_authorization = self.writer.authorize_code_change(
+                    provenance_id=context.provenance_id,
+                    run_id=context.run_id,
+                    tenant_id=context.tenant_id,
+                    task_id=context.task_id,
+                    agent_id=context.agent_id,
+                    principal_id=context.principal_id,
+                    request_digest=digest_payload(
+                        {
+                            "git_operation": operation.value,
+                            "request_digest": request_digest,
+                        }
+                    ),
+                )
+                ids.append(code_authorization.authorization_id)
+            git_authorization = self.writer.authorize_git_operation(
+                provenance_id=context.provenance_id,
+                run_id=context.run_id,
+                tenant_id=context.tenant_id,
+                task_id=context.task_id,
+                agent_id=context.agent_id,
+                principal_id=context.principal_id,
+                git_operation=operation.value,
+                request_digest=request_digest,
+            )
+            ids.append(git_authorization.authorization_id)
+            return ids
+        except ProvenanceDenied as exc:
+            raise ProvenanceGitError(f"Run Provenance denied Git operation: {exc}") from exc
+
+    def _execute(
+        self,
+        context: GitExecutionContext,
+        *,
+        operation: GitOperation,
+        args: Sequence[str],
+        request: Mapping[str, Any],
+        code_change: bool,
+    ) -> GitCommandResult:
+        authorization_ids = self._authorize(
+            context,
+            operation=operation,
+            request=request,
+            code_change=code_change,
+        )
+        started = utc_now()
+        before = self._head()
+        process = self._run_raw(args, check=False)
+        after = self._head()
+        result = GitCommandResult(
+            operation_id=f"gitop_{uuid.uuid4().hex}",
+            operation=operation.value,
+            authorization_ids=tuple(authorization_ids),
+            command=tuple(["git", "-C", "<repo>", *[str(item) for item in args]]),
+            exit_code=int(process.returncode),
+            stdout=self._redact(process.stdout, self.output_limit),
+            stderr=self._redact(process.stderr, self.output_limit),
+            head_before=before,
+            head_after=after,
+            started_at=started,
+            completed_at=utc_now(),
+        )
+        if process.returncode != 0:
+            raise ProvenanceGitError(
+                result.stderr or result.stdout or f"Git {operation.value} failed"
+            )
+        return result
+
+    def read_ref(self, context: GitExecutionContext, ref: str) -> GitCommandResult:
+        normalized = str(ref or "").strip()
+        if not normalized or any(ch.isspace() for ch in normalized):
+            raise ProvenanceGitError("invalid Git ref")
+        return self._execute(
+            context,
+            operation=GitOperation.READ_REF,
+            args=["rev-parse", "--verify", normalized],
+            request={"ref": normalized},
+            code_change=False,
+        )
+
+    def fetch(
+        self,
+        context: GitExecutionContext,
+        *,
+        remote: str,
+        refspecs: Sequence[str],
+    ) -> GitCommandResult:
+        normalized_remote = str(remote or "").strip()
+        normalized_refspecs = tuple(str(item).strip() for item in refspecs if str(item).strip())
+        if not normalized_remote or not normalized_refspecs:
+            raise ProvenanceGitError("remote and refspecs are required")
+        if any(item.startswith("+") and ":refs/heads/" in item for item in normalized_refspecs):
+            raise ProvenanceGitError(
+                "fetching directly into local branch refs is prohibited; use remote-tracking refs"
+            )
+        return self._execute(
+            context,
+            operation=GitOperation.FETCH,
+            args=["fetch", "--prune", normalized_remote, *normalized_refspecs],
+            request={"remote": normalized_remote, "refspecs": normalized_refspecs},
+            code_change=False,
+        )
+
+    def create_branch(
+        self,
+        context: GitExecutionContext,
+        *,
+        branch: str,
+        start_point: str,
+    ) -> GitCommandResult:
+        normalized = self._validate_branch(branch)
+        start = str(start_point or "").strip()
+        if not start:
+            raise ProvenanceGitError("start_point is required")
+        return self._execute(
+            context,
+            operation=GitOperation.CREATE_BRANCH,
+            args=["branch", normalized, start],
+            request={"branch": normalized, "start_point": start},
+            code_change=False,
+        )
+
+    def commit(
+        self,
+        context: GitExecutionContext,
+        *,
+        paths: Sequence[str | Path],
+        message: str,
+    ) -> GitCommandResult:
+        normalized_paths = tuple(self._validate_path(item) for item in paths)
+        normalized_message = str(message or "").strip()
+        if not normalized_paths or not normalized_message:
+            raise ProvenanceGitError("commit paths and message are required")
+        if len(normalized_message) > 500:
+            raise ProvenanceGitError("commit message is too long")
+        self._run_raw(["add", "--", *normalized_paths], check=True)
+        return self._execute(
+            context,
+            operation=GitOperation.COMMIT,
+            args=["commit", "--no-gpg-sign", "-m", normalized_message, "--", *normalized_paths],
+            request={"paths": normalized_paths, "message": normalized_message},
+            code_change=True,
+        )
+
+    def merge(
+        self,
+        context: GitExecutionContext,
+        *,
+        source_ref: str,
+        message: str,
+    ) -> GitCommandResult:
+        source = str(source_ref or "").strip()
+        normalized_message = str(message or "").strip()
+        if not source or not normalized_message:
+            raise ProvenanceGitError("source_ref and message are required")
+        return self._execute(
+            context,
+            operation=GitOperation.MERGE,
+            args=["merge", "--no-ff", "--no-edit", "-m", normalized_message, source],
+            request={"source_ref": source, "message": normalized_message},
+            code_change=True,
+        )
+
+    def push(
+        self,
+        context: GitExecutionContext,
+        *,
+        remote: str,
+        branch: str,
+    ) -> GitCommandResult:
+        normalized_remote = str(remote or "").strip()
+        normalized_branch = self._validate_branch(branch)
+        if not normalized_remote:
+            raise ProvenanceGitError("remote is required")
+        return self._execute(
+            context,
+            operation=GitOperation.PUSH,
+            args=["push", normalized_remote, f"refs/heads/{normalized_branch}:refs/heads/{normalized_branch}"],
+            request={"remote": normalized_remote, "branch": normalized_branch},
+            code_change=False,
+        )
+
+    def update_evidence(
+        self,
+        context: GitExecutionContext,
+        *,
+        paths: Sequence[str | Path],
+        message: str,
+    ) -> GitCommandResult:
+        normalized_paths = tuple(self._validate_path(item) for item in paths)
+        normalized_message = str(message or "").strip()
+        if not normalized_paths or not normalized_message:
+            raise ProvenanceGitError("evidence paths and message are required")
+        if any(not path.startswith(".agent/") for path in normalized_paths):
+            raise ProvenanceGitError("UPDATE_EVIDENCE may only commit .agent/ paths")
+        self._run_raw(["add", "--", *normalized_paths], check=True)
+        return self._execute(
+            context,
+            operation=GitOperation.UPDATE_EVIDENCE,
+            args=["commit", "--no-gpg-sign", "-m", normalized_message, "--", *normalized_paths],
+            request={"paths": normalized_paths, "message": normalized_message},
+            code_change=False,
+        )