308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""Strict service-level exact-path policy for isolated Git writers."""
|
|
from __future__ import annotations
|
|
|
|
import configparser
|
|
import grp
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
from dataclasses import dataclass
|
|
from functools import lru_cache
|
|
from pathlib import Path, PurePosixPath, PureWindowsPath
|
|
from typing import Any
|
|
from urllib.parse import unquote
|
|
|
|
|
|
MAX_POLICY_BYTES = 65_536
|
|
POLICY_FIELDS = frozenset(
|
|
{
|
|
"schema_version",
|
|
"policy_id",
|
|
"audience",
|
|
"repository_identity",
|
|
"allowed_branch",
|
|
"allowed_operations",
|
|
}
|
|
)
|
|
OPERATION_FIELDS = frozenset({"exact_paths", "max_content_bytes", "content_profile"})
|
|
SUPPORTED_OPERATIONS = frozenset({"commit_file.atomic"})
|
|
SUPPORTED_CONTENT_PROFILES = frozenset({"systemd-codex-review-worker-v1"})
|
|
_SECRET_RE = re.compile(
|
|
r"(?i)(authorization\s*:|bearer\s+|(?:token|password|secret)\s*[:=])"
|
|
)
|
|
|
|
|
|
class PathPolicyError(PermissionError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OperationPolicy:
|
|
exact_paths: tuple[str, ...]
|
|
max_content_bytes: int
|
|
content_profile: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServicePathPolicy:
|
|
schema_version: int
|
|
policy_id: str
|
|
audience: str
|
|
repository_identity: str
|
|
allowed_branch: str
|
|
allowed_operations: dict[str, OperationPolicy]
|
|
sha256: str
|
|
|
|
def operation(self, name: str) -> OperationPolicy:
|
|
try:
|
|
return self.allowed_operations[name]
|
|
except KeyError:
|
|
raise PathPolicyError("isolated_writer_service_operation_rejected") from None
|
|
|
|
def require_exact_path(self, operation: str, path: str) -> OperationPolicy:
|
|
selected = self.operation(operation)
|
|
if path not in selected.exact_paths:
|
|
raise PathPolicyError("isolated_writer_service_path_rejected")
|
|
return selected
|
|
|
|
|
|
def _canonical_json(value: Any) -> bytes:
|
|
return (
|
|
json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
+ "\n"
|
|
).encode("utf-8")
|
|
|
|
|
|
def validate_relative_path_syntax(path: str) -> str:
|
|
if not isinstance(path, str):
|
|
raise PathPolicyError("isolated_writer_path_rejected")
|
|
if not path or path != path.strip() or "\x00" in path or "\\" in path:
|
|
raise PathPolicyError("isolated_writer_path_rejected")
|
|
try:
|
|
decoded = unquote(path, errors="strict")
|
|
except (UnicodeDecodeError, ValueError):
|
|
raise PathPolicyError("isolated_writer_path_rejected") from None
|
|
if decoded != path:
|
|
raise PathPolicyError("isolated_writer_path_rejected")
|
|
posix = PurePosixPath(path)
|
|
windows = PureWindowsPath(path)
|
|
parts = path.split("/")
|
|
if (
|
|
posix.is_absolute()
|
|
or windows.is_absolute()
|
|
or bool(windows.drive)
|
|
or any(part in {"", ".", ".."} for part in parts)
|
|
):
|
|
raise PathPolicyError("isolated_writer_path_rejected")
|
|
return path
|
|
|
|
|
|
def _require_fields(value: dict[str, Any], allowed: frozenset[str], code: str) -> None:
|
|
if set(value) != allowed:
|
|
raise PathPolicyError(code)
|
|
|
|
|
|
def _parse_policy(document: Any, canonical: bytes) -> ServicePathPolicy:
|
|
if not isinstance(document, dict):
|
|
raise PathPolicyError("isolated_writer_path_policy_object_required")
|
|
_require_fields(document, POLICY_FIELDS, "isolated_writer_path_policy_fields_rejected")
|
|
if document["schema_version"] != 1:
|
|
raise PathPolicyError("isolated_writer_path_policy_schema_rejected")
|
|
text_fields = (
|
|
"policy_id",
|
|
"audience",
|
|
"repository_identity",
|
|
"allowed_branch",
|
|
)
|
|
if any(
|
|
not isinstance(document[field], str) or not document[field].strip()
|
|
for field in text_fields
|
|
):
|
|
raise PathPolicyError("isolated_writer_path_policy_value_rejected")
|
|
raw_operations = document["allowed_operations"]
|
|
if not isinstance(raw_operations, dict) or not raw_operations:
|
|
raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
|
|
if not set(raw_operations).issubset(SUPPORTED_OPERATIONS):
|
|
raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
|
|
operations: dict[str, OperationPolicy] = {}
|
|
for name, raw in raw_operations.items():
|
|
if not isinstance(raw, dict):
|
|
raise PathPolicyError("isolated_writer_path_policy_operation_rejected")
|
|
_require_fields(
|
|
raw,
|
|
OPERATION_FIELDS,
|
|
"isolated_writer_path_policy_operation_fields_rejected",
|
|
)
|
|
paths = raw["exact_paths"]
|
|
if (
|
|
not isinstance(paths, list)
|
|
or not paths
|
|
or any(not isinstance(item, str) for item in paths)
|
|
):
|
|
raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
|
|
validated = tuple(validate_relative_path_syntax(item) for item in paths)
|
|
if len(set(validated)) != len(validated):
|
|
raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
|
|
maximum = raw["max_content_bytes"]
|
|
if not isinstance(maximum, int) or isinstance(maximum, bool) or not 1 <= maximum <= 1_048_576:
|
|
raise PathPolicyError("isolated_writer_path_policy_size_rejected")
|
|
profile = raw["content_profile"]
|
|
if profile not in SUPPORTED_CONTENT_PROFILES:
|
|
raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
|
|
operations[name] = OperationPolicy(validated, maximum, profile)
|
|
return ServicePathPolicy(
|
|
schema_version=1,
|
|
policy_id=document["policy_id"],
|
|
audience=document["audience"],
|
|
repository_identity=document["repository_identity"],
|
|
allowed_branch=document["allowed_branch"],
|
|
allowed_operations=operations,
|
|
sha256=hashlib.sha256(canonical).hexdigest(),
|
|
)
|
|
|
|
|
|
def load_path_policy(
|
|
path: Path,
|
|
*,
|
|
expected_owner_uid: int = 0,
|
|
expected_group_gid: int | None = None,
|
|
expected_parent_mode: int = 0o750,
|
|
expected_file_mode: int = 0o640,
|
|
) -> ServicePathPolicy:
|
|
lexical = Path(os.path.abspath(str(path)))
|
|
if lexical.is_symlink():
|
|
raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
|
|
try:
|
|
parent_info = lexical.parent.stat(follow_symlinks=False)
|
|
file_info = lexical.stat(follow_symlinks=False)
|
|
except FileNotFoundError:
|
|
raise PathPolicyError("isolated_writer_path_policy_missing") from None
|
|
if stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode):
|
|
raise PathPolicyError("isolated_writer_path_policy_parent_rejected")
|
|
if stat.S_ISLNK(file_info.st_mode) or not stat.S_ISREG(file_info.st_mode):
|
|
raise PathPolicyError("isolated_writer_path_policy_file_rejected")
|
|
if lexical.resolve(strict=True) != lexical:
|
|
raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
|
|
if (
|
|
parent_info.st_uid != expected_owner_uid
|
|
or stat.S_IMODE(parent_info.st_mode) != expected_parent_mode
|
|
):
|
|
raise PathPolicyError("isolated_writer_path_policy_parent_permissions_rejected")
|
|
if (
|
|
file_info.st_uid != expected_owner_uid
|
|
or stat.S_IMODE(file_info.st_mode) != expected_file_mode
|
|
):
|
|
raise PathPolicyError("isolated_writer_path_policy_permissions_rejected")
|
|
if expected_group_gid is not None and (
|
|
parent_info.st_gid != expected_group_gid or file_info.st_gid != expected_group_gid
|
|
):
|
|
raise PathPolicyError("isolated_writer_path_policy_group_rejected")
|
|
if file_info.st_size <= 0 or file_info.st_size > MAX_POLICY_BYTES:
|
|
raise PathPolicyError("isolated_writer_path_policy_file_size_rejected")
|
|
raw = lexical.read_bytes()
|
|
try:
|
|
document = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
raise PathPolicyError("isolated_writer_path_policy_json_rejected") from None
|
|
canonical = _canonical_json(document)
|
|
if raw != canonical:
|
|
raise PathPolicyError("isolated_writer_path_policy_not_canonical")
|
|
return _parse_policy(document, canonical)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def configured_path_policy() -> ServicePathPolicy | None:
|
|
configured = os.getenv("AIOS_GITOPS_PATH_POLICY_FILE", "").strip()
|
|
if not configured:
|
|
return None
|
|
if not Path(configured).is_absolute():
|
|
raise PathPolicyError("isolated_writer_path_policy_path_rejected")
|
|
try:
|
|
gitops_gid = grp.getgrnam("gitops").gr_gid
|
|
except KeyError:
|
|
raise PathPolicyError("isolated_writer_path_policy_group_rejected") from None
|
|
loaded = load_path_policy(Path(configured), expected_group_gid=gitops_gid)
|
|
if loaded.audience != os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""):
|
|
raise PathPolicyError("isolated_writer_path_policy_audience_rejected")
|
|
if loaded.repository_identity != os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", ""):
|
|
raise PathPolicyError("isolated_writer_path_policy_repository_rejected")
|
|
if loaded.allowed_branch != os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", ""):
|
|
raise PathPolicyError("isolated_writer_path_policy_branch_rejected")
|
|
return loaded
|
|
|
|
|
|
def clear_path_policy_cache() -> None:
|
|
configured_path_policy.cache_clear()
|
|
|
|
|
|
def require_service_path(operation: str, path: str) -> OperationPolicy:
|
|
loaded = configured_path_policy()
|
|
if loaded is None:
|
|
raise PathPolicyError("isolated_writer_service_path_policy_required")
|
|
return loaded.require_exact_path(operation, path)
|
|
|
|
|
|
def configured_path_policy_sha256() -> str | None:
|
|
loaded = configured_path_policy()
|
|
return None if loaded is None else loaded.sha256
|
|
|
|
|
|
def validate_content(operation: str, path: str, content: str) -> None:
|
|
selected = require_service_path(operation, path)
|
|
encoded = content.encode("utf-8")
|
|
if len(encoded) > selected.max_content_bytes:
|
|
raise PathPolicyError("isolated_writer_service_content_too_large")
|
|
if selected.content_profile == "systemd-codex-review-worker-v1":
|
|
_validate_systemd_codex_review_worker(content)
|
|
return
|
|
raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
|
|
|
|
|
|
def _validate_systemd_codex_review_worker(content: str) -> None:
|
|
if "\x00" in content or _SECRET_RE.search(content):
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|
|
lowered = content.lower()
|
|
forbidden = (
|
|
"bash -c",
|
|
"sh -c",
|
|
"eval ",
|
|
"/bin/bash",
|
|
"/bin/sh",
|
|
"/data/aios-gitops/worktrees/",
|
|
)
|
|
if any(value in lowered for value in forbidden):
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|
|
parser = configparser.ConfigParser(interpolation=None, strict=False)
|
|
parser.optionxform = str
|
|
try:
|
|
parser.read_string(content)
|
|
except configparser.Error:
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
|
|
if "Service" not in parser:
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|
|
service = parser["Service"]
|
|
if service.get("ExecStart") != "/usr/bin/python3 -m task_plane.workers.codex_review_test":
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|
|
required = {
|
|
"NoNewPrivileges": "true",
|
|
"ProtectSystem": "strict",
|
|
"PrivateTmp": "true",
|
|
"AIOS_TASK_PLANE_EXECUTION_ENABLED": "false",
|
|
}
|
|
environments: dict[str, str] = {}
|
|
for line in content.splitlines():
|
|
if line.startswith("Environment="):
|
|
value = line.removeprefix("Environment=").strip().strip('"')
|
|
if "=" in value:
|
|
key, inner = value.split("=", 1)
|
|
environments[key] = inner
|
|
if environments.get("AIOS_TASK_PLANE_EXECUTION_ENABLED") != "false":
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|
|
for key in ("NoNewPrivileges", "ProtectSystem", "PrivateTmp"):
|
|
if service.get(key, "").lower() != required[key]:
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|
|
if service.get("User", "").strip().lower() == "root":
|
|
raise PathPolicyError("isolated_writer_systemd_profile_rejected")
|