Merge pull request 'fix(gitops): add exact-path service policy for isolated writer' (#18) from patch/isolated-writer-exact-path-policy-20260725 into main

active-runtime/app/aios_isolated_writer_api.py

@@ -78,6 +78,7 @@ def _isolated_runtime_status(handler):
         "repository_identity": os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY"),
         "token_policy_version": 2,
         "atomic_commit_file": True,
+        "path_policy_sha256": policy.configured_path_policy_sha256(),
     })
 
 

active-runtime/app/aios_isolated_writer_policy.py

@@ -9,6 +9,8 @@ import re
 from pathlib import Path, PurePosixPath, PureWindowsPath
 from urllib.parse import unquote
 
+import isolated_writer_path_policy as path_policy
+
 WRITE_ENDPOINTS = frozenset({"/api/commit-file"})
 ALLOWED_ROOTS = (
     "active-runtime",
@@ -74,6 +76,7 @@ def require_service_invariants(workspace: Path, bare_repo: Path) -> None:
         raise PermissionError("isolated_writer_token_registry_required")
     if not os.getenv("AIOS_GITOPS_WRITER_OPERATION_LEDGER", "").strip():
         raise PermissionError("isolated_writer_operation_ledger_required")
+    path_policy.configured_path_policy()
     for name in _DENY_FLAGS:
         if os.getenv(name, "").strip().lower() not in _FALSE:
             raise PermissionError("isolated_writer_forbidden_capability_flag")
@@ -245,9 +248,14 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
     ):
         raise PermissionError("isolated_writer_path_rejected")
     first = parts[0]
+    service_exact_path = False
     if first in DENIED_ROOTS:
-        raise PermissionError("isolated_writer_protected_path_rejected")
-    if value != "README.md" and first not in ALLOWED_ROOTS:
+        try:
+            path_policy.require_service_path("commit_file.atomic", value)
+            service_exact_path = True
+        except path_policy.PathPolicyError as exc:
+            raise PermissionError(str(exc)) from None
+    if value != "README.md" and first not in ALLOWED_ROOTS and not service_exact_path:
         raise PermissionError("isolated_writer_path_outside_allowed_roots")
     if workspace is not None:
         lexical_root = Path(os.path.abspath(str(workspace)))
@@ -267,6 +275,22 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
     return value
 
 
+def validate_content_for_path(path: str, content: str) -> None:
+    if path.split("/", 1)[0] not in DENIED_ROOTS:
+        return
+    try:
+        path_policy.validate_content("commit_file.atomic", path, content)
+    except path_policy.PathPolicyError as exc:
+        raise PermissionError(str(exc)) from None
+
+
+def configured_path_policy_sha256() -> str | None:
+    try:
+        return path_policy.configured_path_policy_sha256()
+    except path_policy.PathPolicyError as exc:
+        raise PermissionError(str(exc)) from None
+
+
 def validate_commit_paths(paths: list[str], workspace: Path | None = None) -> None:
     if enforcement_required():
         for path in paths:

active-runtime/app/isolated_writer_commit_file.py

@@ -209,6 +209,10 @@ class AtomicCommitFile:
             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"]
@@ -378,6 +382,7 @@ class AtomicCommitFile:
             "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,

active-runtime/app/isolated_writer_path_policy.py

@@ -0,0 +1,307 @@
+"""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")

docs/gitops/isolated-writer-path-policy-v1.md

@@ -0,0 +1,48 @@
+# Isolated Writer Exact-Path Policy V1
+
+This policy is an additional, service-owned authorization layer for
+`POST /api/commit-file`. It never expands a Token Policy V2 grant.
+
+An atomic write is permitted only when all of the following agree:
+
+1. socket peer, audience, operation, branch, base and exact path in the
+   Token Policy V2 grant;
+2. service audience, repository identity, branch, operation and exact path
+   in `AIOS_GITOPS_PATH_POLICY_FILE`;
+3. lexical path, resolved containment and symlink checks;
+4. the configured content-size and content-profile checks;
+5. operation digest, expected branch HEAD and durable operation ledger.
+
+Any failed layer returns a fail-closed policy error. Coarse `/api/file` and
+`/api/commit` routes remain forbidden.
+
+## File contract
+
+The configured path must be absolute. The parent directory must be
+`root:gitops 0750`; the regular, non-symlink file must be
+`root:gitops 0640`. JSON must use UTF-8 canonical encoding with sorted keys,
+compact separators and one trailing newline. Unknown fields, unsupported
+operations or profiles, duplicate paths, oversized policies and
+group/world-writable files are rejected during Writer startup.
+
+Example:
+
+```json
+{"allowed_branch":"patch/codex-review-test-worker-20260724","allowed_operations":{"commit_file.atomic":{"content_profile":"systemd-codex-review-worker-v1","exact_paths":["deploy/systemd/aios-codex-review-test-worker.service"],"max_content_bytes":32768}},"audience":"aios-codex-review-worker-writer.service","policy_id":"codex-review-worker-path-policy-v1","repository_identity":"aios/aios","schema_version":1}
+```
+
+The SHA-256 of the canonical policy is exposed in the isolated runtime
+status and written to the operation audit record. The request body cannot
+select or override the policy file.
+
+## Content profile
+
+`systemd-codex-review-worker-v1` requires the fixed Python module
+`task_plane.workers.codex_review_test`, Shadow execution, `NoNewPrivileges`,
+`ProtectSystem=strict` and `PrivateTmp`. It rejects shell execution, root
+user declarations, worktree paths, authorization material and plaintext
+credential markers.
+
+Policy changes require a separate maintenance branch and Owner-reviewed
+ordinary merge commit. A task-scoped Writer must not edit its own policy
+file.

tests/test_isolated_writer_exact_path_policy.py

@@ -0,0 +1,259 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest import mock
+
+import aios_isolated_writer_policy as writer_policy
+import isolated_writer_path_policy as path_policy
+from isolated_writer_authz import AuthorizationError, _validate_grant
+
+
+TARGET = "deploy/systemd/aios-codex-review-test-worker.service"
+BRANCH = "patch/codex-review-test-worker-20260724"
+AUDIENCE = "aios-codex-review-worker-writer.service"
+
+VALID_UNIT = """\
+[Unit]
+Description=AIOS Codex Review/Test Worker
+
+[Service]
+User=aiman
+Environment=AIOS_CODEX_WORKER_MODE=shadow
+Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test
+NoNewPrivileges=true
+ProtectSystem=strict
+PrivateTmp=true
+"""
+
+
+def document(*, paths=None, **updates):
+    value = {
+        "schema_version": 1,
+        "policy_id": "codex-review-worker-path-policy-v1",
+        "audience": AUDIENCE,
+        "repository_identity": "aios/aios",
+        "allowed_branch": BRANCH,
+        "allowed_operations": {
+            "commit_file.atomic": {
+                "exact_paths": paths or [TARGET],
+                "max_content_bytes": 32768,
+                "content_profile": "systemd-codex-review-worker-v1",
+            }
+        },
+    }
+    value.update(updates)
+    return value
+
+
+class ExactPathPolicyTests(unittest.TestCase):
+    def setUp(self):
+        self.saved = dict(os.environ)
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.policy_dir = self.root / "policies"
+        self.policy_dir.mkdir(mode=0o750)
+        self.policy_file = self.policy_dir / "writer.json"
+        self.workspace = self.root / "workspace"
+        self.workspace.mkdir(mode=0o700)
+        os.environ.update(
+            {
+                "AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
+                "AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
+                "AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios",
+                "AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
+            }
+        )
+
+    def tearDown(self):
+        path_policy.clear_path_policy_cache()
+        os.environ.clear()
+        os.environ.update(self.saved)
+        self.temp.cleanup()
+
+    def write_policy(self, value=None):
+        payload = document() if value is None else value
+        raw = (
+            json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+            + "\n"
+        )
+        self.policy_file.write_text(raw, encoding="utf-8")
+        self.policy_file.chmod(0o640)
+        return path_policy.load_path_policy(
+            self.policy_file,
+            expected_owner_uid=os.getuid(),
+            expected_group_gid=os.getgid(),
+        )
+
+    def configured(self, loaded):
+        return mock.patch.object(
+            path_policy,
+            "configured_path_policy",
+            return_value=loaded,
+        )
+
+    def test_exact_systemd_path_and_exact_token_path_are_allowed(self):
+        loaded = self.write_policy()
+        with self.configured(loaded):
+            self.assertEqual(TARGET, writer_policy.validate_path(TARGET, self.workspace))
+            writer_policy.validate_content_for_path(TARGET, VALID_UNIT)
+
+    def test_token_path_cannot_expand_service_policy(self):
+        loaded = self.write_policy(document(paths=["deploy/systemd/other.service"]))
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_service_path_rejected"
+        ):
+            writer_policy.validate_path(TARGET, self.workspace)
+
+    def test_service_policy_cannot_expand_token_path(self):
+        now = datetime.now(timezone.utc)
+        grant = {
+            "schema_version": 2,
+            "audience": AUDIENCE,
+            "source_cidrs": ["127.0.0.1/32"],
+            "created_at": (now - timedelta(seconds=5)).isoformat(),
+            "not_before": (now - timedelta(seconds=2)).isoformat(),
+            "expires_at": (now + timedelta(minutes=5)).isoformat(),
+            "revoked": False,
+            "allowed_operations": ["commit_file.atomic"],
+            "allowed_branch": BRANCH,
+            "allowed_paths": ["deploy/systemd/other.service"],
+            "approved_base_oid": "a" * 40,
+        }
+        with mock.patch(
+            "isolated_writer_authz._git_is_ancestor", return_value=True
+        ), self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
+            _validate_grant(
+                grant,
+                audience=AUDIENCE,
+                source_ip="127.0.0.1",
+                operation="commit_file.atomic",
+                branch=BRANCH,
+                path=TARGET,
+                workspace=self.workspace,
+                current_time=now,
+            )
+
+    def test_other_service_file_is_rejected(self):
+        loaded = self.write_policy()
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_service_path_rejected"
+        ):
+            writer_policy.validate_path(
+                "deploy/systemd/other.service", self.workspace
+            )
+
+    def test_path_traversal_is_rejected(self):
+        for value in ("deploy/systemd/../x", "../deploy/systemd/x"):
+            with self.subTest(value=value), self.assertRaises(PermissionError):
+                writer_policy.validate_path(value, self.workspace)
+
+    def test_url_encoded_traversal_is_rejected(self):
+        with self.assertRaises(PermissionError):
+            writer_policy.validate_path(
+                "deploy/systemd/%2e%2e/x", self.workspace
+            )
+
+    def test_symlink_policy_file_is_rejected(self):
+        self.write_policy()
+        link = self.policy_dir / "link.json"
+        link.symlink_to(self.policy_file)
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError, "isolated_writer_path_policy_symlink_rejected"
+        ):
+            path_policy.load_path_policy(
+                link,
+                expected_owner_uid=os.getuid(),
+                expected_group_gid=os.getgid(),
+            )
+
+    def test_world_writable_policy_file_is_rejected(self):
+        self.write_policy()
+        self.policy_file.chmod(0o646)
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_permissions_rejected",
+        ):
+            path_policy.load_path_policy(
+                self.policy_file,
+                expected_owner_uid=os.getuid(),
+                expected_group_gid=os.getgid(),
+            )
+
+    def test_unknown_policy_field_is_rejected(self):
+        value = document()
+        value["unexpected"] = True
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_fields_rejected",
+        ):
+            self.write_policy(value)
+
+    def test_audience_mismatch_is_rejected(self):
+        loaded = self.write_policy()
+        os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
+        os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "other.service"
+        path_policy.clear_path_policy_cache()
+        with mock.patch.object(
+            path_policy, "load_path_policy", return_value=loaded
+        ), self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_audience_rejected",
+        ):
+            path_policy.configured_path_policy()
+
+    def test_branch_mismatch_is_rejected(self):
+        loaded = self.write_policy()
+        os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
+        os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/other"
+        path_policy.clear_path_policy_cache()
+        with mock.patch.object(
+            path_policy, "load_path_policy", return_value=loaded
+        ), self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_branch_rejected",
+        ):
+            path_policy.configured_path_policy()
+
+    def test_content_profile_rejects_shell_or_production_execution(self):
+        loaded = self.write_policy()
+        bad_values = (
+            VALID_UNIT.replace(
+                "/usr/bin/python3 -m task_plane.workers.codex_review_test",
+                "/bin/bash -c evil",
+            ),
+            VALID_UNIT.replace(
+                "AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
+                "AIOS_TASK_PLANE_EXECUTION_ENABLED=true",
+            ),
+            VALID_UNIT + "Environment=Authorization: Bearer placeholder\n",
+        )
+        with self.configured(loaded):
+            for value in bad_values:
+                with self.subTest(value=value), self.assertRaises(PermissionError):
+                    writer_policy.validate_content_for_path(TARGET, value)
+
+    def test_writer_request_cannot_modify_policy_file(self):
+        with self.assertRaises(PermissionError):
+            writer_policy.validate_path(
+                "etc/aios-gitops/writer-path-policies/codex-review-worker.json",
+                self.workspace,
+            )
+
+    def test_coarse_file_and_commit_endpoints_remain_forbidden(self):
+        for endpoint in ("/api/file", "/api/commit"):
+            with self.subTest(endpoint=endpoint), self.assertRaisesRegex(
+                PermissionError, "isolated_writer_operation_forbidden"
+            ):
+                writer_policy.validate_endpoint(
+                    "POST", endpoint, self.workspace, self.root
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()