404 lines
17 KiB
Python
404 lines
17 KiB
Python
"""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")
|
|
if data.get("force") or data.get("force_push"):
|
|
raise CommitFileError("isolated_writer_force_rejected", 403)
|
|
if set(data) != set(required):
|
|
raise CommitFileError("writer_request_fields_rejected")
|
|
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")
|
|
try:
|
|
policy.validate_content_for_path(item["path"], item["content"])
|
|
except PermissionError as exc:
|
|
raise CommitFileError(str(exc), 403) from None
|
|
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,
|
|
"path_policy_sha256": policy.configured_path_policy_sha256(),
|
|
"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,
|
|
}
|