Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 32 additions & 1 deletion tests/test_cleanup_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,12 @@

import base_cli
import base_cli._cleanup as cleanup_module
from base_cli._cleanup import UnsafeCleanupPathError, _is_root_like, remove_owned_temp_directory
from base_cli._cleanup import (
UnsafeCleanupPathError,
_is_root_like,
_validated_cleanup_paths,
remove_owned_temp_directory,
)


class CleanupSecurityTests(unittest.TestCase):
Expand Down Expand Up @@ -463,6 +468,32 @@ def is_mount(path: Path) -> bool:
self.assertIn("mounted temp directories", stream.getvalue())
self.assertEqual(context.log.handlers, [])

def test_validation_rejects_traversal_outside_root_and_invalid_marker(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "run"
cases = (
(root / "tmp" / ".." / "run-123", root, "run-123", "path traversal"),
(Path(tmpdir) / "other" / "run-123", root, "run-123", "outside the run root"),
(root / "tmp" / "run-123", root, "../run-123", "ownership marker"),
)
for target, run_root, run_id, message in cases:
with self.subTest(target=target, run_id=run_id):
with self.assertRaisesRegex(UnsafeCleanupPathError, message):
_validated_cleanup_paths(target, run_root, run_id, (1, 2))

def test_validation_rejects_resolved_path_mismatch(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "run"
target = root / "tmp" / "cleanup-security" / "run-123"
target.mkdir(parents=True)
with mock.patch.object(
cleanup_module,
"_strict_relative_path",
side_effect=(Path("tmp/cleanup-security/run-123"), Path("elsewhere/run-123")),
):
with self.assertRaisesRegex(UnsafeCleanupPathError, "does not match"):
_validated_cleanup_paths(target, root, "run-123", (target.stat().st_dev, target.stat().st_ino))

def test_platform_without_safe_directory_handles_fails_closed(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
Expand Down
13 changes: 13 additions & 0 deletions tests/test_run_bundle_retention.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from pathlib import Path
from unittest import mock

import base_cli._private_files as private_files
from base_cli import RetentionPolicy
from base_cli._private_files import write_private_json
from base_cli._runtime import prune_run_bundles
Expand Down Expand Up @@ -149,6 +150,18 @@ def test_atomic_json_write_refuses_symlink_destination(self) -> None:
write_private_json(destination, {"status": "error"})
self.assertEqual(victim.read_text(encoding="utf-8"), "unchanged")

def test_atomic_json_write_refuses_symlink_destination_without_directory_handles(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
victim = root / "victim.json"
victim.write_text("unchanged", encoding="utf-8")
destination = root / "run.json"
destination.symlink_to(victim)
with mock.patch.object(private_files, "_open_parent_directory", return_value=None):
with self.assertRaisesRegex(OSError, "refusing to replace symlink"):
write_private_json(destination, {"status": "error"})
self.assertEqual(victim.read_text(encoding="utf-8"), "unchanged")

def test_concurrent_metadata_writers_leave_one_valid_snapshot(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "run.json"
Expand Down
Loading