XFESystem

"""Test the experimental media verifier using copies of an actual boot image.

No physical device is enumerated or opened. The input image/proof are read-only;
all negative fixtures live in a regular temporary directory.
"""
import argparse
import copy
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import struct
import sys
import tempfile
import unittest
import zlib

parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--image", required=True)
parser.add_argument("--boot-result", required=True)
parser.add_argument("--profile", choices=("CSharpExperimental128", "IndependentCSharp128"), default="CSharpExperimental128")
arguments, remaining = parser.parse_known_args()
spec = importlib.util.spec_from_file_location("experimental_media", Path(__file__).with_name("physical-media-verify.py"))
media = importlib.util.module_from_spec(spec)
spec.loader.exec_module(media)


class ExperimentalMediaTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        temporary_root = media.ROOT / "build/IndependentKernel/PhysicalMediaGuardTests"
        temporary_root.mkdir(parents=True, exist_ok=True)
        cls.temp = tempfile.TemporaryDirectory(prefix="xfe-physical-fixture-", dir=temporary_root)
        folder = Path(cls.temp.name)
        original = Path(arguments.image).resolve()
        cls.original_hash = media.digest(original)
        cls.original = original
        cls.image = folder / "System.img"
        shutil.copyfile(original, cls.image)
        cls.manifest_original = media.read_json(str(original) + ".json")
        cls.boot_original = media.read_json(arguments.boot_result)
        cls.original_boot = Path(arguments.boot_result).resolve()
        cls.original_boot_hash = media.digest(cls.original_boot)
        cls.bing_original = None
        if arguments.profile == "IndependentCSharp128":
            # The production strict launcher verifies actual guest PNG/log
            # artifacts beside this copied receipt. Copy their unchanged bytes
            # once; no physical disk or original proof is modified by tests.
            for source in cls.original_boot.parent.iterdir():
                if source.is_file() and (source.suffix.lower() in (".png", ".log", ".txt") or source.name == "BingEvidence.json"):
                    shutil.copyfile(source, folder / source.name)
            cls.bing_path = folder / "BingEvidence.json"
            cls.bing_original = cls.bing_path.read_bytes()
        snapshot = Path(cls.boot_original["SourceSnapshotPath"])
        if not snapshot.is_absolute():
            snapshot = Path(arguments.boot_result).resolve().parent / snapshot
        cls.snapshot = folder / "SourceSnapshot.json"
        shutil.copyfile(snapshot, cls.snapshot)
        cls.boot_original["SourceSnapshotPath"] = str(cls.snapshot)
        cls.manifest = Path(str(cls.image) + ".json")
        cls.boot = folder / "Boot.json"

    @classmethod
    def tearDownClass(cls):
        assert media.digest(cls.original) == cls.original_hash, "The input image changed"
        assert media.digest(cls.original_boot) == cls.original_boot_hash, "The input boot proof changed"
        cls.temp.cleanup()

    def setUp(self):
        self.manifest.write_text(json.dumps(self.manifest_original), encoding="utf-8")
        self.boot.write_text(json.dumps(self.boot_original), encoding="utf-8")
        if self.bing_original is not None:
            self.bing_path.write_bytes(self.bing_original)

    def verify(self):
        return media.verify(self.image, self.boot, arguments.profile)

    def reject_boot(self, **changes):
        altered = copy.deepcopy(self.boot_original)
        altered.update(changes)
        self.boot.write_text(json.dumps(altered), encoding="utf-8")
        with self.assertRaises((ValueError, AssertionError)):
            self.verify()

    def reject_manifest(self, **changes):
        altered = copy.deepcopy(self.manifest_original)
        altered.update(changes)
        self.manifest.write_text(json.dumps(altered), encoding="utf-8")
        with self.assertRaises((ValueError, AssertionError)):
            self.verify()

    def corrupt_range(self, offset, replacement):
        with self.image.open("r+b") as stream:
            stream.seek(offset)
            original = stream.read(len(replacement))
            stream.seek(offset)
            stream.write(replacement)
        try:
            with self.assertRaises((ValueError, AssertionError)):
                self.verify()
        finally:
            with self.image.open("r+b") as stream:
                stream.seek(offset)
                stream.write(original)

    def test_actual_csharp_image_and_evidence_bind_without_release_claim(self):
        result = self.verify()
        self.assertEqual(result["ImageSha256"], self.original_hash)
        self.assertEqual(result["KernelSha256"], self.boot_original["KernelSha256"])
        self.assertEqual(result["ImageBytes"], 128 * 1024 * 1024)
        for field in ("FullReleaseQualified", "StabilityPassed", "CudaExecutionVerified", "GpuRasterizationVerified"):
            self.assertIs(result[field], False)
        self.assertTrue(result["IndependentImageInspection"]["FatMirrorsVerified"])
        self.assertIs(result["RamOnlyConfigurationVerified"], True)
        self.assertIs(result["StorageEnabled"], False)
        self.assertIs(result["PhysicalInputHardwareAcceptancePassed"], False)

    def test_ram_only_configuration_is_source_bound_and_unconditional(self):
        project = Path(self.temp.name) / "Kernel.csproj"
        relative = "Source/XFESystem.Kernel/XFESystem.Kernel.csproj"
        project.write_text("<Project><PropertyGroup><CosmosEnableStorage>false</CosmosEnableStorage></PropertyGroup></Project>", encoding="utf-8")
        snapshot = {relative: media.digest(project)}
        self.assertIs(media.verify_ram_only_project(snapshot, project)["StorageEnabled"], False)
        project.write_text("<Project><PropertyGroup><CosmosEnableStorage>true</CosmosEnableStorage></PropertyGroup></Project>", encoding="utf-8")
        with self.assertRaisesRegex(ValueError, "differs from the compiled"):
            media.verify_ram_only_project(snapshot, project)
        snapshot[relative] = media.digest(project)
        with self.assertRaisesRegex(ValueError, "CosmosEnableStorage=false"):
            media.verify_ram_only_project(snapshot, project)
        project.write_text("<Project><PropertyGroup Condition=\"'x'=='x'\"><CosmosEnableStorage>false</CosmosEnableStorage></PropertyGroup></Project>", encoding="utf-8")
        snapshot[relative] = media.digest(project)
        with self.assertRaisesRegex(ValueError, "unconditional"):
            media.verify_ram_only_project(snapshot, project)

    def test_x64_native_binary_header_guards(self):
        elf = bytearray(64)
        elf[:7] = b"\x7fELF\x02\x01\x01"
        struct.pack_into("<HH", elf, 16, 2, 62)
        media.verify_elf(elf)
        elf[4] = 1
        with self.assertRaises(ValueError): media.verify_elf(elf)
        pe = bytearray(512)
        pe[:2] = b"MZ"
        struct.pack_into("<I", pe, 60, 64)
        pe[64:68] = b"PE\0\0"
        struct.pack_into("<HH", pe, 68, 0x8664, 1)
        struct.pack_into("<H", pe, 84, 240)
        struct.pack_into("<H", pe, 88, 0x20b)
        struct.pack_into("<H", pe, 156, 10)
        media.verify_efi(pe, "fixture")
        struct.pack_into("<H", pe, 156, 3)
        with self.assertRaises(ValueError): media.verify_efi(pe, "fixture")

    def test_failed_boot_is_refused(self): self.reject_boot(Passed=False)
    def test_independent_render_failure_is_refused(self):
        if arguments.profile != "IndependentCSharp128": self.skipTest("Independent profile only")
        input_proof = copy.deepcopy(self.boot_original["DesktopInputProof"])
        input_proof["NoRenderFailures"] = False
        self.reject_boot(DesktopInputProof=input_proof)

    def test_independent_model_only_is_refused(self):
        if arguments.profile != "IndependentCSharp128": self.skipTest("Independent profile only")
        input_proof = copy.deepcopy(self.boot_original["DesktopInputProof"])
        input_proof.pop("MonacoRasterPaintCompleted", None)
        self.reject_boot(DesktopInputProof=input_proof)

    def test_independent_unedited_model_is_refused(self):
        if arguments.profile != "IndependentCSharp128": self.skipTest("Independent profile only")
        input_proof = copy.deepcopy(self.boot_original["DesktopInputProof"])
        input_proof["MonacoKeyboardEditedModel"] = False
        self.reject_boot(DesktopInputProof=input_proof)
    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent final delivery")
    def test_independent_bing_documents_and_terminal_are_mandatory(self):
        for changes in ({"BingAcceptanceRequired": False}, {"BingAcceptancePassed": False},
                        {"DocumentAcceptanceRequired": False}, {"DocumentAcceptancePassed": False}):
            with self.subTest(changes=changes): self.reject_boot(**changes)
        proof = copy.deepcopy(self.boot_original["DesktopInputProof"])
        proof.pop("TerminalWorkflowProof", None)
        self.reject_boot(DesktopInputProof=proof)

    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent final delivery")
    def test_independent_exact_nine_cli_commands_are_required(self):
        proof = copy.deepcopy(self.boot_original["DesktopInputProof"])
        proof["TerminalWorkflowProof"]["Commands"].pop()
        self.reject_boot(DesktopInputProof=proof)

    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent final delivery")
    def test_independent_bing_body_dom_and_actual_present_bindings_are_required(self):
        for name in ("body", "present"):
            with self.subTest(change=name):
                document = json.loads(self.bing_original)
                if name == "body": document["RemotePage"]["BodySha256"] = "0" * 64
                else: document["Screenshots"][0]["PresentationCorrelation"]["After"]["SnapshotSequence"] += 1
                self.bing_path.write_text(json.dumps(document), encoding="utf-8")
                # Rebind the JSON file hash deliberately: validation must reject
                # its semantic mismatch, rather than only the outer file hash.
                proof = copy.deepcopy(self.boot_original)
                proof["DesktopInputProof"]["BingEvidence"]["Sha256"] = media.digest(self.bing_path)
                self.boot.write_text(json.dumps(proof), encoding="utf-8")
                with self.assertRaises((ValueError, AssertionError)): self.verify()

    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent final delivery")
    def test_independent_bing_and_cli_framebuffers_are_bound_by_hash(self):
        bing = json.loads(self.bing_original)
        paths = (bing["Screenshots"][0]["Path"], self.boot_original["DesktopInputProof"]["TerminalWorkflowProof"]["Screenshot"]["Path"])
        for relative in paths:
            with self.subTest(path=relative):
                path = self.boot.parent / relative; original = path.read_bytes()
                try:
                    path.write_bytes(original + b"changed")
                    with self.assertRaises((ValueError, AssertionError)): self.verify()
                finally: path.write_bytes(original)
    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent kernel evidence")
    def test_bootstrap_diagnostic_is_refused(self):
        changed = copy.deepcopy(self.manifest_original)
        changed["IndependentKernelBuild"]["BootstrapOnly"] = True
        self.manifest.write_text(json.dumps(changed), encoding="utf-8")
        with self.assertRaisesRegex(ValueError, "complete independent application"):
            self.verify()
    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent kernel evidence")
    def test_missing_managed_desktop_execution_is_refused(self):
        self.reject_boot(DesktopAcceptancePassed=False)
        self.reject_boot(ManagedRuntimeAcceptancePassed=False)
    @unittest.skipUnless(arguments.profile == "IndependentCSharp128", "Independent kernel evidence")
    def test_cosmos_linked_kernel_is_refused(self):
        changed = copy.deepcopy(self.manifest_original)
        changed["IndependentKernelBuild"]["ContainsCosmos"] = True
        self.manifest.write_text(json.dumps(changed), encoding="utf-8")
        with self.assertRaisesRegex(ValueError, "complete independent application"):
            self.verify()
    def test_string_true_is_refused(self): self.reject_boot(Passed="true")
    def test_unclean_qemu_is_refused(self): self.reject_boot(QemuExitCode=1)
    def test_bool_exitcode_is_refused(self): self.reject_boot(QemuExitCode=False)
    def test_changed_image_receipt_is_refused(self): self.reject_boot(ImageUnchanged=False)
    def test_wrong_boot_image_hash_is_refused(self): self.reject_boot(ImageSha256="0" * 64)
    def test_wrong_boot_kernel_hash_is_refused(self): self.reject_boot(KernelSha256="0" * 64)
    def test_wrong_source_snapshot_hash_is_refused(self): self.reject_boot(SourceSnapshotSha256="0" * 64)
    def test_wrong_pack_image_hash_is_refused(self): self.reject_manifest(ImageSha256="0" * 64)
    def test_wrong_pack_kernel_hash_is_refused(self): self.reject_manifest(KernelSha256="0" * 64)
    def test_legacy_manifest_is_refused(self): self.reject_manifest(Kind="LegacyRustPhysicalImage")
    def test_unverified_pack_readback_is_refused(self): self.reject_manifest(FullImageReadback=False)
    def test_hybrid_mbr_is_refused(self): self.corrupt_range(462, b"\x01")
    def test_fat_mirror_corruption_is_refused(self):
        with self.image.open("rb") as stream:
            stream.seek(2048 * 512)
            bpb = stream.read(512)
            reserved = struct.unpack_from("<H", bpb, 14)[0]
            sectors = struct.unpack_from("<I", bpb, 36)[0]
            at = (2048 + reserved + sectors) * 512 + 8
            stream.seek(at)
            byte = stream.read(1)[0]
        self.corrupt_range(at, bytes([byte ^ 1]))

    def test_extra_valid_gpt_partition_is_refused(self):
        with self.image.open("rb") as stream:
            stream.seek(1024)
            old_table = stream.read(16384)
            stream.seek(512)
            old_primary = stream.read(512)
            stream.seek(-512, 2)
            old_backup = stream.read(512)
        table = bytearray(old_table)
        table[128:256] = table[:128]
        table[128 + 16] ^= 1
        primary, backup = bytearray(old_primary), bytearray(old_backup)
        for header in (primary, backup):
            struct.pack_into("<I", header, 88, zlib.crc32(table))
            struct.pack_into("<I", header, 16, 0)
            struct.pack_into("<I", header, 16, zlib.crc32(header[:92]))
        last = self.image.stat().st_size // 512 - 1
        offsets = [(1024, old_table, table), ((last - 32) * 512, old_table, table),
                   (512, old_primary, primary), (last * 512, old_backup, backup)]
        try:
            with self.image.open("r+b") as stream:
                for offset, _, replacement in offsets:
                    stream.seek(offset); stream.write(replacement)
            with self.assertRaisesRegex(ValueError, "exactly one ESP"):
                self.verify()
        finally:
            with self.image.open("r+b") as stream:
                for offset, original, _ in offsets:
                    stream.seek(offset); stream.write(original)


if __name__ == "__main__":
    unittest.main(argv=[sys.argv[0], *remaining])