from __future__ import annotations import hashlib import os import sys import threading import time from pathlib import Path import pytest import subprocess ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import validate_p3_wp14_local_configs as validator # noqa: E402 def _snapshot() -> dict: return { "schema_version": "1.0", "environment": {"label": "LOCAL_ISOLATED_A", "is_enterprise": False}, "runtime": {"DATAOPS_LOCAL_ENV_LABEL": "LOCAL_ISOLATED_A", "BACKEND_PORT": "19014"}, "compose": {"file": "deploy/docker/docker-compose.yml", "render_command": validator.RENDER_COMMAND}, } @pytest.mark.parametrize("mutate", [ lambda x: x["runtime"].update({"BACKEND_PORT": "0"}), lambda x: x["runtime"].update({"DATAOPS_LOCAL_ENV_LABEL": "Ａ"}), lambda x: x["runtime"].update({"TOKEN": "secret"}), lambda x: x["environment"].update({"label": 1}), ]) def test_closed_snapshot_rejects_malicious_scalars(mutate) -> None: snapshot = _snapshot() mutate(snapshot) with pytest.raises(ValueError): validator._validate_snapshot(snapshot) def test_render_diff_rejects_extra_path() -> None: with pytest.raises(ValueError): validator._assert_render_diff({"/services/backend/labels/com.dataops.local_env"}, {"/extra"}) def test_secure_reader_rejects_symlink_and_hardlink(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) read_bytes_once = secure.read_bytes_once target = tmp_path / "target.json" target.write_text("{}") link = tmp_path / "link.json" link.symlink_to(target) hard = tmp_path / "hard.json" hard.hardlink_to(target) with pytest.raises(ValueError): read_bytes_once(link) with pytest.raises(ValueError): read_bytes_once(hard) def test_secure_reader_rejects_parent_symlink_and_lock_replacement(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) exclusive_lock, read_bytes_once = secure.exclusive_lock, secure.read_bytes_once real = tmp_path / "real"; real.mkdir() (real / "input.json").write_text("{}") parent_link = tmp_path / "parent-link"; parent_link.symlink_to(real, target_is_directory=True) with pytest.raises(OSError): read_bytes_once(parent_link / "input.json") marker = tmp_path / "no-marker.lock" with exclusive_lock(marker): pass assert not marker.exists() def test_config_timeout_marks_prior_render_invalid(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) monkeypatch.setattr(validator, "RENDER_RECORD", tmp_path / "render.json") monkeypatch.setattr(validator, "LOCK", tmp_path / "config-lock/lock") (tmp_path / "config-lock").mkdir() monkeypatch.setattr(validator, "_main_locked", lambda: (_ for _ in ()).throw(subprocess.TimeoutExpired(["docker"], 30))) assert validator.main() == 1 import json result = json.loads((tmp_path / "render.json").read_text()) assert result["status"] == "FAILED" and result["invalidates_prior_pass"] is True def test_ledger_timeout_marks_prior_ledger_invalid(monkeypatch, tmp_path: Path) -> None: import generate_p3_wp14_verification_ledger as ledger import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) monkeypatch.setattr(ledger, "STATE", tmp_path / "state.json") monkeypatch.setattr(ledger, "LOCK", tmp_path / "state.lock") monkeypatch.setattr(ledger, "_main_locked", lambda *_: (_ for _ in ()).throw(subprocess.TimeoutExpired(["pytest"], 300))) assert ledger.main() == 1 import json state = json.loads((tmp_path / "state.json").read_text()) assert state["status"] == "FAILED" and state["invalidates_prior_ledger_and_trace"] is True def test_expired_ledger_is_not_fresh() -> None: import generate_p3_wp14_verification_ledger as ledger from datetime import datetime, timezone assert not ledger.ledger_is_fresh({"expires_at": "2000-01-01T00:00:00Z"}, datetime.now(timezone.utc)) def test_dependency_drift_for_nonfirst_target_and_secure_io_is_detectable() -> None: import hashlib, json ledger = json.loads((ROOT / "docs/validation/P3_WP14_LOCAL_VERIFICATION_LEDGER.json").read_text()) run = next(item for item in ledger["runs"] if item["command_id"] == "p3_targeted_regression") for path in ("tests/test_wp13_plugin_platform_api.py", "scripts/p3_wp14_secure_io.py"): actual = hashlib.sha256((ROOT / path).read_bytes()).hexdigest() assert run["dependency_sha256"][path] != "0" * 64 assert ("0" * 64) != actual def test_manifest_minimal_env_rejects_git_injection(monkeypatch) -> None: import generate_p3_wp14_release_manifest as manifest monkeypatch.setenv("GIT_DIR", "/attacker") monkeypatch.setenv("GIT_WORK_TREE", "/attacker") monkeypatch.setenv("COMPOSE_FILE", "/attacker/compose.yml") monkeypatch.setenv("HOME", "/attacker/home-with-gitconfig") seen = {} def fake_run(command, **kwargs): seen.update(command=command, **kwargs) return subprocess.CompletedProcess(command, 0, stdout="ok", stderr="") monkeypatch.setattr(manifest.subprocess, "run", fake_run) command_env = manifest._minimal_command_env() manifest._run_command(["git", "rev-parse", "HEAD"], command_env, text=True) assert set(command_env) == {"PATH", "HOME", "DOCKER_CONFIG", "GIT_CONFIG_NOSYSTEM", "GIT_CONFIG_GLOBAL"} assert "GIT_DIR" not in seen["env"] and "GIT_WORK_TREE" not in seen["env"] assert "COMPOSE_FILE" not in seen["env"] assert seen["env"]["HOME"] == str(manifest.COMMAND_HOME) assert seen["env"]["GIT_CONFIG_NOSYSTEM"] == "1" assert seen["env"]["GIT_CONFIG_GLOBAL"] == "/dev/null" assert seen["cwd"] == manifest.ROOT and seen["timeout"] == manifest.COMMAND_TIMEOUT_SECONDS @pytest.mark.parametrize("operation", ["git", "compose"]) def test_manifest_git_and_compose_timeout_fail_closed(monkeypatch, operation: str) -> None: import generate_p3_wp14_release_manifest as manifest def timeout(*_args, **_kwargs): raise subprocess.TimeoutExpired([operation], manifest.COMMAND_TIMEOUT_SECONDS) monkeypatch.setattr(manifest, "_run_command", timeout) command_env = manifest._minimal_command_env() with pytest.raises(subprocess.TimeoutExpired): if operation == "git": manifest._collect_git_status(command_env, str(manifest.ROOT)) else: manifest._compose_services(command_env) def test_manifest_dirty_status_counts_and_digest_are_recomputed(monkeypatch) -> None: import generate_p3_wp14_release_manifest as manifest porcelain = b" M tracked.py\0?? untracked.txt\0A staged.py\0" def status_only(command, _env, *, text): assert command == ["git", "status", "--porcelain=v1", "-z"] and text is False return subprocess.CompletedProcess(command, 0, stdout=porcelain, stderr=b"") monkeypatch.setattr(manifest, "_run_command", status_only) status = manifest._collect_git_status(manifest._minimal_command_env(), str(manifest.ROOT)) assert status["entry_count"] == 3 assert status["tracked"] == 2 and status["untracked"] == 1 and status["staged"] == 1 assert status["sha256"] == hashlib.sha256(porcelain).hexdigest() def test_manifest_porcelain_rename_copy_paths_count_once(monkeypatch) -> None: import generate_p3_wp14_release_manifest as manifest porcelain = b"R renamed.py\0old.py\0C copied.py\0source.py\0?? new.py\0 M changed.py\0" def status_only(command, _env, *, text): assert command == ["git", "status", "--porcelain=v1", "-z"] and text is False return subprocess.CompletedProcess(command, 0, stdout=porcelain, stderr=b"") monkeypatch.setattr(manifest, "_run_command", status_only) status = manifest._collect_git_status(manifest._minimal_command_env(), str(manifest.ROOT)) assert status["entry_count"] == 4 assert status["tracked"] == 3 and status["untracked"] == 1 and status["staged"] == 2 def test_manifest_root_context_obeys_dockerignore_gz_rules(monkeypatch, tmp_path: Path) -> None: import generate_p3_wp14_release_manifest as manifest import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) (tmp_path / ".dockerignore").write_text("*.tar.gz\n") assets = tmp_path / "assets" assets.mkdir() model = assets / "model.gz" archive = assets / "model.tar.gz" model.write_bytes(b"model-v1") archive.write_bytes(b"archive-v1") initial = manifest._tree_digest(".", root=tmp_path) model.write_bytes(b"model-v2") model_changed = manifest._tree_digest(".", root=tmp_path) archive.write_bytes(b"archive-v2") archive_changed = manifest._tree_digest(".", root=tmp_path) assert model_changed != initial assert archive_changed == model_changed def test_directory_lock_stays_exclusive_when_marker_is_renamed(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) lock = tmp_path / "run.lock" lock.write_text("marker") entered = threading.Event() def second_holder() -> None: with secure.exclusive_lock(lock): entered.set() with secure.exclusive_lock(lock): lock.rename(tmp_path / "run.lock.old") lock.write_text("attacker replacement") thread = threading.Thread(target=second_holder) thread.start() time.sleep(0.1) assert not entered.is_set() thread.join(timeout=1) assert entered.is_set() def test_atomic_write_rejects_replaced_temp_inode(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) target = tmp_path / "record.json" parent, _ = secure._parent(target) temporary = ".record.json.attacker.tmp" fd = os.open(temporary, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=parent) try: os.write(fd, b"original") os.fsync(fd) replacement = tmp_path / "replacement" replacement.write_bytes(b"attacker") os.replace(replacement.name, temporary, src_dir_fd=parent, dst_dir_fd=parent) with pytest.raises(ValueError, match="temp swapped|unsafe path"): secure._verify_temp_matches_open_fd(parent, temporary, fd) finally: os.close(fd) try: os.unlink(temporary, dir_fd=parent) except FileNotFoundError: pass os.close(parent) def test_atomic_write_failure_removes_temp_and_syncs_parent(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) original_write = secure.os.write calls = {"count": 0} def failing_write(fd, data): calls["count"] += 1 if calls["count"] == 1: raise OSError("injected write failure") return original_write(fd, data) monkeypatch.setattr(secure.os, "write", failing_write) with pytest.raises(OSError, match="injected"): secure.atomic_write_bytes(tmp_path / "record.json", b"payload") assert not list(tmp_path.glob(".record.json.*.tmp")) def test_atomic_write_cooperative_reader_observes_only_old_or_new(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) target = tmp_path / "record.json" target.write_bytes(b"old") prefix_written = threading.Event() release_writer = threading.Event() reader_done = threading.Event() observed: list[bytes] = [] original_write = secure.os.write first_write = True def staged_write(fd, data): nonlocal first_write if first_write: first_write = False written = original_write(fd, data[:1]) prefix_written.set() assert release_writer.wait(timeout=2) return written return original_write(fd, data) monkeypatch.setattr(secure.os, "write", staged_write) writer = threading.Thread(target=lambda: secure.atomic_write_bytes(target, b"new-value")) writer.start() assert prefix_written.wait(timeout=1) def reader() -> None: observed.append(secure.read_bytes_once(target)) reader_done.set() reader_thread = threading.Thread(target=reader) reader_thread.start() time.sleep(0.1) assert not reader_done.is_set(), "cooperative reader saw an in-progress leaf" release_writer.set() writer.join(timeout=2) reader_thread.join(timeout=2) assert observed == [b"new-value"] @pytest.mark.parametrize("failure", ["write", "fsync", "rename"]) def test_atomic_write_failure_preserves_prior_verified_record(monkeypatch, tmp_path: Path, failure: str) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) target = tmp_path / "record.json" target.write_bytes(b"old-verified") if failure == "write": monkeypatch.setattr(secure.os, "write", lambda *_args: (_ for _ in ()).throw(OSError("write failure"))) elif failure == "fsync": monkeypatch.setattr(secure.os, "fsync", lambda *_args: (_ for _ in ()).throw(OSError("fsync failure"))) else: monkeypatch.setattr(secure.os, "replace", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("rename failure"))) with pytest.raises(OSError): secure.atomic_write_bytes(target, b"new") assert secure.read_bytes_once(target) == b"old-verified" def test_atomic_write_verify_then_replace_attack_restores_old_before_reader(monkeypatch, tmp_path: Path) -> None: import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) target = tmp_path / "record.json" target.write_bytes(b"old-verified") original = secure._verify_temp_matches_open_fd reader_done = threading.Event() observed: list[bytes] = [] def swap_after_verify(parent, temporary, fd): original(parent, temporary, fd) attacker = tmp_path / "attacker.tmp" attacker.write_bytes(b"attacker-bytes") os.replace(attacker.name, temporary, src_dir_fd=parent, dst_dir_fd=parent) reader = threading.Thread(target=lambda: (observed.append(secure.read_bytes_once(target)), reader_done.set())) reader.start() time.sleep(0.05) assert not reader_done.is_set(), "reader escaped the cooperative publish lock" reader.join(timeout=0.01) monkeypatch.setattr(secure, "_verify_temp_matches_open_fd", swap_after_verify) with pytest.raises(ValueError, match="published leaf"): secure.atomic_write_bytes(target, b"expected") assert secure.read_bytes_once(target) == b"old-verified" assert observed == [b"old-verified"] def test_manifest_input_census_rejects_root_context_change_during_generation(monkeypatch, tmp_path: Path) -> None: import generate_p3_wp14_release_manifest as manifest import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) monkeypatch.setattr(manifest, "ROOT", tmp_path) monkeypatch.setattr(manifest, "OUTPUT", tmp_path / "manifest.json") monkeypatch.setattr(manifest, "TREE_ROOTS", ()) monkeypatch.setattr(manifest, "FILE_PATHS", ()) monkeypatch.setattr(manifest, "COMPOSE", "compose.yml") monkeypatch.setattr(manifest, "COMPOSE_INPUTS", (("build_context", "."),)) (tmp_path / ".dockerignore").write_text("") (tmp_path / "compose.yml").write_text("services: {}\n") root_only = tmp_path / "root-only.py" root_only.write_text("before\n") monkeypatch.setattr(manifest, "_git_top_level", lambda _env: str(tmp_path)) monkeypatch.setattr(manifest, "_collect_git_status", lambda *_args: {"entry_count": 1, "sha256": "status"}) monkeypatch.setattr(manifest, "_compose_services", lambda _env: []) monkeypatch.setattr(manifest, "_image_reference_closure", lambda: []) monkeypatch.setattr( manifest, "_run_command", lambda command, _env, *, text: subprocess.CompletedProcess(command, 0, stdout="head\n", stderr=""), ) original_census = manifest._input_census calls = 0 def census_then_mutate(command_env): nonlocal calls value = original_census(command_env) calls += 1 if calls == 1: root_only.write_text("after\n") return value monkeypatch.setattr(manifest, "_input_census", census_then_mutate) with pytest.raises(RuntimeError, match="input closure changed"): manifest._main_locked() def test_compose_commands_force_empty_env_file_and_ignore_parent_dotenv(monkeypatch) -> None: import generate_p3_wp14_release_manifest as manifest monkeypatch.setenv("BACKEND_PORT", "attacker-port") assert validator.RENDER_COMMAND[1:3] == ["--env-file", "/dev/null"] captured = {} def fake_run(command, command_env, *, text): captured.update(command=command, env=command_env, text=text) return subprocess.CompletedProcess(command, 0, stdout='{"services": {}}', stderr="") monkeypatch.setattr(manifest, "_run_command", fake_run) manifest._compose_services(manifest._minimal_command_env()) assert captured["command"][1:3] == ["--env-file", "/dev/null"] assert "BACKEND_PORT" not in captured["env"] assert captured["env"]["HOME"] == str(manifest.COMMAND_HOME) def test_dockerignore_matcher_rejects_linked_source(monkeypatch, tmp_path: Path) -> None: import generate_p3_wp14_release_manifest as manifest import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) target = tmp_path / "rules" target.write_text("*.tar.gz\n") linked = tmp_path / ".dockerignore" linked.symlink_to(target) with pytest.raises(ValueError): manifest.DockerIgnoreMatcher.from_root(tmp_path) def test_dockerignore_matcher_refuses_unsupported_glob(monkeypatch, tmp_path: Path) -> None: import generate_p3_wp14_release_manifest as manifest import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) (tmp_path / ".dockerignore").write_text("[ab].txt\n") with pytest.raises(ValueError, match="refusing approximation"): manifest.DockerIgnoreMatcher.from_root(tmp_path) def test_image_closure_marks_every_mutable_tag_blocked() -> None: import generate_p3_wp14_release_manifest as manifest records = manifest._image_reference_closure( "services:\n api:\n image: registry.example/model:latest\n pinned:\n image: registry.example/pinned@sha256:abc\n", {"Dockerfile": "FROM registry.example/base:stable AS build\nFROM build\n"}, ) by_reference = {item["reference"]: item for item in records} assert by_reference["registry.example/model:latest"]["status"] == "UNRESOLVED_MUTABLE_TAG" assert by_reference["registry.example/base:stable"]["gate"] == "BLOCKED_EXTERNAL" assert by_reference["registry.example/pinned@sha256:abc"]["status"] == "IMMUTABLE_DIGEST" assert by_reference["build"]["status"] == "INTERNAL_BUILD_STAGE" def test_tree_digest_fails_closed_on_resource_boundaries(monkeypatch, tmp_path: Path) -> None: import generate_p3_wp14_release_manifest as manifest import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) (tmp_path / ".dockerignore").write_text("") (tmp_path / "one").write_text("1") (tmp_path / "two").write_text("2") monkeypatch.setattr(manifest, "MAX_TREE_FILES", 1) with pytest.raises(ValueError, match="resource boundary"): manifest._tree_digest(".", root=tmp_path) @pytest.mark.parametrize("shape", ["deep", "many"]) def test_tree_digest_rejects_ignored_directory_resource_exhaustion(monkeypatch, tmp_path: Path, shape: str) -> None: import generate_p3_wp14_release_manifest as manifest import p3_wp14_secure_io as secure monkeypatch.setattr(secure, "TRUSTED_ROOT", tmp_path) # The negation means a matcher cannot safely prune the ignored directory. (tmp_path / ".dockerignore").write_text("ignored/\n!ignored/keep.txt\n") ignored = tmp_path / "ignored" ignored.mkdir() if shape == "deep": current = ignored for number in range(manifest.MAX_TREE_DEPTH + 8): current = current / f"level{number}" current.mkdir() else: monkeypatch.setattr(manifest, "MAX_TREE_FILES", 4) for number in range(5): (ignored / f"branch{number}").mkdir() with pytest.raises(ValueError, match="resource boundary|path boundary"): manifest._tree_digest(".", root=tmp_path)