DataOps-platform/tests/test_p3_wp14_local_config_validator.py

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)