52 lines
2.1 KiB
Python
52 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail-closed policy for in-process Git mutation handlers."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Mapping
|
|
|
|
_TRUE = frozenset({"1", "true", "yes", "on"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GitWriteDecision:
|
|
allowed: bool
|
|
status_code: int
|
|
error: str
|
|
reason: str
|
|
workspace: str
|
|
|
|
def payload(self) -> dict[str, str]:
|
|
return {
|
|
"status": "allowed" if self.allowed else "blocked",
|
|
"error": self.error,
|
|
"reason": self.reason,
|
|
"workspace": self.workspace,
|
|
}
|
|
|
|
|
|
def _real(path: Path) -> Path:
|
|
return Path(os.path.realpath(os.path.abspath(str(path))))
|
|
|
|
|
|
def evaluate_git_write(workspace: Path | str, env: Mapping[str, str] | None = None) -> GitWriteDecision:
|
|
values = os.environ if env is None else env
|
|
lexical = Path(os.path.abspath(str(workspace)))
|
|
canonical = _real(lexical)
|
|
shown = str(lexical)
|
|
if str(values.get("AIOS_GITOPS_WRITES_ENABLED", "false")).strip().lower() not in _TRUE:
|
|
return GitWriteDecision(False, 423, "gitops_writes_disabled", "AIOS_GITOPS_WRITES_ENABLED is false", shown)
|
|
if lexical != canonical:
|
|
return GitWriteDecision(False, 409, "workspace_symlink_forbidden", "workspace must be a canonical non-symlink path", shown)
|
|
controller = _real(Path(values.get("AIOS_GITOPS_CONTROLLER_PATH", "/data/aios-gitops/worktrees/controller")))
|
|
role = str(values.get("AIOS_GITOPS_WORKSPACE_ROLE", "")).strip().lower()
|
|
if canonical == controller or role == "runtime-anchor":
|
|
return GitWriteDecision(False, 423, "runtime_anchor_read_only", "controller/runtime-anchor is read-only", str(canonical))
|
|
writer_root = _real(Path(values.get("AIOS_GITOPS_APPROVED_WRITER_ROOT", "/data/aios-gitops/worktrees")))
|
|
try:
|
|
canonical.relative_to(writer_root)
|
|
except ValueError:
|
|
return GitWriteDecision(False, 409, "workspace_not_approved_writer", "workspace is outside the approved writer root", str(canonical))
|
|
return GitWriteDecision(True, 200, "", "approved isolated writer", str(canonical)) |