DataOps-platform/tests/test_phase3_wp04_delivery_contract.py

"""Delivery contracts for the P3-WP04 enterprise edge engineering baseline."""

from __future__ import annotations

import fcntl
import hashlib
import importlib.util
import json
import os
import shutil
import stat
import subprocess
import sys
import tarfile
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace

import jsonschema
import pytest
import yaml
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ed25519, rsa
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID

from app.core.edge_gateway.contracts import canonical_json_bytes, canonical_sha256

ROOT = Path(__file__).resolve().parents[1]
CLI = ROOT / "scripts" / "edge_gateway_admin.py"
COMPOSE = ROOT / "deploy" / "edge" / "docker-compose.edge.yml"
ENV_EXAMPLE = ROOT / "deploy" / "edge" / "edge.env.example"
OPENAPI = ROOT / "docs" / "architecture" / "OPENAPI.yaml"
FAILURES = ROOT / "docs" / "validation" / "P3_WP04_FAILURE_INJECTION.json"
EDGE_CONFIG_EXAMPLE = ROOT / "deploy" / "edge" / "edge.config.example.json"
EDGE_CONFIG_SCHEMA = ROOT / "deploy" / "edge" / "edge.config.schema.json"
RELEASE_SECRET_SCANNER = ROOT / "deployment" / "scripts" / "scan_release_secrets.py"


def _load_cli_module():
    spec = importlib.util.spec_from_file_location(
        f"edge_gateway_admin_test_{os.getpid()}_{id(object())}", CLI
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _run(*args: str, expect: int = 0) -> tuple[dict[str, object], str]:
    completed = subprocess.run(
        [sys.executable, str(CLI), *args],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=False,
        env={**os.environ, "PYTHONPATH": str(ROOT)},
    )
    assert completed.returncode == expect, completed.stderr
    payload = json.loads(completed.stdout)
    return payload, completed.stderr


def _init(root: Path) -> dict[str, object]:
    return _run(
        "init", "--root", str(root), "--runtime-uid", str(os.getuid()),
        "--runtime-gid", str(os.getgid()),
    )[0]


def _certificate_for(
    key_path: Path, certificate_path: Path, ca_path: Path, crl_path: Path
) -> str:
    key = serialization.load_pem_private_key(key_path.read_bytes(), password=None)
    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "edge-test-ca")])
    now = datetime.now(UTC)
    ca = (
        x509.CertificateBuilder()
        .subject_name(ca_name)
        .issuer_name(ca_name)
        .public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(minutes=1))
        .not_valid_after(now + timedelta(days=365))
        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
        .sign(ca_key, hashes.SHA256())
    )
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "edge-test")])
    certificate = (
        x509.CertificateBuilder()
        .subject_name(name)
        .issuer_name(ca.subject)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(minutes=1))
        .not_valid_after(now + timedelta(days=30))
        .add_extension(
            x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]), critical=True
        )
        .sign(ca_key, hashes.SHA256())
    )
    certificate_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM))
    ca_path.write_bytes(ca.public_bytes(serialization.Encoding.PEM))
    crl = (
        x509.CertificateRevocationListBuilder()
        .issuer_name(ca.subject)
        .last_update(now - timedelta(minutes=1))
        .next_update(now + timedelta(days=7))
        .sign(ca_key, hashes.SHA256())
    )
    crl_path.write_bytes(crl.public_bytes(serialization.Encoding.PEM))
    return certificate.fingerprint(hashes.SHA256()).hex()


def _release_manifest(
    private: ed25519.Ed25519PrivateKey,
    *,
    release_id: str,
    version: str,
    rollback_version: str,
    artifact: Path,
    status: str = "installed",
) -> dict[str, object]:
    unsigned = {
        "artifact_digest": hashlib.sha256(artifact.read_bytes()).hexdigest(),
        "artifact_name": artifact.name,
        "deadline_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat().replace("+00:00", "Z"),
        "key_id": "release-key-1",
        "release_id": release_id,
        "rollback_version": rollback_version,
        "signature_algorithm": "Ed25519",
        "status": status,
        "version": version,
    }
    return {
        **unsigned,
        "manifest_digest": canonical_sha256(unsigned),
        "signature": private.sign(canonical_json_bytes(unsigned)).hex(),
    }


def _verify_release(
    root: Path, manifest: dict[str, object], artifact: Path, public_hex: str
) -> dict[str, object]:
    source = root / "releases" / f"source-{manifest['release_id']}.json"
    source.write_text(json.dumps(manifest))
    return _run(
        "release-check", "--root", str(root), "--manifest", str(source),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", str(manifest["rollback_version"]),
    )[0]


def _signed_status(
    private: ed25519.Ed25519PrivateKey,
    *,
    gateway_id: str = "gw-test",
    generation: int = 1,
    certificate_sha256: str = "b" * 64,
    status: str = "revoked",
    expires_delta: timedelta = timedelta(hours=1),
) -> dict[str, object]:
    unsigned = {
        "certificate_sha256": certificate_sha256,
        "expires_at": (datetime.now(UTC) + expires_delta).isoformat().replace("+00:00", "Z"),
        "gateway_id": gateway_id,
        "generation": generation,
        "issued_at": (datetime.now(UTC) - timedelta(seconds=1)).isoformat().replace("+00:00", "Z"),
        "key_id": "status-key-1",
        "signature_algorithm": "Ed25519",
        "status": status,
    }
    return {
        **unsigned,
        "envelope_digest": canonical_sha256(unsigned),
        "signature": private.sign(canonical_json_bytes(unsigned)).hex(),
    }


def test_admin_init_csr_and_registration_material_never_expose_private_key(tmp_path):
    root = tmp_path / "edge-root"
    initialized = _init(root)
    init_stderr = ""
    assert initialized["status"] == "initialized"
    assert stat.S_IMODE(root.stat().st_mode) == 0o700
    assert {"queue", "artifacts", "releases", "secrets", "certificates"} <= {
        path.name for path in root.iterdir()
    }

    created, csr_stderr = _run(
        "csr", "--root", str(root), "--common-name", "edge-test"
    )
    key_path = Path(str(created["private_key_path"]))
    csr_path = Path(str(created["csr_path"]))
    assert key_path.is_file() and csr_path.is_file()
    assert stat.S_IMODE(key_path.stat().st_mode) == 0o600
    assert x509.load_pem_x509_csr(csr_path.read_bytes()).is_signature_valid
    certificate_path = root / "certificates" / "client.pem"
    ca_path = root / "certificates" / "client-ca.pem"
    crl_path = root / "certificates" / "client-ca.crl.pem"
    digest = _certificate_for(key_path, certificate_path, ca_path, crl_path)

    registered, register_stderr = _run(
        "register-manifest",
        "--root", str(root),
        "--certificate", str(certificate_path),
        "--ca-bundle", str(ca_path),
        "--crl", str(crl_path),
        "--gateway-id", "gw-test",
        "--environment", "production",
        "--network-zone", "zone-a",
        "--policy-digest", "a" * 64,
        "--control-origin", "https://control.example.test",
    )
    manifest_path = root / "registration" / "registration.json"
    assert registered["certificate_sha256"] == digest
    registration = json.loads(manifest_path.read_text())
    assert registration["certificate_sha256"] == digest
    combined = "\n".join(
        [json.dumps(initialized), json.dumps(created), json.dumps(registered),
         init_stderr, csr_stderr, register_stderr, manifest_path.read_text()]
    )
    assert "BEGIN PRIVATE KEY" not in combined
    assert key_path.read_text() not in combined
    assert "credential" not in registration
    assert not any("path" in key for key in registration)
    assert registration["ca_chain_verified"] is True
    assert registration["crl_verified"] is True
    assert registered["replayed"] is False
    replayed, _ = _run(
        "register-manifest", "--root", str(root),
        "--certificate", str(certificate_path), "--ca-bundle", str(ca_path),
        "--crl", str(crl_path), "--gateway-id", "gw-test",
        "--environment", "production", "--network-zone", "zone-a",
        "--policy-digest", "a" * 64,
        "--control-origin", "https://control.example.test",
    )
    assert replayed["replayed"] is True
    conflict, _ = _run(
        "register-manifest", "--root", str(root),
        "--certificate", str(certificate_path), "--ca-bundle", str(ca_path),
        "--crl", str(crl_path), "--gateway-id", "gw-other",
        "--environment", "production", "--network-zone", "zone-a",
        "--policy-digest", "a" * 64,
        "--control-origin", "https://control.example.test", expect=3,
    )
    assert conflict["error"]["code"] == "IDEMPOTENCY_CONFLICT"


def test_init_rejects_broad_or_nonempty_roots_without_mutation(tmp_path):
    spec = importlib.util.spec_from_file_location("edge_gateway_admin_root_test", CLI)
    assert spec is not None and spec.loader is not None
    admin = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(admin)
    protected = [Path("/"), Path.home(), Path("/opt"), Path("/var"), Path("/usr"),
                 Path("/etc"), Path("/tmp")]
    snapshots = {
        path: (path.stat().st_mode, path.stat().st_uid, path.stat().st_gid)
        for path in protected if path.exists()
    }
    for path in snapshots:
        with pytest.raises(admin.AdminError, match="PATH_POLICY"):
            admin._root(str(path), create=False)
    assert snapshots == {
        path: (path.stat().st_mode, path.stat().st_uid, path.stat().st_gid)
        for path in snapshots
    }

    nonempty = tmp_path / "existing-nonempty"
    nonempty.mkdir()
    sentinel = nonempty / "do-not-touch.txt"
    sentinel.write_text("preserve")
    rejected, _ = _run(
        "init", "--root", str(nonempty), "--runtime-uid", str(os.getuid()),
        "--runtime-gid", str(os.getgid()), expect=2,
    )
    assert rejected["error"]["code"] == "PATH_POLICY"
    assert sentinel.read_text() == "preserve"
    assert set(nonempty.iterdir()) == {sentinel}


def test_mutating_commands_share_fail_closed_process_lock(tmp_path):
    root = tmp_path / "edge-root"
    _init(root)
    lock_path = root / "state" / "admin.lock"
    descriptor = os.open(
        lock_path,
        os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0),
        0o600,
    )
    try:
        fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        blocked, _ = _run(
            "csr", "--root", str(root), "--common-name", "edge-test", expect=3,
        )
        assert blocked["error"]["code"] == "CONCURRENT_OPERATION"
        assert not (root / "secrets" / "client.key").exists()
    finally:
        os.close(descriptor)
    status_payload, _ = _run("status", "--root", str(root))
    assert status_payload["status"] == "read_only"
    assert status_payload["active_leases"] == 0
    rendered = json.dumps(status_payload)
    assert str(root) not in rendered
    assert "credential" not in rendered.casefold()


@pytest.mark.parametrize(
    "version",
    ["1.0", "1.0.0-alpha", "1.0.0+build", "01.0.0", "2147483648.0.0"],
)
def test_release_rejects_noncanonical_or_unbounded_versions(tmp_path, version):
    root = tmp_path / "edge-root"
    _init(root)
    artifact = root / "releases" / "edge.bin"
    artifact.write_bytes(b"version policy")
    key = ed25519.Ed25519PrivateKey.generate()
    public_hex = key.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    manifest = _release_manifest(
        key, release_id="version-policy", version=version,
        rollback_version="0.9.0", artifact=artifact, status="offered",
    )
    source = root / "releases" / "version-policy.json"
    source.write_text(json.dumps(manifest))
    rejected, _ = _run(
        "release-check", "--root", str(root), "--manifest", str(source),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", "0.9.0", expect=3,
    )
    assert rejected["error"]["code"] == "RELEASE_REJECTED"
    assert not list((root / "releases" / "verified").glob(".bundle-*"))


@pytest.mark.parametrize(
    "version", ["1.0", "1.0.0-alpha", "1.0.0+build", "1.2.3.4", "01.0.0", "2147483648.0.0"],
)
def test_control_plane_release_version_matches_edge_cli_policy(version):
    from app.core.edge_gateway.service import (
        EdgeGatewayValidationError,
        _release_version,
    )

    with pytest.raises(EdgeGatewayValidationError):
        _release_version(version, "version")


def test_admin_atomic_outputs_inherit_runtime_parent_owner(monkeypatch, tmp_path):
    module = _load_cli_module()
    parent = tmp_path / "runtime-owned"
    parent.mkdir(mode=0o700)
    directory_chowns: list[tuple[Path, int, int]] = []
    descriptor_chowns: list[tuple[int, int, int]] = []
    monkeypatch.setattr(module.os, "geteuid", lambda: 0)
    monkeypatch.setattr(
        module.os, "chown",
        lambda path, uid, gid, **_kwargs: directory_chowns.append((Path(path), uid, gid)),
    )
    monkeypatch.setattr(
        module.os, "fchown",
        lambda fd, uid, gid: descriptor_chowns.append((fd, uid, gid)),
    )
    child = module._mkdir(parent, "state")
    expected = (parent.stat().st_uid, parent.stat().st_gid)
    assert directory_chowns == [(child, *expected)]
    monkeypatch.setattr(
        module.os, "fstat", lambda _fd: SimpleNamespace(st_uid=-1, st_gid=-1)
    )
    module._inherit_parent_owner(99, parent)
    assert any((uid, gid) == expected for _fd, uid, gid in descriptor_chowns)


def test_release_check_rejects_hardlinked_artifact(tmp_path):
    root = tmp_path / "edge-root"
    _init(root)
    original = root / "releases" / "original.bin"
    original.write_bytes(b"hardlink must fail")
    artifact = root / "releases" / "edge.bin"
    os.link(original, artifact)
    key = ed25519.Ed25519PrivateKey.generate()
    public_hex = key.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    manifest = _release_manifest(
        key, release_id="hardlink-release", version="1.0.0",
        rollback_version="0.9.0", artifact=artifact, status="offered",
    )
    source = root / "releases" / "hardlink.json"
    source.write_text(json.dumps(manifest))
    rejected, _ = _run(
        "release-check", "--root", str(root), "--manifest", str(source),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", "0.9.0", expect=2,
    )
    assert rejected["error"]["code"] == "PATH_POLICY"


def test_admin_requires_absolute_root_and_rejects_symlink_escape(tmp_path):
    payload, _ = _run(
        "init", "--root", "relative-edge-root", "--runtime-uid", str(os.getuid()),
        "--runtime-gid", str(os.getgid()), expect=2,
    )
    assert payload == {"error": {"code": "INVALID_INPUT"}, "status": "error"}
    root = tmp_path / "root"
    outside = tmp_path / "outside"
    outside.mkdir()
    root.mkdir()
    (root / "secrets").symlink_to(outside, target_is_directory=True)
    payload, _ = _run(
        "init", "--root", str(root), "--runtime-uid", str(os.getuid()),
        "--runtime-gid", str(os.getgid()), expect=2,
    )
    assert payload["error"]["code"] == "PATH_POLICY"
    mismatched = tmp_path / "owner-mismatch"
    owner_error, _ = _run(
        "init", "--root", str(mismatched),
        "--runtime-uid", str(os.getuid() + 1),
        "--runtime-gid", str(os.getgid()), expect=2,
    )
    assert owner_error["error"]["code"] == "OWNER_POLICY"


def test_rotate_revoke_and_rollback_are_idempotent_and_fail_closed(
    tmp_path, monkeypatch
):
    root = tmp_path / "edge-root"
    _init(root)
    rotated, _ = _run(
        "rotate", "--root", str(root), "--common-name", "edge-test",
        "--request-id", "rotate-1",
        "--expected-current-certificate-digest", "b" * 64,
    )
    assert rotated["status"] == "pending_server_approval"
    assert Path(str(rotated["private_key_path"])).stat().st_mode & 0o077 == 0
    replayed, _ = _run(
        "rotate", "--root", str(root), "--common-name", "edge-test",
        "--request-id", "rotate-1",
        "--expected-current-certificate-digest", "b" * 64,
    )
    assert replayed == rotated
    conflict, _ = _run(
        "rotate", "--root", str(root), "--common-name", "edge-other",
        "--request-id", "rotate-1",
        "--expected-current-certificate-digest", "b" * 64, expect=3,
    )
    assert conflict["error"]["code"] == "IDEMPOTENCY_CONFLICT"

    status_key = ed25519.Ed25519PrivateKey.generate()
    status_public = status_key.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    signed_status = _signed_status(status_key)
    status_path = root / "state" / "server-revocation.json"
    status_path.write_text(json.dumps(signed_status))
    revoked, _ = _run(
        "revoke-status", "--root", str(root), "--server-status", str(status_path),
        "--trusted-key", f"status-key-1={status_public}",
        "--expected-gateway-id", "gw-test", "--expected-generation", "1",
        "--expected-certificate-digest", "b" * 64,
    )
    assert revoked["local_stop_required"] is True
    assert (root / "state" / "REVOKED_STOP").is_file()

    forged = _signed_status(status_key, gateway_id="gw-other")
    status_path.write_text(json.dumps(forged))
    rejected, _ = _run(
        "revoke-status", "--root", str(root), "--server-status", str(status_path),
        "--trusted-key", f"status-key-1={status_public}",
        "--expected-gateway-id", "gw-test", "--expected-generation", "1",
        "--expected-certificate-digest", "b" * 64, expect=3,
    )
    assert rejected["error"]["code"] == "STATUS_REJECTED"

    expired = _signed_status(status_key, expires_delta=timedelta(seconds=-1))
    status_path.write_text(json.dumps(expired))
    expired_result, _ = _run(
        "revoke-status", "--root", str(root), "--server-status", str(status_path),
        "--trusted-key", f"status-key-1={status_public}",
        "--expected-gateway-id", "gw-test", "--expected-generation", "1",
        "--expected-certificate-digest", "b" * 64, expect=3,
    )
    assert expired_result["error"]["code"] == "STATUS_REJECTED"

    bad_signature = _signed_status(status_key)
    bad_signature["signature"] = "0" * 128
    status_path.write_text(json.dumps(bad_signature))
    forged_result, _ = _run(
        "revoke-status", "--root", str(root), "--server-status", str(status_path),
        "--trusted-key", f"status-key-1={status_public}",
        "--expected-gateway-id", "gw-test", "--expected-generation", "1",
        "--expected-certificate-digest", "b" * 64, expect=3,
    )
    assert forged_result["error"]["code"] == "STATUS_REJECTED"

    active = _signed_status(status_key, status="active")
    status_path.write_text(json.dumps(active))
    terminal, _ = _run(
        "revoke-status", "--root", str(root), "--server-status", str(status_path),
        "--trusted-key", f"status-key-1={status_public}",
        "--expected-gateway-id", "gw-test", "--expected-generation", "1",
        "--expected-certificate-digest", "b" * 64,
    )
    assert terminal["status"] == "revoked"
    assert terminal["local_stop_required"] is True

    release_key = ed25519.Ed25519PrivateKey.generate()
    public_hex = release_key.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    target_artifact = root / "releases" / "edge-1.0.0.tar"
    target_artifact.write_bytes(b"prior verified release")
    target = _release_manifest(
        release_key, release_id="prior-release", version="1.0.0",
        rollback_version="0.9.0", artifact=target_artifact,
    )
    target_checked = _verify_release(root, target, target_artifact, public_hex)
    current_artifact = root / "releases" / "edge-2.0.0.tar"
    current_artifact.write_bytes(b"current verified release")
    current = _release_manifest(
        release_key, release_id="current-release", version="2.0.0",
        rollback_version="1.0.0", artifact=current_artifact,
    )
    current_checked = _verify_release(root, current, current_artifact, public_hex)
    current_bundle = json.loads(
        Path(str(current_checked["verified_bundle_path"]), "bundle.json").read_text()
    )
    pointer = {
        "artifact_name": current["artifact_name"],
        "bundle_digest": current_bundle["bundle_digest"],
        "manifest_digest": current["manifest_digest"],
        "schema_version": 1,
        "version": current["version"],
    }
    (root / "releases" / "current.json").write_text(json.dumps(pointer))
    spec = importlib.util.spec_from_file_location("edge_gateway_admin_fsync_test", CLI)
    assert spec is not None and spec.loader is not None
    admin = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(admin)
    real_fsync_directory = admin._fsync_directory
    failed = False

    def fail_after_pointer_replace(path):
        nonlocal failed
        if Path(path) == root / "releases" and not failed:
            failed = True
            raise OSError("injected directory fsync failure")
        return real_fsync_directory(path)

    monkeypatch.setattr(admin, "_fsync_directory", fail_after_pointer_replace)
    with pytest.raises(OSError, match="injected"):
        admin.command_rollback(SimpleNamespace(
            root=str(root), yes=True,
            expected_current_digest=str(current["manifest_digest"]),
            target_digest=str(target_checked["manifest_digest"]),
            trusted_key=[f"release-key-1={public_hex}"],
        ))
    assert (root / "state" / "rollback-intent.json").is_file()
    rolled_back, _ = _run(
        "rollback", "--root", str(root), "--yes",
        "--expected-current-digest", str(current["manifest_digest"]),
        "--target-digest", str(target_checked["manifest_digest"]),
        "--trusted-key", f"release-key-1={public_hex}",
    )
    assert rolled_back["status"] == "rolled_back"
    assert rolled_back["replayed"] is True
    assert rolled_back["version"] == "1.0.0"
    committed = json.loads((root / "releases" / "current.json").read_text())
    assert committed["manifest_digest"] == target["manifest_digest"]
    replayed_rollback, _ = _run(
        "rollback", "--root", str(root), "--yes",
        "--expected-current-digest", str(current["manifest_digest"]),
        "--target-digest", str(target_checked["manifest_digest"]),
        "--trusted-key", f"release-key-1={public_hex}",
    )
    assert replayed_rollback["replayed"] is True

    target_bundle = json.loads(
        Path(str(target_checked["verified_bundle_path"]), "bundle.json").read_text()
    )
    target_pointer = {
        "artifact_name": target["artifact_name"],
        "bundle_digest": target_bundle["bundle_digest"],
        "manifest_digest": target["manifest_digest"],
        "schema_version": 1,
        "version": target["version"],
    }
    rollback_request_digest = admin._protocol_digest({
        "expected_current_digest": str(current["manifest_digest"]),
        "target_digest": str(target["manifest_digest"]),
    })
    matching_intent = {
        "current_pointer": pointer,
        "request_digest": rollback_request_digest,
        "target_pointer": target_pointer,
    }
    intent_path = root / "state" / "rollback-intent.json"
    intent_path.write_text(json.dumps({**matching_intent, "request_digest": "f" * 64}))
    mismatched, _ = _run(
        "rollback", "--root", str(root), "--yes",
        "--expected-current-digest", str(current["manifest_digest"]),
        "--target-digest", str(target["manifest_digest"]),
        "--trusted-key", f"release-key-1={public_hex}", expect=3,
    )
    assert mismatched["error"]["code"] == "STATE_CONFLICT"
    assert intent_path.is_file()

    intent_path.write_text(json.dumps(matching_intent))
    real_unlink = Path.unlink
    unlink_failed = False

    def fail_intent_unlink(path, *args, **kwargs):
        nonlocal unlink_failed
        if path == intent_path and not unlink_failed:
            unlink_failed = True
            raise OSError("injected intent unlink failure")
        return real_unlink(path, *args, **kwargs)

    monkeypatch.setattr(Path, "unlink", fail_intent_unlink)
    with pytest.raises(OSError, match="unlink"):
        admin.command_rollback(SimpleNamespace(
            root=str(root), yes=True,
            expected_current_digest=str(current["manifest_digest"]),
            target_digest=str(target["manifest_digest"]),
            trusted_key=[f"release-key-1={public_hex}"],
        ))
    assert intent_path.is_file()
    monkeypatch.setattr(Path, "unlink", real_unlink)

    state_fsync_failed = False

    def fail_state_fsync_once(path):
        nonlocal state_fsync_failed
        if Path(path) == root / "state" and not state_fsync_failed:
            state_fsync_failed = True
            raise OSError("injected state fsync failure")
        return real_fsync_directory(path)

    monkeypatch.setattr(admin, "_fsync_directory", fail_state_fsync_once)
    with pytest.raises(OSError, match="state fsync"):
        admin.command_rollback(SimpleNamespace(
            root=str(root), yes=True,
            expected_current_digest=str(current["manifest_digest"]),
            target_digest=str(target["manifest_digest"]),
            trusted_key=[f"release-key-1={public_hex}"],
        ))
    assert not intent_path.exists()
    monkeypatch.setattr(admin, "_fsync_directory", real_fsync_directory)
    recovered_replay = admin.command_rollback(SimpleNamespace(
        root=str(root), yes=True,
        expected_current_digest=str(current["manifest_digest"]),
        target_digest=str(target["manifest_digest"]),
        trusted_key=[f"release-key-1={public_hex}"],
    ))
    assert recovered_replay["replayed"] is True

    older_artifact = root / "releases" / "edge-0.9.0.tar"
    older_artifact.write_bytes(b"older verified release")
    older = _release_manifest(
        release_key, release_id="older-release", version="0.9.0",
        rollback_version="0.8.0", artifact=older_artifact,
    )
    older_checked = _verify_release(root, older, older_artifact, public_hex)
    next_rollback, _ = _run(
        "rollback", "--root", str(root), "--yes",
        "--expected-current-digest", str(target["manifest_digest"]),
        "--target-digest", str(older_checked["manifest_digest"]),
        "--trusted-key", f"release-key-1={public_hex}",
    )
    assert next_rollback["status"] == "rolled_back"
    assert next_rollback["version"] == "0.9.0"


def test_release_check_verifies_ed25519_artifact_digest_and_replay(tmp_path):
    root = tmp_path / "edge-root"
    _init(root)
    artifact = root / "releases" / "edge-2.0.0.tar"
    artifact.write_bytes(b"verified edge release")
    private = ed25519.Ed25519PrivateKey.generate()
    public_hex = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    unsigned = {
        "artifact_digest": hashlib.sha256(artifact.read_bytes()).hexdigest(),
        "artifact_name": artifact.name,
        "deadline_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat().replace("+00:00", "Z"),
        "key_id": "release-key-1",
        "release_id": "rel-1",
        "rollback_version": "1.0.0",
        "signature_algorithm": "Ed25519",
        "status": "offered",
        "version": "2.0.0",
    }
    manifest = {
        **unsigned,
        "manifest_digest": canonical_sha256(unsigned),
        "signature": private.sign(canonical_json_bytes(unsigned)).hex(),
    }
    manifest_path = root / "releases" / "manifest.json"
    manifest_path.write_text(json.dumps(manifest))
    checked, _ = _run(
        "release-check", "--root", str(root), "--manifest", str(manifest_path),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", "1.0.0",
    )
    assert checked["status"] == "verified"
    assert checked["manifest_digest"] == manifest["manifest_digest"]
    bundle = Path(str(checked["verified_bundle_path"]))
    assert (bundle / "manifest.json").is_file()
    assert (bundle / artifact.name).read_bytes() == artifact.read_bytes()
    assert (bundle / "bundle.json").is_file()
    exact, _ = _run(
        "release-check", "--root", str(root), "--manifest", str(manifest_path),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", "1.0.0",
    )
    assert exact["replayed"] is True
    changed_unsigned = {**unsigned, "status": "accepted"}
    changed_replay = {
        **changed_unsigned,
        "manifest_digest": canonical_sha256(changed_unsigned),
        "signature": private.sign(canonical_json_bytes(changed_unsigned)).hex(),
    }
    manifest_path.write_text(json.dumps(changed_replay))
    replay_rejected, _ = _run(
        "release-check", "--root", str(root), "--manifest", str(manifest_path),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", "1.0.0", expect=3,
    )
    assert replay_rejected["error"]["code"] == "RELEASE_REJECTED"
    tampered = dict(manifest)
    tampered["version"] = "2.0.1"
    manifest_path.write_text(json.dumps(tampered))
    rejected, _ = _run(
        "release-check", "--root", str(root), "--manifest", str(manifest_path),
        "--artifact", str(artifact), "--trusted-key", f"release-key-1={public_hex}",
        "--current-version", "1.0.0", expect=3,
    )
    assert rejected["error"]["code"] == "RELEASE_REJECTED"


def test_rollback_rejects_forged_verified_manifest_and_tampered_artifact(tmp_path):
    root = tmp_path / "edge-root"
    _init(root)
    key = ed25519.Ed25519PrivateKey.generate()
    public_hex = key.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    target_artifact = root / "releases" / "target.bin"
    target_artifact.write_bytes(b"target")
    target = _release_manifest(
        key, release_id="target", version="1.0.0", rollback_version="0.9.0",
        artifact=target_artifact,
    )
    target_checked = _verify_release(root, target, target_artifact, public_hex)
    current_artifact = root / "releases" / "current.bin"
    current_artifact.write_bytes(b"current")
    current = _release_manifest(
        key, release_id="current", version="2.0.0", rollback_version="1.0.0",
        artifact=current_artifact,
    )
    current_checked = _verify_release(root, current, current_artifact, public_hex)
    current_bundle = Path(str(current_checked["verified_bundle_path"]))
    bundle_record = json.loads((current_bundle / "bundle.json").read_text())
    (root / "releases" / "current.json").write_text(json.dumps({
        "artifact_name": current["artifact_name"],
        "bundle_digest": bundle_record["bundle_digest"],
        "manifest_digest": current["manifest_digest"],
        "schema_version": 1,
        "version": current["version"],
    }))
    for expected_digest, target_digest in (
        ("c" * 64, str(target["manifest_digest"])),
        (str(current["manifest_digest"]), "d" * 64),
    ):
        wrong, _ = _run(
            "rollback", "--root", str(root), "--yes",
            "--expected-current-digest", expected_digest,
            "--target-digest", target_digest,
            "--trusted-key", f"release-key-1={public_hex}", expect=3,
        )
        assert wrong["error"]["code"] in {"RELEASE_REJECTED", "STATE_CONFLICT"}
    target_bundle = Path(str(target_checked["verified_bundle_path"]))
    manifest_path = target_bundle / "manifest.json"
    manifest_record = json.loads(manifest_path.read_text())
    manifest_path.write_text(json.dumps({**manifest_record, "signature": "0" * 128}))
    forged, _ = _run(
        "rollback", "--root", str(root), "--yes",
        "--expected-current-digest", str(current["manifest_digest"]),
        "--target-digest", str(target["manifest_digest"]),
        "--trusted-key", f"release-key-1={public_hex}", expect=3,
    )
    assert forged["error"]["code"] == "RELEASE_REJECTED"
    manifest_path.write_text(json.dumps(manifest_record))
    (target_bundle / str(target["artifact_name"])).write_bytes(b"tampered")
    rejected, _ = _run(
        "rollback", "--root", str(root), "--yes",
        "--expected-current-digest", str(current["manifest_digest"]),
        "--target-digest", str(target["manifest_digest"]),
        "--trusted-key", f"release-key-1={public_hex}", expect=3,
    )
    assert rejected["error"]["code"] == "RELEASE_REJECTED"
    assert json.loads((root / "releases" / "current.json").read_text())["manifest_digest"] == current["manifest_digest"]


def test_edge_compose_is_pull_only_and_hardened():
    model = yaml.safe_load(COMPOSE.read_text())
    assert set(model["services"]) == {"edge-gateway"}
    service = model["services"]["edge-gateway"]
    assert service["build"] == {
        "context": "../..", "dockerfile": "deploy/edge/Dockerfile.edge"
    }
    assert service["pull_policy"] == "${EDGE_PULL_POLICY:-build}"
    assert service["command"] == ["python", "-m", "app.edge_gateway"]
    assert "ports" not in service and "expose" not in service
    assert service["user"] not in {"0", 0, "root"}
    assert service["read_only"] is True
    assert service["cap_drop"] == ["ALL"]
    assert "no-new-privileges:true" in service["security_opt"]
    assert service["restart"] == "unless-stopped"
    assert service["pids_limit"] <= 256
    assert service["mem_limit"]
    assert service["cpus"]
    assert service["healthcheck"]["test"][0] == "CMD"
    assert service["user"] == "${EDGE_RUNTIME_UID:?}:${EDGE_RUNTIME_GID:?}"
    assert any(
        "uid=${EDGE_RUNTIME_UID:?},gid=${EDGE_RUNTIME_GID:?}" in item
        for item in service["tmpfs"]
    )
    volumes = "\n".join(service["volumes"])
    for marker in ("queue", "artifacts", "releases", "secrets", "certificates"):
        assert marker in volumes
    assert "/certificates:ro" in volumes and "/secrets:ro" in volumes
    env_text = ENV_EXAMPLE.read_text()
    assert "EDGE_CONTROL_ORIGIN=https://" in env_text
    assert "EDGE_ALLOWED_CONTROL_HOSTS=" in env_text
    assert "EDGE_RUNTIME_UID=" in env_text and "EDGE_RUNTIME_GID=" in env_text
    assert "EDGE_CLIENT_PRIVATE_KEY_PATH=/run/edge/secrets/client.key" in env_text
    assert service["environment"]["EDGE_POLL_INTERVAL_SECONDS"] == (
        "${EDGE_POLL_INTERVAL_SECONDS:-5}"
    )
    assert "EDGE_CREDENTIAL=" not in env_text
    assert "BEGIN PRIVATE KEY" not in env_text


def test_real_cli_private_material_is_readable_by_configured_container_uid(tmp_path):
    image = subprocess.run(
        ["docker", "image", "inspect", "nginx:alpine"], capture_output=True
    )
    if image.returncode != 0:
        pytest.skip("local nginx:alpine image is required; this test never pulls")
    root = tmp_path / "edge-root"
    _init(root)
    _run("csr", "--root", str(root), "--common-name", "edge-test")
    completed = subprocess.run(
        [
            "docker", "run", "--rm", "--user", f"{os.getuid()}:{os.getgid()}",
            "-v", f"{root / 'secrets'}:/run/edge/secrets:ro",
            "nginx:alpine", "sh", "-c", "test -r /run/edge/secrets/client.key",
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    assert completed.returncode == 0, completed.stderr


def test_openapi_contains_closed_edge_contract_without_secret_response_fields():
    document = yaml.safe_load(OPENAPI.read_text())
    paths = document["paths"]
    expected = {
        "/api/datasource/edge/enrollments",
        "/api/datasource/edge/register",
        "/api/datasource/edge/gateways",
        "/api/datasource/edge/gateways/{gateway_id}/heartbeat",
        "/api/datasource/edge/gateways/{gateway_id}/tasks/pull",
        "/api/datasource/edge/gateways/{gateway_id}/tasks/{task_id}/outcome",
        "/api/datasource/edge/gateways/{gateway_id}/events",
        "/api/datasource/edge/gateways/{gateway_id}/reconcile",
        "/api/datasource/edge/gateways/{gateway_id}/releases",
        "/api/datasource/edge/gateways/{gateway_id}/releases/ack",
    }
    assert expected <= set(paths)
    machine = paths["/api/datasource/edge/gateways/{gateway_id}/heartbeat"]["post"]
    assert machine["x-max-request-bytes"] == 262_144
    assert machine["x-max-json-depth"] == 32
    assert machine["x-max-json-nodes"] == 5_000
    assert {parameter["name"] for parameter in machine["parameters"]} >= {
        "gateway_id", "X-Edge-Credential", "X-Edge-Certificate-SHA256",
        "X-DataOps-Edge-Client-Verify", "X-DataOps-Edge-Client-Cert",
    }
    assert machine["responses"]["200"]["headers"]["Cache-Control"]["schema"]["const"] == "no-store"
    assert machine["security"] == [{"mutualTLS": [], "edgeCredential": []}]
    assert paths["/api/datasource/edge/enrollments"]["post"]["security"] == [
        {"bearerAuth": []}
    ]
    assert paths["/api/datasource/edge/gateways/{gateway_id}/rotate"]["post"][
        "security"
    ] == [{"bearerAuth": []}]
    assert "requestBody" not in paths["/api/datasource/edge/tasks/{task_id}/cancel"][
        "post"
    ]
    schemas = document["components"]["schemas"]
    for name in (
        "EdgeEnrollmentRequest", "EdgeRegisterRequest", "EdgeSignedTaskEnvelope",
        "EdgeEventRequest", "EdgeReconcileResponse", "EdgeReleaseManifest",
        "EdgeTaskOutcomeRequest", "EdgeErrorEnvelope",
    ):
        assert name in schemas
    rendered = json.dumps(schemas).casefold()
    assert "private_key" not in rendered
    assert "credential_secret" not in rendered
    assert "lineage" in schemas["EdgeTaskContract"]["properties"]["classification"]["enum"]
    assert schemas["EdgeSafeSummary"]["x-max-policy-depth"] == 12
    assert schemas["EdgeRegisterRequest"]["properties"]["version"] == {
        "$ref": "#/components/schemas/EdgeAgentVersion"
    }
    assert schemas["EdgeHeartbeatRequest"]["properties"]["version"] == {
        "$ref": "#/components/schemas/EdgeAgentVersion"
    }
    assert schemas["EdgeReleaseEnvelope"]["properties"]["data"]["properties"][
        "version"
    ] == {"$ref": "#/components/schemas/EdgeVersion"}
    assert schemas["EdgeReleaseEnvelope"]["properties"]["data"]["properties"][
        "rollback_version"
    ] == {"$ref": "#/components/schemas/EdgeVersion"}
    list_parameters = {
        item["name"]: item["schema"]
        for item in paths["/api/datasource/edge/gateways"]["get"]["parameters"]
    }
    assert list_parameters["limit"] == {
        "type": "integer", "minimum": 1, "maximum": 100
    }
    assert list_parameters["offset"] == {
        "type": "integer", "minimum": 0, "maximum": 10_000
    }
    heartbeat_sample = {
        "environment": "production", "network_zone": "zone-a", "generation": 1,
        "version": "1.0.0", "safe_summary": {"queue_depth": 0},
    }
    release_data = schemas["EdgeReleaseEnvelope"]["properties"]["data"]
    assert release_data["properties"]["manifest"] == {
        "$ref": "#/components/schemas/EdgeReleaseManifest"
    }
    security = document["components"]["securitySchemes"]
    assert security["mutualTLS"] == {"type": "mutualTLS"}
    assert security["edgeCredential"]["name"] == "X-Edge-Credential"

    def local_refs(value):
        if isinstance(value, dict):
            return {
                key: (
                    f"#/$defs/{item.rsplit('/', 1)[-1]}"
                    if key == "$ref"
                    and isinstance(item, str)
                    and item.startswith("#/components/schemas/")
                    else local_refs(item)
                )
                for key, item in value.items()
            }
        if isinstance(value, list):
            return [local_refs(item) for item in value]
        return value

    definitions = local_refs(schemas)
    heartbeat_validator = jsonschema.Draft202012Validator({
        "$defs": definitions, "$ref": "#/$defs/EdgeHeartbeatRequest"
    })
    heartbeat_validator.validate(heartbeat_sample)
    safe_validator = jsonschema.Draft202012Validator({
        "$defs": definitions, "$ref": "#/$defs/EdgeSafeSummary"
    })
    safe_validator.validate({"queue_depth": 0, "status": "healthy"})
    with pytest.raises(jsonschema.ValidationError):
        safe_validator.validate({"raw_rows": [{"secret": "x"}]})
    with pytest.raises(jsonschema.ValidationError):
        safe_validator.validate({"RawRows": [{"value": 1}]})
    with pytest.raises(jsonschema.ValidationError):
        safe_validator.validate({"PASSWORD": "masked"})
    too_deep: object = "leaf"
    for _ in range(13):
        too_deep = {"level": too_deep}
    with pytest.raises(jsonschema.ValidationError):
        safe_validator.validate({"summary": too_deep})
    with pytest.raises(jsonschema.ValidationError):
        safe_validator.validate({"message": "x" * 4097})
    cursor_validator = jsonschema.Draft202012Validator({
        "$defs": definitions, "$ref": "#/$defs/EdgeCursor"
    })
    with pytest.raises(jsonschema.ValidationError):
        cursor_validator.validate("../invalid cursor")


def test_docs_and_machine_evidence_preserve_enterprise_uat_boundary():
    required = [
        ROOT / "deploy" / "edge" / "README.md",
        ROOT / "docs" / "runbooks" / "EDGE_GATEWAY_OPERATIONS.md",
        ROOT / "docs" / "phase3" / "P3_WP04_EDGE_GATEWAY.md",
        ROOT / "docs" / "validation" / "P3_WP04_EDGE_GATEWAY_EVIDENCE.md",
    ]
    combined = "\n".join(path.read_text() for path in required)
    assert "ENGINEERING_COMPLETE_ENTERPRISE_EDGE_UAT_BLOCKED" in combined
    assert "TBD_EXTERNAL" in combined
    for topic in (
        "默认拒绝", "数据分级", "证书轮换", "证书撤销", "断网", "重连",
        "取消", "回滚", "诊断", "备份", "恢复", "CRL", "v1", "v2",
    ):
        assert topic in combined
    assert "真实企业来源已通过" not in combined
    failures = json.loads(FAILURES.read_text())
    assert failures["work_package"] == "P3-WP04"
    assert failures["engineering_status"] == "ENGINEERING_COMPLETE_ENTERPRISE_EDGE_UAT_BLOCKED"
    assert failures["cases"]
    required_keys = {
        "case_id", "scope", "preconditions", "steps", "safety_boundary",
        "expected", "evidence", "rollback", "status",
    }
    assert all(required_keys <= set(case) for case in failures["cases"])
    assert any(case["status"] == "TBD_EXTERNAL" for case in failures["cases"])


def test_operator_runbook_contains_executable_install_recovery_and_uat_gates():
    runbook = (ROOT / "docs/runbooks/EDGE_GATEWAY_OPERATIONS.md").read_text()
    for marker in (
        "Docker Engine 26", "Docker Compose 2.27", "Python 3.11",
        "sha256sum -c SHA256SUMS", "cosign verify-blob",
        "edge.config.schema.json", "docker image inspect",
        "health.json", "120 秒", "X-Edge-Enrollment", "X-Edge-Credential",
        "/tasks/{task_id}/outcome", "/reconcile", "/releases/ack",
        "edge_gateway_admin.py status", "人工介入阈值", "legal hold",
        "delete receipt", "sqlite3", "VACUUM INTO", "quick_check",
        "rollback-intent.json", "state/rollback-history/", "DB_ROLE_INIT_DATABASE_URL",
        "MIGRATION_DATABASE_URL", "has_schema_privilege", "CRL_CHECK_LEAF",
        "nginx -s reload", "TBD_EXTERNAL", "runtime-self-check",
        "format_checker=jsonschema.FormatChecker()", "healthy=0",
        "julianday(retention_until)<=julianday('now')",
        "PGSERVICE=dataops_migrator",
    ):
        assert marker in runbook
    assert "当前 queue schema 是 v1" not in runbook
    assert "20260809_477 (head)" not in runbook


def test_edge_config_example_matches_closed_runtime_file_schema():
    schema = json.loads(EDGE_CONFIG_SCHEMA.read_text())
    example = json.loads(EDGE_CONFIG_EXAMPLE.read_text())
    jsonschema.Draft202012Validator.check_schema(schema)
    jsonschema.Draft202012Validator(schema).validate(example)
    assert schema["additionalProperties"] is False
    assert set(schema["required"]) == set(schema["properties"])


def test_generator_is_reproducible_for_edge_contract(tmp_path):
    output = tmp_path / "openapi.yaml"
    subprocess.run(
        [sys.executable, str(ROOT / "scripts" / "generate_openapi.py"), "--output", str(output)],
        cwd=ROOT, check=True, capture_output=True, text=True,
    )
    assert output.read_bytes() == OPENAPI.read_bytes()


def test_release_archive_contains_exact_edge_cli_compose_and_runbook(tmp_path):
    assert (ROOT / "deployment/scripts/edge_gateway_admin.py").read_bytes() == CLI.read_bytes()
    completed = subprocess.run(
        ["bash", str(ROOT / "deployment/package_release.sh")],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=False,
        env={
            **os.environ,
            "DATAOPS_RELEASE_OUTPUT_DIR": str(tmp_path),
            "DATAOPS_RELEASE_VERSION": "wp04-test",
        },
    )
    assert completed.returncode == 0, completed.stderr
    archive = tmp_path / "dataops-platform-release-wp04-test.tar.gz"
    with tarfile.open(archive) as bundle:
        names = set(bundle.getnames())
        required = {
            "dataops-platform/scripts/edge_gateway_admin.py",
            "dataops-platform/deploy/edge/docker-compose.edge.yml",
            "dataops-platform/deploy/edge/Dockerfile.edge",
            "dataops-platform/deploy/edge/edge.config.example.json",
            "dataops-platform/deploy/edge/edge.config.schema.json",
            "dataops-platform/deploy/edge/edge.env.example",
            "dataops-platform/deploy/edge/README.md",
            "dataops-platform/docs/runbooks/EDGE_GATEWAY_OPERATIONS.md",
            "dataops-platform/docs/architecture/OPENAPI.yaml",
            "dataops-platform/docs/validation/P3_WP04_FAILURE_INJECTION.json",
            "dataops-platform/SHA256SUMS",
        }
        assert required <= names
        cli_member = bundle.extractfile("dataops-platform/scripts/edge_gateway_admin.py")
        assert cli_member is not None and cli_member.read() == CLI.read_bytes()
        assert not any("__pycache__" in name or name.endswith(".pyc") for name in names)
        readme_member = bundle.extractfile("dataops-platform/deploy/edge/README.md")
        assert readme_member is not None
        archive_readme = readme_member.read().decode("utf-8")
        assert "../../docs/runbooks/EDGE_GATEWAY_OPERATIONS.md" in archive_readme
        checksum_member = bundle.extractfile("dataops-platform/SHA256SUMS")
        assert checksum_member is not None
        checksum_lines = checksum_member.read().decode("utf-8").splitlines()
        assert checksum_lines
        for line in checksum_lines:
            digest, relative = line.split("  ", 1)
            member = bundle.extractfile(f"dataops-platform/{relative.removeprefix('./')}")
            assert member is not None
            assert hashlib.sha256(member.read()).hexdigest() == digest
        for name in names:
            member = bundle.extractfile(name)
            if member is None:
                continue
            content = member.read()
            assert b"-----BEGIN PRIVATE KEY-----" not in content
            assert b"-----BEGIN RSA PRIVATE KEY-----" not in content

    if shutil.which("docker") is not None:
        extracted = tmp_path / "extracted"
        extracted.mkdir()
        with tarfile.open(archive) as bundle:
            bundle.extractall(extracted, filter="data")
        release_root = extracted / "dataops-platform"
        build = subprocess.run(
            [
                "docker", "build", "-f", "deploy/edge/Dockerfile.edge",
                "-t", "dataops-edge:wp04-archive-smoke", ".",
            ],
            cwd=release_root,
            capture_output=True,
            text=True,
            check=False,
        )
        assert build.returncode == 0, build.stderr[-4000:]


def test_release_secret_scanner_rejects_binary_and_extended_secret_shapes(tmp_path):
    clean = tmp_path / "clean"
    clean.mkdir()
    (clean / "README.md").write_text("credential is mounted from a secret file\n")
    accepted = subprocess.run(
        [sys.executable, str(RELEASE_SECRET_SCANNER), str(clean)],
        capture_output=True,
        text=True,
        check=False,
    )
    assert accepted.returncode == 0, accepted.stderr

    cases = {
        "encrypted.pem": b"-----BEGIN ENCRYPTED PRIVATE KEY-----\nAAAA\n",
        "identity.p12": b"0\x82\x01\x00binary-pkcs12",
        "config.txt": b"credential = actual-production-value-123456\n",
    }
    for name, content in cases.items():
        candidate = tmp_path / name
        candidate.mkdir()
        (candidate / name).write_bytes(content)
        rejected = subprocess.run(
            [sys.executable, str(RELEASE_SECRET_SCANNER), str(candidate)],
            capture_output=True,
            text=True,
            check=False,
        )
        assert rejected.returncode != 0, name