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):
        cls.temp = tempfile.TemporaryDirectory(prefix="xfe-physical-fixture-")
        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)
        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"
        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")

    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 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])