#!/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))