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", "IndependentCSharpUsb256"), 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_bytes = Path(str(original) + ".json").read_bytes()
        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 in ("IndependentCSharp128", "IndependentCSharpUsb256"):
            # 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 in ("BingEvidence.json", "GpuDiscoveryEvidence.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.storage_original = None
        if arguments.profile == 'IndependentCSharpUsb256':
            original_proof = Path(cls.boot_original['UsbStorageProof']['Path'])
            cls.storage_original = original_proof.read_bytes()
            cls.storage_path = folder / 'UsbStorage/Result.json'
            cls.storage_path.parent.mkdir()
            for item in json.loads(cls.storage_original)['EvidenceFiles']:
                target = cls.storage_path.parent / item['Path']
                target.parent.mkdir(parents=True, exist_ok=True)
                shutil.copyfile(original_proof.parent / item['Path'], target)
            cls.boot_original['UsbStorageProof']['Path'] = str(cls.storage_path)
        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):
        # The actual display proof binds the complete manifest bytes. The
        # positive copy must retain them; negative tests mutate explicitly.
        self.manifest.write_bytes(self.manifest_original_bytes)
        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)
        if self.storage_original is not None:
            self.storage_path.write_bytes(self.storage_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"])
        persistent = arguments.profile == 'IndependentCSharpUsb256'
        self.assertEqual(result["ImageBytes"], (256 if persistent else 128) * 1024 * 1024)
        for field in ("FullReleaseQualified", "StabilityPassed", "CudaExecutionVerified", "GpuRasterizationVerified"):
            self.assertIs(result[field], False)
        self.assertTrue(result["IndependentImageInspection"]["FatMirrorsVerified"])
        self.assertIs(result["RamOnlyConfigurationVerified"], not persistent)
        self.assertIs(result["StorageEnabled"], persistent)
        self.assertIs(result["PhysicalInputHardwareAcceptancePassed"], False)

    def test_persistent_delivery_refuses_fixed_virtual_monitor_resolution(self):
        if arguments.profile != 'IndependentCSharpUsb256':
            self.skipTest('Automatic-monitor policy applies to new persistent USB deliveries')
        altered = copy.deepcopy(self.manifest_original)
        altered['Resolution'] = '1280x800'
        self.manifest.write_text(json.dumps(altered), encoding='utf-8')
        with self.assertRaisesRegex(ValueError, 'automatic firmware/monitor resolution'):
            self.verify()

    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)
    @unittest.skipUnless(arguments.profile == 'IndependentCSharpUsb256', 'USB persistence delivery')
    def test_missing_actual_usb_acceptance_is_refused(self):
        self.reject_boot(UsbStorageAcceptancePassed=False)
        self.reject_boot(UsbStorageProof=None)

    @unittest.skipUnless(arguments.profile == 'IndependentCSharpUsb256', 'USB persistence delivery')
    def test_rebound_forged_cold_boot_and_program_proofs_are_refused(self):
        for change in ('generation', 'program', 'snapshot'):
            with self.subTest(change=change):
                proof = json.loads(self.storage_original)
                if change == 'generation': proof['Boots'][1]['LoadedGeneration'] = 0
                elif change == 'program': proof['StandardNetPeReadFromFatAndExecuted'] = False
                else:
                    args = proof['Boots'][1]['QemuArguments']
                    index = next(i for i, arg in enumerate(args) if arg.startswith('if=none,id=BootUsb,'))
                    args[index] += ',snapshot=on'
                self.storage_path.write_text(json.dumps(proof), encoding='utf-8')
                boot = copy.deepcopy(self.boot_original)
                boot['UsbStorageProof']['Sha256'] = media.digest(self.storage_path)
                self.boot.write_text(json.dumps(boot), encoding='utf-8')
                with self.assertRaises((ValueError, AssertionError)): self.verify()

    @unittest.skipUnless(arguments.profile == 'IndependentCSharpUsb256', 'USB persistence delivery')
    def test_usb_framebuffer_and_rendered_commands_are_bound(self):
        item = next(item for item in json.loads(self.storage_original)['EvidenceFiles'] if item['Path'].endswith('RenderedCommands.log'))
        path = self.storage_path.parent / item['Path']; original = path.read_bytes()
        try:
            path.write_bytes(original + b'changed')
            with self.assertRaises((ValueError, AssertionError)): self.verify()
        finally: path.write_bytes(original)
    def test_independent_render_failure_is_refused(self):
        if arguments.profile not in ("IndependentCSharp128", "IndependentCSharpUsb256"): 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 not in ("IndependentCSharp128", "IndependentCSharpUsb256"): 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 not in ("IndependentCSharp128", "IndependentCSharpUsb256"): 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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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 in ("IndependentCSharp128", "IndependentCSharpUsb256"), "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])