From 0c6633f85f408b15adcb65976f1fc7f70928975f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sun, 1 Nov 2026 04:09:18 -0500 Subject: [PATCH] Reject canonical escapes through declared storage directories --- src/agent_comms/storage_rules.py | 41 +++++++++++----- tests/test_storage_rules.py | 82 ++++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 11 deletions(-) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 113d4e69ee..932353015a 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -11,7 +11,7 @@ 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 @@ -19,6 +19,7 @@ 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) diff --git a/tests/test_storage_rules.py b/tests/test_storage_rules.py index 5c783019bd..38b3e59b4e 100644 --- a/tests/test_storage_rules.py +++ b/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