XFESystem

"""Offline controls for the independent USB delivery gate; no VM/disk is used.

The PowerShell verifier is mocked only to test dispatch and fail-closed guards.
These tiny fixtures are not boot evidence or physical-media acceptance.
"""
import base64
import importlib.util
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("physical_media_verify", ROOT / "scripts/physical-media-verify.py")
verifier = importlib.util.module_from_spec(spec)
spec.loader.exec_module(verifier)


class IndependentDeliveryGateTests(unittest.TestCase):
    def setUp(self):
        directory = ROOT / 'build/IndependentKernel/DeliveryGateGuardTests'; directory.mkdir(parents=True, exist_ok=True)
        self.temporary = tempfile.TemporaryDirectory(dir=directory)
        self.addCleanup(self.temporary.cleanup)
        folder = Path(self.temporary.name)
        self.image = folder / "System-'`$Name.img"; self.image.write_bytes(b'Not a native image; verifier guard fixture only')
        self.boot = folder / 'Result.json'; self.boot.write_bytes(b'Not a boot receipt; verifier guard fixture only')
        self.snapshot = {relative: verifier.digest(ROOT / relative) for relative in
                         ('Tools/IndependentKernel/TestBoot.py', 'Tools/IndependentKernel/IndependentBingEvidence.py',
                          'Tools/IndependentKernel/IndependentDocumentEvidence.py')}
        self.image_hash, self.boot_hash = verifier.digest(self.image), verifier.digest(self.boot)

    def invoke(self):
        return verifier.verify_independent_delivery(self.image, self.boot, self.snapshot, self.image_hash, self.boot_hash)

    def test_uses_actual_readonly_launcher_and_structured_arguments_with_full_bindings(self):
        oracle = subprocess.CompletedProcess([], 0, 'XFE_INDEPENDENT_DELIVERY_GATE_OK\n', '')
        with patch.object(verifier.shutil, 'which', return_value='powershell.exe'), patch.object(verifier.subprocess, 'run', return_value=oracle) as run:
            result = self.invoke()
        self.assertTrue(result['Passed']); self.assertTrue(result['NoVmStarted']); self.assertTrue(result['NoDiskWrites'])
        self.assertTrue(result['BingNativeHttpsPassed']); self.assertTrue(result['DocumentWorkflowPassed']); self.assertTrue(result['TerminalWorkflowPassed'])
        self.assertFalse(result['FullCssParityVerified'])
        self.assertEqual(self.image_hash, result['ImageSha256']); self.assertEqual(self.boot_hash, result['BootResultSha256'])
        args = run.call_args.args[0]
        self.assertIn('-NonInteractive', args); self.assertIn('-EncodedCommand', args)
        script = base64.b64decode(args[-1]).decode('utf-16-le')
        self.assertIn('-ValidateOnly', script); self.assertNotIn('-PlanOnly', script); self.assertNotIn('-Write', script)
        self.assertNotIn(str(self.image), script)  # It exists only in the encoded JSON payload.
        self.assertEqual(60, run.call_args.kwargs['timeout'])
        self.assertEqual(ROOT, run.call_args.kwargs['cwd'])

    def test_missing_or_stale_boot_evidence_harness_is_rejected_before_powershell(self):
        for changed in ('missing', 'stale'):
            with self.subTest(changed=changed):
                current = dict(self.snapshot)
                if changed == 'missing': del current['Tools/IndependentKernel/IndependentBingEvidence.py']
                else: current['Tools/IndependentKernel/TestBoot.py'] = '0' * 64
                with patch.object(verifier.subprocess, 'run') as run, self.assertRaisesRegex(ValueError, 'evidence harness'):
                    verifier.verify_independent_delivery(self.image, self.boot, current, self.image_hash, self.boot_hash)
                run.assert_not_called()

    def test_false_boolean_receipt_without_successful_real_verifier_cannot_qualify(self):
        for oracle in (subprocess.CompletedProcess([], 1, '', 'Bing actual presentation/PNG verification failed'),
                       subprocess.CompletedProcess([], 0, 'No strict verifier acknowledgement', ''),
                       subprocess.CompletedProcess([], 1, 'XFE_INDEPENDENT_DELIVERY_GATE_OK\n', 'Rejected')):
            with self.subTest(oracle=oracle), patch.object(verifier.shutil, 'which', return_value='powershell.exe'), \
                 patch.object(verifier.subprocess, 'run', return_value=oracle), self.assertRaisesRegex(ValueError, 'delivery verification failed'):
                self.invoke()

    def test_missing_powershell_or_timeout_does_not_bypass_acceptance(self):
        with patch.object(verifier.shutil, 'which', return_value=None), patch.object(verifier.subprocess, 'run') as run, \
             self.assertRaisesRegex(ValueError, 'Windows PowerShell'):
            self.invoke()
        run.assert_not_called()
        with patch.object(verifier.shutil, 'which', return_value='powershell.exe'), \
             patch.object(verifier.subprocess, 'run', side_effect=subprocess.TimeoutExpired('read-only verifier', 60)), \
             self.assertRaises(subprocess.TimeoutExpired):
            self.invoke()

    def test_image_or_receipt_changed_during_verification_is_rejected(self):
        for name in ('image', 'boot'):
            with self.subTest(name=name):
                path = getattr(self, name); original = path.read_bytes()
                def oracle(*_, **__):
                    path.write_bytes(b'Changed during the verifier')
                    return subprocess.CompletedProcess([], 0, 'XFE_INDEPENDENT_DELIVERY_GATE_OK\n', '')
                try:
                    with patch.object(verifier.shutil, 'which', return_value='powershell.exe'), \
                         patch.object(verifier.subprocess, 'run', side_effect=oracle), self.assertRaisesRegex(ValueError, 'changed during'):
                        self.invoke()
                finally: path.write_bytes(original)


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