feat(gitops): add idempotent atomic commit-file endpoint

2026-07-24 22:56:58 +08:00
parent bbada7c863
commit 7ecf7cbfce
3 changed files with 573 additions and 0 deletions
@@ -5,13 +5,146 @@ import json
 import os
 import subprocess
 from pathlib import Path
+import urllib.parse

 import aios_chatgpt_actions_api as api
 import aios_isolated_writer_policy as policy
+from isolated_writer_authz import AuthorizationError, authorize
+from isolated_writer_commit_file import AtomicCommitFile, CommitFileError, OperationLedger
+from isolated_writer_token_registry import TokenRegistry

 _reject = api.reject_non_read_http
 _write_file = api.write_file
 _sync_branch = api.sync_branch
+_original_do_get = api.H.do_GET
+
+
+def _registry() -> TokenRegistry:
+    return TokenRegistry(Path(os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"]))
+
+
+def _atomic_writer() -> AtomicCommitFile:
+    return AtomicCommitFile(
+        workspace=api.WS,
+        bare_repo=api.BARE_REPO,
+        ledger=OperationLedger(Path(os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"])),
+        remote=os.getenv("AIOS_GITOPS_GITEA_PUSH_REMOTE", ""),
+        audit=policy.append_audit_record,
+    )
+
+
+def _bearer(handler) -> str:
+    value = handler.headers.get("Authorization", "")
+    return value[7:].strip() if value.startswith("Bearer ") else ""
+
+
+def _authorize(handler, *, method: str, path: str, branch=None, target_path=None):
+    grant = authorize(
+        _registry(),
+        _bearer(handler),
+        method=method,
+        path=path,
+        source_ip=str(handler.client_address[0]),
+        workspace=api.WS,
+        branch=branch,
+        target_path=target_path,
+    )
+    grant.record["_source_ip"] = str(handler.client_address[0])
+    return grant
+
+
+def _authorization_error(handler, exc: AuthorizationError):
+    return handler.sendj({"error": exc.code}, exc.status)
+
+
+def _isolated_runtime_status(handler):
+    branch = api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"]
+    head = api.sh(["rev-parse", "HEAD"], True)["stdout"]
+    return handler.sendj({
+        "service": os.getenv(
+            "AIOS_GITOPS_SERVICE_AUDIENCE",
+            "aios-isolated-writer.service",
+        ),
+        "status": "active",
+        "runtime_role": os.getenv("AIOS_GITOPS_RUNTIME_ROLE"),
+        "workspace_role": os.getenv("AIOS_GITOPS_WORKSPACE_ROLE"),
+        "writes_enabled": os.getenv("AIOS_GITOPS_WRITES_ENABLED", "").lower()
+        in {"1", "true", "yes", "on"},
+        "allowed_branch": policy.allowed_branch(),
+        "approved_base_oid": policy.allowed_base_oid(),
+        "workspace_branch": branch,
+        "workspace_head_oid": head,
+        "repository_identity": os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY"),
+        "token_policy_version": 2,
+        "atomic_commit_file": True,
+    })
+
+
+def _isolated_do_get(self):
+    path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+    if path in {"/health", "/api/health"}:
+        return self.sendj({
+            "status": "ok",
+            "service": os.getenv(
+                "AIOS_GITOPS_SERVICE_AUDIENCE",
+                "aios-isolated-writer.service",
+            ),
+        })
+    if path not in {"/api/runtime-status", "/api/baseline", "/api/patches"}:
+        return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
+    try:
+        _authorize(self, method="GET", path=path)
+    except AuthorizationError as exc:
+        return _authorization_error(self, exc)
+    if path == "/api/runtime-status":
+        return _isolated_runtime_status(self)
+    return _original_do_get(self)
+
+
+def _read_commit_file_body(handler):
+    try:
+        length = int(handler.headers.get("Content-Length", "0"))
+    except ValueError:
+        raise CommitFileError("writer_content_length_rejected") from None
+    if length <= 0:
+        raise CommitFileError("writer_request_object_required")
+    if length > 1_100_000:
+        raise CommitFileError("writer_request_too_large", 413)
+    raw = handler.rfile.read(length)
+    try:
+        value = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise CommitFileError("writer_request_json_rejected") from None
+    if not isinstance(value, dict):
+        raise CommitFileError("writer_request_object_required")
+    return value
+
+
+def _isolated_do_post(self):
+    path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+    if path != "/api/commit-file":
+        return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
+    try:
+        data = _read_commit_file_body(self)
+        branch = str(data.get("branch") or "")
+        target_path = str(data.get("path") or "")
+        grant = _authorize(
+            self,
+            method="POST",
+            path=path,
+            branch=branch,
+            target_path=target_path,
+        )
+        if api.content_has_secret(str(data.get("content") or "")):
+            raise CommitFileError("writer_secret_content_rejected", 403)
+        result = _atomic_writer().execute(grant, data)
+        return self.sendj(result)
+    except AuthorizationError as exc:
+        return _authorization_error(self, exc)
+    except CommitFileError as exc:
+        return self.sendj({"error": exc.code}, exc.status)
+    except PermissionError:
+        return self.sendj({"error": "isolated_writer_policy_rejected"}, 403)


 def reject_non_read_http(handler, method: str, path: str) -> bool:
@@ -173,6 +306,8 @@ api.H.branch, api.H.file, api.H.commit, api.H.patch = (
    _checked_commit,
    _checked_patch,
 )
+api.H.do_GET = _isolated_do_get
+api.H.do_POST = _isolated_do_post

 if __name__ == "__main__":
    policy.require_service_invariants(api.WS, api.BARE_REPO)
@@ -0,0 +1,394 @@
+"""Atomic, idempotent commit-file Saga for the isolated writer."""
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import subprocess
+import tempfile
+import threading
+from pathlib import Path
+from typing import Any
+
+import aios_isolated_writer_policy as policy
+
+
+OID_RE = re.compile(r"[0-9a-f]{40}")
+IDEMPOTENCY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+STATES = frozenset({
+    "PENDING", "FILE_WRITTEN", "COMMITTED", "PUSHED", "VERIFIED", "COMPLETED",
+    "FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED",
+})
+_LOCKS: dict[str, threading.Lock] = {}
+_LOCKS_GUARD = threading.Lock()
+_LEDGER_INIT_LOCK = threading.Lock()
+
+
+class CommitFileError(RuntimeError):
+    def __init__(self, code: str, status: int = 400):
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+class SimulatedCrash(RuntimeError):
+    pass
+
+
+def canonical_digest(data: dict[str, Any]) -> str:
+    payload = {
+        "branch": data["branch"],
+        "path": data["path"],
+        "content_sha256": hashlib.sha256(data["content"].encode("utf-8")).hexdigest(),
+        "message": data["message"],
+        "expected_head_oid": data["expected_head_oid"],
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+class OperationLedger:
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        if self.path.parent.is_symlink() or self.path.is_symlink():
+            raise PermissionError("writer_operation_ledger_symlink_rejected")
+        with _LEDGER_INIT_LOCK:
+            with self.connect() as db:
+                db.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS writer_operations (
+                        operation_id TEXT PRIMARY KEY,
+                        token_id TEXT NOT NULL,
+                        task_id TEXT NOT NULL,
+                        idempotency_key TEXT NOT NULL UNIQUE,
+                        operation_digest TEXT NOT NULL,
+                        branch TEXT NOT NULL,
+                        path TEXT NOT NULL,
+                        content_sha256 TEXT NOT NULL,
+                        pre_head_oid TEXT NOT NULL,
+                        commit_oid TEXT,
+                        push_status TEXT,
+                        verification_status TEXT,
+                        state TEXT NOT NULL,
+                        created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        error_code TEXT
+                    )
+                    """
+                )
+        os.chmod(self.path, 0o600)
+
+    def connect(self) -> sqlite3.Connection:
+        db = sqlite3.connect(self.path, timeout=30)
+        db.row_factory = sqlite3.Row
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        return db
+
+    def get_or_create(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
+        key = data["idempotency_key"]
+        with self.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute(
+                "SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
+            ).fetchone()
+            if row:
+                db.commit()
+                record = dict(row)
+                if record["operation_digest"] != digest:
+                    raise CommitFileError("writer_idempotency_conflict", 409)
+                if (
+                    record["token_id"] != grant.token_id
+                    or record["task_id"] != grant.task_id
+                ):
+                    raise CommitFileError("writer_idempotency_owner_conflict", 409)
+                return record
+            operation_id = hashlib.sha256(
+                f"{grant.task_id}\0{key}".encode("utf-8")
+            ).hexdigest()
+            content_sha = hashlib.sha256(data["content"].encode("utf-8")).hexdigest()
+            db.execute(
+                """
+                INSERT INTO writer_operations(
+                    operation_id,token_id,task_id,idempotency_key,operation_digest,
+                    branch,path,content_sha256,pre_head_oid,state
+                ) VALUES(?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    operation_id, grant.token_id, grant.task_id, key, digest,
+                    data["branch"], data["path"], content_sha,
+                    data["expected_head_oid"], "PENDING",
+                ),
+            )
+            db.commit()
+        return self.get(key)
+
+    def get(self, key: str) -> dict[str, Any]:
+        with self.connect() as db:
+            row = db.execute(
+                "SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
+            ).fetchone()
+        if row is None:
+            raise KeyError(key)
+        return dict(row)
+
+    def update(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
+        if state not in STATES:
+            raise ValueError("invalid writer operation state")
+        allowed = {"commit_oid", "push_status", "verification_status", "error_code"}
+        assignments = ["state=?", "updated_at=CURRENT_TIMESTAMP"]
+        values: list[Any] = [state]
+        for name, value in fields.items():
+            if name not in allowed:
+                raise ValueError("invalid ledger field")
+            assignments.append(f"{name}=?")
+            values.append(value)
+        values.append(key)
+        with self.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            db.execute(
+                f"UPDATE writer_operations SET {','.join(assignments)} WHERE idempotency_key=?",
+                values,
+            )
+            db.commit()
+        return self.get(key)
+
+
+class AtomicCommitFile:
+    def __init__(
+        self,
+        *,
+        workspace: Path,
+        bare_repo: Path,
+        ledger: OperationLedger,
+        remote: str,
+        audit,
+        crash_after_state: str | None = None,
+    ):
+        self.workspace = workspace
+        self.bare_repo = bare_repo
+        self.ledger = ledger
+        self.remote = remote
+        self.audit = audit
+        self.crash_after_state = crash_after_state
+
+    def _git(self, *args: str, check: bool = True) -> str:
+        result = subprocess.run(
+            ["git", *args], cwd=self.workspace, text=True, capture_output=True
+        )
+        if check and result.returncode != 0:
+            raise CommitFileError("writer_git_operation_failed", 500)
+        return result.stdout.strip()
+
+    def _transition(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
+        record = self.ledger.update(key, state, **fields)
+        if self.crash_after_state == state:
+            raise SimulatedCrash(state)
+        return record
+
+    def _validate(self, data: Any) -> dict[str, Any]:
+        if not isinstance(data, dict):
+            raise CommitFileError("writer_request_object_required")
+        required = ("branch", "path", "content", "message", "expected_head_oid", "idempotency_key")
+        if any(field not in data for field in required):
+            raise CommitFileError("writer_request_incomplete")
+        item = {field: data[field] for field in required}
+        if item["branch"] != policy.allowed_branch():
+            raise CommitFileError("isolated_writer_branch_rejected", 403)
+        try:
+            item["path"] = policy.validate_path(item["path"], self.workspace)
+        except PermissionError as exc:
+            raise CommitFileError(str(exc), 403) from None
+        if not isinstance(item["content"], str) or "\x00" in item["content"]:
+            raise CommitFileError("writer_content_rejected")
+        if len(item["content"].encode("utf-8")) > 1_048_576:
+            raise CommitFileError("writer_content_too_large", 413)
+        message = item["message"]
+        if (
+            not isinstance(message, str)
+            or not 1 <= len(message) <= 200
+            or "\x00" in message
+            or "\n" in message
+            or "\r" in message
+        ):
+            raise CommitFileError("writer_message_rejected")
+        if not isinstance(item["expected_head_oid"], str) or not OID_RE.fullmatch(item["expected_head_oid"]):
+            raise CommitFileError("writer_expected_head_rejected")
+        if not isinstance(item["idempotency_key"], str) or not IDEMPOTENCY_RE.fullmatch(item["idempotency_key"]):
+            raise CommitFileError("writer_idempotency_key_rejected")
+        return item
+
+    def _atomic_write(self, relative: str, content: str) -> None:
+        target = self.workspace / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        fd, temporary = tempfile.mkstemp(prefix=".writer-", dir=target.parent)
+        try:
+            with os.fdopen(fd, "wb", closefd=True) as stream:
+                stream.write(content.encode("utf-8"))
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(temporary, target)
+            parent_fd = os.open(target.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+            try:
+                os.fsync(parent_fd)
+            finally:
+                os.close(parent_fd)
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
+
+    def _verify_remote(self, branch: str, oid: str) -> None:
+        bare = subprocess.run(
+            ["git", "--git-dir", str(self.bare_repo), "rev-parse", f"refs/heads/{branch}"],
+            text=True, capture_output=True,
+        )
+        if bare.returncode != 0 or bare.stdout.strip() != oid:
+            raise CommitFileError("writer_authoritative_bare_oid_mismatch", 500)
+        remote = subprocess.run(
+            ["git", "ls-remote", "--heads", self.remote, f"refs/heads/{branch}"],
+            cwd=self.workspace, text=True, capture_output=True,
+        )
+        if remote.returncode != 0:
+            raise CommitFileError("writer_gitea_verification_failed", 503)
+        fields = remote.stdout.strip().split()
+        if len(fields) != 2:
+            raise CommitFileError("writer_gitea_ref_missing", 503)
+        if fields[0] != oid:
+            raise CommitFileError("writer_gitea_oid_conflict", 409)
+
+    def execute(self, grant: Any, raw_data: Any) -> dict[str, Any]:
+        data = self._validate(raw_data)
+        digest = canonical_digest(data)
+        branch = data["branch"]
+        with _LOCKS_GUARD:
+            lock = _LOCKS.setdefault(branch, threading.Lock())
+        lock_path = self.ledger.path.parent / f"{hashlib.sha256(branch.encode()).hexdigest()}.lock"
+        with lock, open(lock_path, "a+b") as branch_lock:
+            fcntl.flock(branch_lock.fileno(), fcntl.LOCK_EX)
+            return self._execute_locked(grant, data, digest)
+
+    def _execute_locked(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
+        key = data["idempotency_key"]
+        record = self.ledger.get_or_create(grant, data, digest)
+        if record["state"] == "COMPLETED":
+            return self._result(record, idempotent=True)
+        if self._git("rev-parse", "--abbrev-ref", "HEAD") != data["branch"]:
+            record = self._transition(
+                key, "MANUAL_REVIEW_REQUIRED", error_code="writer_current_branch_conflict"
+            )
+            self._audit(grant, data, digest, record, "manual_review_required")
+            raise CommitFileError("writer_current_branch_conflict", 409)
+        head = self._git("rev-parse", "HEAD")
+        commit_oid = record.get("commit_oid")
+        if head != record["pre_head_oid"] and head != commit_oid:
+            record = self._transition(
+                key, "MANUAL_REVIEW_REQUIRED", error_code="writer_branch_head_conflict"
+            )
+            self._audit(grant, data, digest, record, "manual_review_required")
+            raise CommitFileError("writer_branch_head_conflict", 409)
+        try:
+            state = record["state"]
+            if state == "MANUAL_REVIEW_REQUIRED":
+                raise CommitFileError("writer_manual_review_required", 409)
+            if state == "FAILED_RECOVERABLE":
+                if record.get("push_status") == "confirmed":
+                    state = "PUSHED"
+                else:
+                    state = "COMMITTED" if commit_oid else "FILE_WRITTEN"
+            if state == "PENDING":
+                self._atomic_write(data["path"], data["content"])
+                record = self._transition(key, "FILE_WRITTEN")
+                state = record["state"]
+            if state == "FILE_WRITTEN":
+                actual_sha = hashlib.sha256(
+                    (self.workspace / data["path"]).read_bytes()
+                ).hexdigest()
+                if actual_sha != record["content_sha256"]:
+                    raise CommitFileError("writer_file_content_conflict", 409)
+                staged_before = self._git("diff", "--cached", "--name-only", check=False).splitlines()
+                if staged_before:
+                    raise CommitFileError("writer_extra_staged_files_rejected", 409)
+                self._git("add", "--", data["path"])
+                staged = self._git("diff", "--cached", "--name-only").splitlines()
+                if staged != [data["path"]]:
+                    raise CommitFileError("writer_extra_staged_files_rejected", 409)
+                self._git("commit", "-m", data["message"])
+                commit_oid = self._git("rev-parse", "HEAD")
+                record = self._transition(key, "COMMITTED", commit_oid=commit_oid)
+                state = record["state"]
+            if state == "COMMITTED":
+                commit_oid = str(record["commit_oid"])
+                pushed = subprocess.run(
+                    ["git", "push", "--porcelain", self.remote, f"HEAD:refs/heads/{data['branch']}"],
+                    cwd=self.workspace, text=True, capture_output=True,
+                )
+                if pushed.returncode != 0:
+                    self._transition(
+                        key, "FAILED_RECOVERABLE",
+                        push_status="failed", error_code="writer_gitea_push_failed",
+                    )
+                    raise CommitFileError("writer_gitea_push_failed", 503)
+                record = self._transition(key, "PUSHED", push_status="confirmed")
+                state = record["state"]
+            if state == "PUSHED":
+                commit_oid = str(record["commit_oid"])
+                self._verify_remote(data["branch"], commit_oid)
+                record = self._transition(
+                    key, "VERIFIED", verification_status="bare+gitea"
+                )
+                state = record["state"]
+            if state == "VERIFIED":
+                record = self._transition(key, "COMPLETED", error_code=None)
+            self._audit(grant, data, digest, record, "completed")
+            return self._result(record, idempotent=False)
+        except SimulatedCrash:
+            raise
+        except CommitFileError as exc:
+            latest = self.ledger.get(key)
+            if latest["state"] not in {"FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED"}:
+                failure_state = (
+                    "MANUAL_REVIEW_REQUIRED"
+                    if exc.code in {
+                        "writer_gitea_oid_conflict",
+                        "writer_file_content_conflict",
+                        "writer_current_branch_conflict",
+                    }
+                    else "FAILED_RECOVERABLE"
+                )
+                latest = self._transition(key, failure_state, error_code=exc.code)
+            self._audit(grant, data, digest, latest, "failed")
+            raise
+
+    def _audit(self, grant: Any, data: dict[str, Any], digest: str, record: dict[str, Any], result: str) -> None:
+        self.audit({
+            "token_id": grant.token_id,
+            "task_id": grant.task_id,
+            "audience": grant.record["audience"],
+            "source_ip": grant.record.get("_source_ip", "socket-peer"),
+            "operation": grant.operation,
+            "branch": data["branch"],
+            "path": data["path"],
+            "idempotency_key": data["idempotency_key"],
+            "operation_digest": digest,
+            "pre_head_oid": record["pre_head_oid"],
+            "commit_oid": record.get("commit_oid"),
+            "result": result,
+        })
+
+    @staticmethod
+    def _result(record: dict[str, Any], *, idempotent: bool) -> dict[str, Any]:
+        return {
+            "status": "completed",
+            "state": record["state"],
+            "operation_id": record["operation_id"],
+            "branch": record["branch"],
+            "path": record["path"],
+            "commit_oid": record.get("commit_oid"),
+            "push_status": record.get("push_status"),
+            "verification_status": record.get("verification_status"),
+            "idempotent": idempotent,
+        }
@@ -51,6 +51,31 @@ components:
        path: {type: string, example: docs/gitops/chatgpt-change.md}
        content: {type: string}
        message: {type: string}
+    AtomicFileCommitRequest:
+      type: object
+      additionalProperties: false
+      required: [branch, path, content, message, expected_head_oid, idempotency_key]
+      properties:
+        branch:
+          type: string
+          pattern: '^patch/[A-Za-z0-9._/-]+$'
+        path:
+          type: string
+          minLength: 1
+          maxLength: 512
+        content:
+          type: string
+          maxLength: 1048576
+        message:
+          type: string
+          minLength: 1
+          maxLength: 200
+        expected_head_oid:
+          type: string
+          pattern: '^[0-9a-f]{40}$'
+        idempotency_key:
+          type: string
+          pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$'
    CommitRequest:
      type: object
      properties:
@@ -428,6 +453,25 @@ paths:
            schema: {$ref: '#/components/schemas/FileCommitRequest'}
      responses: {'200': {description: File committed}, '400': {description: Secret rejected or invalid path}, '401': {description: Unauthorized}}
      x-openai-isConsequential: true
+  /api/commit-file:
+    post:
+      operationId: commitFileAtomic
+      summary: Atomically write one allowlisted file, commit, push, and verify using Token Policy V2
+      security: [{bearerAuth: []}]
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema: {$ref: '#/components/schemas/AtomicFileCommitRequest'}
+      responses:
+        '200': {description: Completed or idempotently replayed}
+        '400': {description: Invalid canonical request}
+        '401': {description: Missing, invalid, expired, or revoked Token Policy V2 grant}
+        '403': {description: Audience, source, branch, operation, path, or policy denied}
+        '409': {description: Idempotency conflict or manual review required}
+        '413': {description: Request too large}
+        '503': {description: Push or remote verification unavailable; recoverable}
+      x-openai-isConsequential: true
  /api/commit:
    post:
      operationId: commitChanges