diff --git a/tests/test_cleanup_security.py b/tests/test_cleanup_security.py index c756413..36e6ee7 100644 --- a/tests/test_cleanup_security.py +++ b/tests/test_cleanup_security.py @@ -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): @@ -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) diff --git a/tests/test_run_bundle_retention.py b/tests/test_run_bundle_retention.py index ddf8169..eeaee47 100644 --- a/tests/test_run_bundle_retention.py +++ b/tests/test_run_bundle_retention.py @@ -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 @@ -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"