Reject canonical escapes in private storage policies

src/agent_comms/storage_rules.py

@@ -11,14 +11,15 @@
 from functools import cached_property
 from pathlib import Path
 from subprocess import TimeoutExpired
-from typing import TYPE_CHECKING, ClassVar, Self
+from typing import TYPE_CHECKING, Annotated, ClassVar, Self
 
 from agent_comms.nominal.declared_family import DeclaredFamily
 from agent_comms.child_process import ResidentMemory
 from agent_comms.nominal.field_codec import (
     FieldCodec,
     GoalId,
     MalformedRecord,
+    ValueDirective,
 )
 from agent_comms.locked_store import DurableFile
 from agent_comms.errors import TodoConflict, TodoError
@@ -310,6 +311,16 @@ def owns_declared(self, rule: StorageRule, owner: Thread, path: Path,
         return path == rule.repository_checkout(rule.roots_of(owner), self.goal, repository)
 
 
+class ScopedDirectory(ValueDirective[Path]):
+    """A declared relative directory whose physical target must remain below its root."""
+
+    def within(self, root: Path) -> Path:
+        path = root / self.value
+        if not path.resolve().is_relative_to(root.resolve()):
+            raise ValueError(f"Storage directory {path} escapes its root {root}")
+        return path
+
+
 @dataclass(frozen=True)
 class StorageRule(RoomAllocation, DurableFile):
     filename: ClassVar[str] = "storage.json"
@@ -321,14 +332,15 @@ class StorageRule(RoomAllocation, DurableFile):
     cleanup_gib: float = 30.0  # under this a cleanup owner is due; it does not stop builds
     memory_gib: float = 3.0  # available memory under this makes a cleanup owner due
     idle_minutes: float = 30.0  # a pytest temp dir untouched this long holds memory for nothing
-    temp_root: StoragePath = Path('/tmp')  # RAM-backed, so its pytest dirs hold memory
+    temp_root: Annotated[StoragePath, ScopedDirectory(Path('tmp'))] = Path('/tmp')
     proc: StoragePath = Path('/proc')
     shown: ClassVar[int] = 5  # holders named in each list
-    # Home paths are read when the rule is built, never frozen at import.
-    volume: StoragePath = field(default_factory=Path.home)
-    scratch_base: StoragePath = field(
+    # Defaults use the current home; each declaration also owns its private-root location.
+    volume: Annotated[StoragePath, ScopedDirectory(Path())] = field(default_factory=Path.home)
+    scratch_base: Annotated[StoragePath, ScopedDirectory(Path('scratch'))] = field(
         default_factory=lambda: Path.home() / '.cache' / 'agent-scratch')
-    checkout_base: StoragePath = field(default_factory=lambda: Path.home() / 'wt')
+    checkout_base: Annotated[StoragePath, ScopedDirectory(Path('checkouts'))] = field(
+        default_factory=lambda: Path.home() / 'wt')
     # Joins a checkout's allocation, problem and companion repository: one name, one format.
     checkout_separator: ClassVar[str] = '--'
 
@@ -349,8 +361,16 @@ def load(cls, root: Path) -> Self:
     def within(self, root: Path) -> Self:
         """Keep allocation and disposable output inside a verification root, retaining limits."""
         root = root.resolve()
-        return replace(self, volume=root, scratch_base=root / "scratch",
-                       checkout_base=root / "checkouts", temp_root=root / "tmp")
+        return replace(self, **{
+            plan.name: directory.within(root)
+            for plan in FieldCodec.field_plans(type(self))
+            for directory in ScopedDirectory.single(plan.annotation)
+        })
+
+    def require_isolated(self, root: Path) -> None:
+        """Observe private scope without writing or accepting an escaping symlink."""
+        if self != self.within(root):
+            raise ValueError(f"Existing storage policy escapes verification root {root}")
 
     def prepare_isolated(self, root: Path) -> Self:
         """Persist private allocation before registration loads it; never rewrite an old policy."""
@@ -361,13 +381,12 @@ def prepare_isolated(self, root: Path) -> Self:
                 FieldCodec.compact(FieldCodec.encode(scoped)))
         except FileExistsError:
             stored = self.load(root)
-            if stored != stored.within(root):
-                raise ValueError(f"Existing storage policy escapes verification root {root}")
+            stored.require_isolated(root)
             return stored
         return scoped
 
     def initial_roots(self, name: str) -> StorageRoots:
-        return StorageRoots(self.scratch_base / name)
+        return StorageRoots(ScopedDirectory(Path(name)).within(self.scratch_base))
 
     def roots_of(self, thread: Thread) -> StorageRoots:
         return thread.storage_roots.assigned(self, thread.name)

tests/test_storage_rules.py

@@ -3,7 +3,9 @@
 import runpy
 import shutil
 import subprocess
+from dataclasses import dataclass
 from pathlib import Path
+from typing import Annotated
 
 import pytest
 
@@ -22,11 +24,13 @@
 )
 from agent_comms.standing_rules import StorageHomeMissing
 from agent_comms.storage_rules import (
+    ScopedDirectory,
     StorageProjection,
     StorageRule,
     UnitStorage,
 )
 from agent_comms.thread_storage import (
+    StoragePath,
     StorageRoots,
     UnassignedThreadStorage,
 )
@@ -40,6 +44,7 @@ def test_isolated_policy_is_durable_and_preserves_existing_private_limits(tmp_pa
     policy = configured.prepare_isolated(root)
     before = StorageRule.path_in(root).read_bytes()
     assert policy == configured.within(root) == StorageRule.load(root)
+    policy.require_isolated(root)
     assert StorageRule().prepare_isolated(root) == policy
     assert StorageRule.path_in(root).read_bytes() == before
     reopened = wire(root)
@@ -62,6 +67,83 @@ def test_isolated_preparation_refuses_an_existing_foreign_policy_without_rewriti
     assert not foreign.scratch_base.exists()
 
 
+@pytest.mark.parametrize("scoped_directory", [
+    pytest.param(directory, id=directory.value.as_posix())
+    for plan in FieldCodec.field_plans(StorageRule)
+    for directory in ScopedDirectory.single(plan.annotation)
+    if directory.value.parts
+])
+class TestIsolatedDirectories:
+    def test_projection_refuses_a_canonical_escape_before_writing(
+        self, tmp_path: Path, scoped_directory: ScopedDirectory,
+    ):
+        root = tmp_path / "verification"
+        root.mkdir()
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        (root / scoped_directory.value).symlink_to(outside, target_is_directory=True)
+        with pytest.raises(ValueError, match="escapes"):
+            StorageRule().within(root)
+        with pytest.raises(ValueError, match="escapes"):
+            StorageRule().prepare_isolated(root)
+        assert not StorageRule.path_in(root).exists()
+        assert tuple(outside.iterdir()) == ()
+
+    def test_existing_policy_rejects_a_new_escape_without_rewriting(
+        self, tmp_path: Path, scoped_directory: ScopedDirectory,
+    ):
+        root = tmp_path / "verification"
+        policy = StorageRule(floor_gib=0).prepare_isolated(root)
+        policy_path = StorageRule.path_in(root)
+        before = policy_path.read_bytes()
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        (root / scoped_directory.value).symlink_to(outside, target_is_directory=True)
+        loaded = StorageRule.load(root)
+        for current in (policy, loaded):
+            with pytest.raises(ValueError, match="escapes"):
+                current.require_isolated(root)
+        with pytest.raises(ValueError, match="escapes"):
+            StorageRule().prepare_isolated(root)
+        assert policy_path.read_bytes() == before
+        assert tuple(outside.iterdir()) == ()
+
+    def test_inward_links_preserve_private_scope(
+        self, tmp_path: Path, scoped_directory: ScopedDirectory,
+    ):
+        root = tmp_path / "verification"
+        root.mkdir()
+        destination = root / "inward"
+        destination.mkdir()
+        (root / scoped_directory.value).symlink_to(destination, target_is_directory=True)
+        policy = StorageRule(floor_gib=0).prepare_isolated(root)
+        policy.require_isolated(root)
+        assert StorageRule.load(root) == policy
+        assert StorageRule().prepare_isolated(root) == policy
+
+
+def test_future_child_cannot_follow_an_escaping_scratch_link(tmp_path: Path):
+    policy = StorageRule(floor_gib=0).prepare_isolated(tmp_path / "verification")
+    policy.scratch_base.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (policy.scratch_base / "future-child").symlink_to(outside, target_is_directory=True)
+    with pytest.raises(ValueError, match="escapes"):
+        policy.initial_roots("future-child")
+    assert tuple(outside.iterdir()) == ()
+
+
+def test_private_projection_uses_new_field_declarations_without_another_roster(tmp_path: Path):
+    @dataclass(frozen=True)
+    class ExtendedStorageRule(StorageRule):
+        output: Annotated[StoragePath, ScopedDirectory(Path("output"))] = Path("/external")
+
+    scoped = ExtendedStorageRule(floor_gib=0).within(tmp_path)
+    assert scoped.output == tmp_path / "output"
+    scoped.require_isolated(tmp_path)
+    assert FieldCodec.decode(ExtendedStorageRule, FieldCodec.encode(scoped)) == scoped
+
+
 def test_resource_checker_uses_storage_allocation_floor(tmp_path):
     checker = runpy.run_path(str(Path(__file__).resolve().parents[1] / 'tools' / 'agent_resource_check.py'))
     floor = StorageRule.load(tmp_path).floor_gib