XFESystem

"""Offline dual-media negative controls; mocks never qualify guest acceptance."""
import copy
import hashlib
import json
from pathlib import Path
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import Mock

import VerifyDualDelivery as dual


def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def read_json(path):
    return json.loads(Path(path).read_text(encoding='utf-8-sig'))


def write_json(path, value):
    Path(path).write_text(json.dumps(value, sort_keys=True), encoding='utf-8')


def require(condition, message):
    if not condition: raise ValueError(message)


class DualDeliveryGuardTests(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory()
        self.addCleanup(self.temporary.cleanup)
        self.folder = Path(self.temporary.name).resolve()
        desktop_folder = self.folder / 'Desktop'; desktop_folder.mkdir()
        console_folder = self.folder / 'Console'; console_folder.mkdir()
        self.desktop_path = desktop_folder / 'Result.json'
        self.console_path = console_folder / 'Result.json'
        self.image = self.folder / 'System.img'; self.image.write_bytes(b'OFFLINE TEST IMAGE - NO GUEST ACCEPTANCE')
        self.manifest_path = Path(str(self.image) + '.json')
        self.composite_path = self.folder / 'Dual.json'
        self.merged_path = self.folder / 'MergedSourceSnapshot.json'
        self.desktop_snapshot = desktop_folder / 'SourceSnapshot.json'
        self.console_snapshot = console_folder / 'SourceSnapshot.json'
        self.harnesses = ('Tools/IndependentKernel/TestBoot.py', 'Tools/IndependentKernel/VerifyDualDelivery.py',
                          'scripts/physical-media-verify.py', 'Tools/CSharpBoot/Program.cs', 'Tools/CSharpBoot/ReadImage.py')
        source_map = {name: digest(dual.ROOT / name) for name in self.harnesses}
        write_json(self.desktop_snapshot, source_map)
        write_json(self.console_snapshot, source_map)
        write_json(self.merged_path, source_map)
        self.inventory = {'System/Kernel.elf': 'a' * 64, 'System/Console.elf': 'b' * 64}
        self.manifest = {'Edition': 'Dual', 'BootEntries': ['Desktop', 'Console'], 'DefaultBootEntry': 'Desktop',
            'BootMenuTimeoutSeconds': 10, 'ConsoleKernelSha256': self.inventory['System/Console.elf'],
            'IndependentKernelBuild': {'Configuration': {'RamOnly': True, 'StorageEnabled': False}},
            'IndependentConsoleKernelBuild': {'Kind': 'XfeIndependentKernelBuild', 'Passed': True, 'Edition': 'Console',
                'ContainsCosmos': False, 'BootstrapOnly': False, 'Configuration': {'RamOnly': True, 'StorageEnabled': False}}}
        write_json(self.manifest_path, self.manifest)
        self.desktop = {'Edition': 'Desktop', 'KernelImagePath': 'System/Kernel.elf', 'BootEntry': 0,
            'SourcesUnchanged': True, 'SourceSnapshotPath': str(self.desktop_snapshot),
            'BootMenuSelection': {'Input': 'Actual QMP PS2 keyboard packets only', 'Index': 0, 'Entry': 'XFESystem Desktop', 'MenuSerialOffset': 100}}
        self.console = {'Kind': 'XfeIndependentKernelBootProof', 'Passed': True, 'Edition': 'Console',
            'KernelImagePath': 'System/Console.elf', 'BootEntry': 1, 'ConsoleAcceptanceRequired': True,
            'ConsoleAcceptancePassed': True, 'ManagedRuntimeAcceptancePassed': True, 'ContainsCosmos': False,
            'ImageUnchanged': True, 'QemuExitCode': 0, 'SourcesUnchanged': True,
            'ImageSha256': digest(self.image), 'KernelSha256': self.inventory['System/Console.elf'],
            'ImageVerification': {'Files': dict(self.inventory)}, 'SourceSnapshotPath': str(self.console_snapshot),
            'SourceSnapshotSha256': digest(self.console_snapshot),
            'BootMenuSelection': {'Input': 'Actual QMP PS2 keyboard packets only', 'Index': 1, 'Entry': 'XFESystem Console', 'MenuSerialOffset': 100}}
        write_json(self.desktop_path, self.desktop); write_json(self.console_path, self.console)
        self.capture = console_folder / 'Console-Commands.png'; self.capture.write_bytes(b'OFFLINE IMAGE FIXTURE')
        self.serial = console_folder / 'Serial.log'; self.serial.write_text('XFE_SYSTEM_EDITION|Console\n', encoding='utf-8')
        self.verified = {'ImageSha256': digest(self.image), 'ImageManifestSha256': digest(self.manifest_path),
            'SourceSnapshotPath': str(self.desktop_snapshot), 'IndependentImageInspection': {'Files': dict(self.inventory)}}
        self.media = SimpleNamespace(require=require, digest=digest, read_json=read_json,
            read_bound_json=lambda path: (read_json(path), digest(path)),
            verify=Mock(return_value=self.verified), verify_independent_delivery=Mock(return_value={'Passed': True}))
        self.composite = {'Kind': 'XfeIndependentDualBootProof', 'Passed': True, 'ImageSha256': digest(self.image),
            'DesktopBoot': self.bound(self.desktop_path), 'ConsoleBoot': self.bound(self.console_path),
            'MergedSourceSnapshot': self.bound(self.merged_path), 'FullReleaseQualified': False}
        self.rebind()

    def bound(self, path):
        return {'Path': str(path), 'Sha256': digest(path)}

    def rebind(self):
        self.composite['DesktopBoot'] = self.bound(self.desktop_path)
        self.composite['ConsoleBoot'] = self.bound(self.console_path)
        self.composite['MergedSourceSnapshot'] = self.bound(self.merged_path)
        self.composite['EvidenceFiles'] = dual.evidence_files((self.desktop_path, self.console_path), self.media)
        write_json(self.composite_path, self.composite)

    def change_console(self, **changes):
        self.console.update(changes); write_json(self.console_path, self.console); self.rebind()

    def verify(self):
        return dual.verify_dual(self.image, self.composite_path, self.media)

    def test_complete_structure_calls_both_strict_existing_gates(self):
        result = self.verify()
        self.assertTrue(result['DualEditionAcceptancePassed'])
        self.assertEqual('IndependentDualCSharp256', result['Profile'])
        self.media.verify.assert_called_once_with(self.image, self.desktop_path, 'IndependentCSharp128', _dual_image=True)
        self.assertEqual('Console', self.media.verify_independent_delivery.call_args.args[-1])
        self.assertEqual(self.composite['EvidenceFiles'], result['AdditionalEvidenceLocks'])

    def test_nested_boot_receipt_changed_after_composite_is_rejected(self):
        self.console['Passed'] = False; write_json(self.console_path, self.console)
        with self.assertRaisesRegex(ValueError, 'artifact changed'): self.verify()

    def test_rehashed_failed_console_and_forced_cleanup_still_fail(self):
        for changes in ({'Passed': False}, {'ForcedCleanup': True}, {'QemuExitCode': True},
                        {'ConsoleAcceptancePassed': False}, {'ManagedRuntimeAcceptancePassed': False}):
            original = copy.deepcopy(self.console)
            with self.subTest(changes=changes):
                self.change_console(**changes)
                with self.assertRaises(ValueError): self.verify()
                self.console = original; write_json(self.console_path, original); self.rebind()

    def test_rehashed_wrong_menu_kernel_or_edition_cannot_qualify(self):
        for changes in ({'BootEntry': 0}, {'KernelImagePath': 'System/Kernel.elf'}, {'Edition': 'Desktop'}):
            original = copy.deepcopy(self.console)
            with self.subTest(changes=changes):
                self.change_console(**changes)
                with self.assertRaises(ValueError): self.verify()
                self.console = original; write_json(self.console_path, original); self.rebind()

    def test_boolean_menu_index_missing_keyboard_selection_and_changed_sources_fail(self):
        for changes in ({'BootEntry': True}, {'BootMenuSelection': {}}, {'SourcesUnchanged': False},
                        {'BootMenuSelection': {'Input': 'Host command injection', 'Index': 1, 'Entry': 'XFESystem Console', 'MenuSerialOffset': 100}},
                        {'BootMenuSelection': {'Input': 'Actual QMP PS2 keyboard packets only', 'Index': True, 'Entry': 'XFESystem Console', 'MenuSerialOffset': 100}}):
            original = copy.deepcopy(self.console)
            with self.subTest(changes=changes):
                self.change_console(**changes)
                with self.assertRaises(ValueError): self.verify()
                self.console = original; write_json(self.console_path, original); self.rebind()

    def test_console_hash_must_match_actual_embedded_elf_inventory(self):
        self.manifest['ConsoleKernelSha256'] = 'c' * 64; write_json(self.manifest_path, self.manifest)
        self.verified['ImageManifestSha256'] = digest(self.manifest_path)
        self.change_console(KernelSha256='c' * 64)
        with self.assertRaisesRegex(ValueError, 'exact dual image inventory'): self.verify()

    def test_selected_nested_capture_cannot_change_after_composite(self):
        self.capture.write_bytes(b'CHANGED PIXELS')
        with self.assertRaisesRegex(ValueError, 'complete evidence set'): self.verify()

    def test_evidence_lock_list_cannot_be_omitted_or_partially_selected(self):
        baseline = list(self.composite['EvidenceFiles'])
        for selected in ([], baseline[:-1]):
            with self.subTest(count=len(selected)):
                self.composite['EvidenceFiles'] = selected; write_json(self.composite_path, self.composite)
                with self.assertRaisesRegex(ValueError, 'complete evidence set'): self.verify()

    def test_console_source_snapshot_bytes_are_bound(self):
        self.console_snapshot.write_text('{}', encoding='utf-8')
        self.rebind()
        with self.assertRaisesRegex(ValueError, 'Console source snapshot changed'): self.verify()

    def test_merged_snapshot_must_be_exact_union(self):
        merged = read_json(self.merged_path); merged['Unrelated.cs'] = 'f' * 64
        write_json(self.merged_path, merged); self.rebind()
        with self.assertRaisesRegex(ValueError, 'exact union'): self.verify()

    def test_mandatory_safety_verifier_cannot_be_omitted_from_both_snapshots(self):
        reduced = read_json(self.console_snapshot); del reduced['Tools/IndependentKernel/VerifyDualDelivery.py']
        for path in (self.desktop_snapshot, self.console_snapshot, self.merged_path): write_json(path, reduced)
        self.console['SourceSnapshotSha256'] = digest(self.console_snapshot)
        write_json(self.console_path, self.console); self.rebind()
        with self.assertRaisesRegex(ValueError, 'Missing or stale dual delivery harness'): self.verify()

    def test_manifest_mutation_during_console_gate_is_rejected(self):
        def change(*unused):
            changed = dict(self.manifest); changed['BootMenuTimeoutSeconds'] = 1
            write_json(self.manifest_path, changed)
            return {'Passed': True}
        self.media.verify_independent_delivery.side_effect = change
        with self.assertRaisesRegex(ValueError, 'changed during verification'): self.verify()

    def test_failures_from_either_strict_edition_gate_are_propagated(self):
        self.media.verify.side_effect = ValueError('desktop actual HTTPS verification failed')
        with self.assertRaisesRegex(ValueError, 'desktop actual HTTPS'): self.verify()
        self.media.verify.side_effect = None
        self.media.verify_independent_delivery.side_effect = ValueError('console actual command verification failed')
        with self.assertRaisesRegex(ValueError, 'console actual command'): self.verify()


if __name__ == '__main__':
    unittest.main()