XFESystem

"""Offline negative controls for the Bing evidence reader; not guest acceptance."""
import copy
import hashlib
import json
from pathlib import Path
import subprocess
import tempfile
import time
import unittest

from PIL import Image, ImageDraw

from IndependentBingEvidence import (HOME, NIC_ARGUMENT, NAT_ARGUMENT, REMOTE, RESPONSE, json_blocks,
    matching_presentation, native_network_profile, network_arguments, presentation_identity,
    native_navigation_failure, remote_text_pixels, run_bing, stable_screenshot_presentation, validate_network_arguments,
    validate_remote_page, validate_response_chain)

ROOT = Path(__file__).resolve().parents[2]
TEMP_ROOT = ROOT / 'build' / 'IndependentKernel' / 'ToolGuardTests'


def temporary_directory():
    TEMP_ROOT.mkdir(parents=True, exist_ok=True)
    return tempfile.TemporaryDirectory(dir=TEMP_ROOT)


def response(uri=HOME, status=200, location=None):
    from urllib.parse import urlsplit
    url = urlsplit(uri)
    target = (url.path or '/') + ('?' + url.query if url.query else '')
    headers = 'HTTP/1.1 ' + str(status) + ' OK\r\nContent-Type: text/html\r\n'
    if location:
        headers += 'Location: ' + location + '\r\n'
    return {'Value': {'Kind': 'NativeResponse', 'RequestId': 1, 'RequestUri': uri, 'ResponseUri': uri,
            'StatusCode': status, 'ContentType': 'text/html; charset=utf-8', 'CertificateValidated': True,
            'NativeTransport': True, 'RequestHeaders': 'GET ' + target + ' HTTP/1.1\r\nHost: ' + url.hostname + '\r\n\r\n',
            'ResponseHeaders': headers + '\r\n', 'BodyBytes': 1000, 'BodySha256': 'a' * 64,
            'HtmlTitle': 'Bing', 'HasHtml': True, 'HasHead': True, 'Location': location}}


def remote():
    return {'Address': HOME, 'Title': '必应', 'BodySha256': 'a' * 64, 'CertificateValidated': True,
            'SnapshotSequence': 12, 'ScrollY': 0, 'ViewportWidth': 200, 'ViewportHeight': 100,
            'SearchInputs': [{'Name': 'q', 'Bounds': {'X': 30, 'Y': 20, 'Width': 120, 'Height': 20}}],
            'TextRuns': [{'Text': text, 'Bounds': {'X': 10, 'Y': y, 'Width': 60, 'Height': 12}, 'Color': '#e0e0e0'} for text, y in (('Bing', 4), ('Search', 60))],
            'LoadedStyles': [{'Address': 'https://www.bing.com/real.css', 'Characters': 1000}],
            'DeclaredStyleLinks': ['https://www.bing.com/real.css'], 'InlineStyleCount': 2}


class IndependentBingEvidenceTests(unittest.TestCase):
    def test_native_validated_200_html_chain_is_required(self):
        self.assertEqual(1, len(validate_response_chain([response()])))
        for field, value in (('NativeTransport', False), ('CertificateValidated', False), ('StatusCode', 500),
                             ('HasHtml', False), ('HasHead', False), ('HtmlTitle', 'Navigation failed'),
                             ('BodySha256', 'missing'), ('BodyBytes', 0), ('ResponseHeaders', 'HTTP/1.1 404 Missing\r\n')):
            item = response(); item['Value'][field] = value
            with self.subTest(field=field), self.assertRaises(AssertionError):
                validate_response_chain([item])

    def test_redirects_require_matching_native_raw_headers_and_tls(self):
        next_url = 'https://cn.bing.com/?ensearch=1'
        chain = [response(status=302, location=next_url), response(next_url)]
        self.assertEqual(2, len(validate_response_chain(chain)))
        for change in ('downgrade', 'missing', 'loop', 'host', 'tls'):
            invalid = copy.deepcopy(chain)
            if change == 'downgrade':
                invalid[0]['Value']['Location'] = 'http://cn.bing.com/'
            elif change == 'missing':
                invalid = invalid[:1]
            elif change == 'loop':
                invalid[0] = response(status=302, location=HOME); invalid.append(response(status=302, location=HOME))
            elif change == 'host':
                invalid[1]['Value']['RequestHeaders'] = 'GET /?ensearch=1 HTTP/1.1\r\nHost: attacker.test\r\n\r\n'
            else:
                invalid[1]['Value']['CertificateValidated'] = False
            with self.subTest(change=change), self.assertRaises(AssertionError):
                validate_response_chain(invalid)

    def test_failure_page_wrong_html_and_missing_styles_are_not_remote_bing(self):
        result = validate_remote_page(remote(), response()['Value'])
        self.assertFalse(result['CssParityVerified'])
        for change in ('failure', 'body', 'search', 'style', 'nonfinite', 'tls', 'snapshot'):
            page = remote()
            if change == 'failure': page['Title'] = 'Bing Navigation failed'
            elif change == 'body': page['BodySha256'] = 'b' * 64
            elif change == 'search': page['SearchInputs'][0]['Name'] = 'fake'
            elif change == 'style': page['LoadedStyles'] = []
            elif change == 'nonfinite': page['SearchInputs'][0]['Bounds']['X'] = float('nan')
            elif change == 'tls': page['CertificateValidated'] = False
            else: page['SnapshotSequence'] = 0
            with self.subTest(change=change), self.assertRaises(AssertionError):
                validate_remote_page(page, response()['Value'])

    def test_non_bing_or_missing_packets_do_not_qualify(self):
        for uri in ('https://bing.com.attacker.test/', 'https://example.com/', 'about:Start', 'http://www.bing.com/'):
            with self.subTest(uri=uri), self.assertRaises(AssertionError):
                validate_response_chain([response(uri)] if '://' in uri else [])

    def test_stylesheet_redirects_bind_the_declared_request_and_actual_https_final(self):
        page = remote()
        style = page['LoadedStyles'][0]
        requested = style['Address']; final = 'https://r.bing.com/styles/final.css?v=2'
        style.update(RequestedAddress=requested, FinalAddress=final, Address=final)
        evidence = validate_remote_page(page, response()['Value'])
        self.assertEqual(final, evidence['LoadedStyles'][0]['FinalAddress'])
        self.assertEqual(requested, evidence['LoadedStyles'][0]['RequestedAddress'])
        self.assertFalse(evidence['CssParityVerified'])
        for name in ('unrelated-request', 'wrong-final', 'http-final', 'http-request', 'empty-resource'):
            changed = copy.deepcopy(page); receipt = changed['LoadedStyles'][0]
            if name == 'unrelated-request': receipt['RequestedAddress'] = 'https://www.bing.com/unrelated.css'
            elif name == 'wrong-final': receipt['FinalAddress'] = 'https://r.bing.com/different.css'
            elif name == 'http-final': receipt['Address'] = receipt['FinalAddress'] = 'http://r.bing.com/styles.css'
            elif name == 'http-request': receipt['RequestedAddress'] = 'http://www.bing.com/real.css'
            else: receipt['Characters'] = 0
            with self.subTest(case=name), self.assertRaises(AssertionError):
                validate_remote_page(changed, response()['Value'])

    def test_native_navigation_error_must_be_fresh_complete_current_window_and_bing(self):
        failure = 'Navigation failed: TypeInitializerException: exact actual guest error'
        def state(window=4, address=HOME, content=failure):
            return f'XFE_BROWSER_STATE_BEGIN\n{window}|{address}\n{content}\nXFE_BROWSER_STATE_END\n'
        actual = state()
        self.assertEqual(failure, native_navigation_failure(actual, 0, 4)['Text'])
        for serial, after in ((actual, len(actual)), (state(window=5), 0),
                              (state(address='https://example.com/'), 0), (state(address='about:Start'), 0),
                              (state(content='Actual Bing home'), 0), (actual[:-10], 0),
                              (actual + state(content='Actual Bing home'), 0)):
            with self.subTest(serial=serial, after=after):
                self.assertIsNone(native_navigation_failure(serial, after, 4))
        prefixed = actual + 'NEW_INPUT\n' + actual
        self.assertGreater(native_navigation_failure(prefixed, len(actual), 4)['SerialOffset'], len(actual))

    def test_real_current_navigation_failure_exits_promptly_and_preserves_false_receipt(self):
        serial = ('XFE_DESKTOP_ACTIVE|Browser\nXFE_WINDOW_STATE|4|Browser|216|132|800|500|0|0|1\nXFE_DESKTOP_STATE_END\n'
                  'XFE_BROWSER_STATE_BEGIN\n4|' + HOME + '\nNavigation failed: exact guest TLS init failure\nXFE_BROWSER_STATE_END\n')
        with temporary_directory() as temporary:
            started = time.monotonic()
            with self.assertRaisesRegex(AssertionError, 'exact guest TLS init failure'):
                run_bing(temporary, lambda: serial, lambda: serial, lambda *_: None,
                         lambda *_: self.fail('A failed navigation must not capture an acceptance screenshot'),
                         lambda *_: None, lambda *_: None, lambda *_: None,
                         timeout=180, open_browser=False, start_offset=0)
            self.assertLess(time.monotonic() - started, 1)
            result = json.loads((Path(temporary) / 'BingEvidence.json').read_text(encoding='utf-8'))
            self.assertFalse(result['Passed']); self.assertFalse(result['NativeHttpsBingLoaded'])
            self.assertEqual('Navigation failed: exact guest TLS init failure', result['NavigationFailure']['Text'])
            self.assertEqual(1, result['RawGuestEvidence']['NavigationFailures'])
            self.assertIn('exact guest TLS init failure', (Path(temporary) / 'BingRequestEvidence.log').read_text(encoding='utf-8'))

    def test_parser_requires_complete_json_packet_and_retains_raw_evidence(self):
        payload = json.dumps(response()['Value'])
        packet = RESPONSE + '_BEGIN\n' + payload + '\n' + RESPONSE + '_END'
        self.assertEqual(packet, json_blocks(packet, RESPONSE)[0]['Raw'])
        self.assertEqual([], json_blocks(packet[:-4], RESPONSE))
        self.assertEqual([], json_blocks(RESPONSE + '_BEGIN\n{} broken\n' + RESPONSE + '_END', RESPONSE))
        self.assertEqual([], json_blocks(packet, RESPONSE, len(packet)))

    def test_blank_uniform_wrong_color_and_wrong_geometry_do_not_count_as_glyphs(self):
        page = remote(); viewport = [20, 30, 200, 100]
        blank = Image.new('RGB', (240, 160), '#101010')
        self.assertFalse(remote_text_pixels(blank, page, viewport)['Passed'])
        self.assertFalse(remote_text_pixels(Image.new('RGB', blank.size, '#e0e0e0'), page, viewport)['Passed'])
        # This generated image tests only the verifier's positive/negative
        # controls; run_bing never uses it or calls a host image generator.
        control = blank.copy(); draw = ImageDraw.Draw(control)
        for run in page['TextRuns']:
            x, y = run['Bounds']['X'] + viewport[0], run['Bounds']['Y'] + viewport[1]
            for offset in range(0, 30, 5): draw.rectangle((x + offset, y + 2, x + offset + 1, y + 8), fill='#e0e0e0')
        self.assertTrue(remote_text_pixels(control, page, viewport)['Passed'])
        changed = copy.deepcopy(page)
        for run in changed['TextRuns']: run['Color'] = '#ff0000'
        self.assertFalse(remote_text_pixels(control, changed, viewport)['Passed'])
        self.assertFalse(remote_text_pixels(control, page, [viewport[0] + 100, viewport[1], *viewport[2:]])['Passed'])

    def test_real_run_without_https_always_keeps_false_timeout_and_raw_packets(self):
        with temporary_directory() as temporary:
            def no_op(*_): pass
            with self.assertRaisesRegex(AssertionError, 'timed out'):
                run_bing(temporary, lambda: '', lambda: '', no_op, no_op, no_op, no_op, no_op, timeout=0)
            result = json.loads((Path(temporary) / 'BingEvidence.json').read_text(encoding='utf-8'))
            self.assertFalse(result['Passed']); self.assertFalse(result['NativeHttpsBingLoaded'])
            self.assertFalse(result['HostBrowserScreenshots']); self.assertFalse(result['AllUserRequirementsMet'])
            self.assertEqual(0, result['RawGuestEvidence']['Packets'])
            self.assertTrue((Path(temporary) / 'BingRequestEvidence.log').exists())

    def test_native_qemu_profile_has_exact_e1000_nat_cpu_and_utc_arguments(self):
        base = ['qemu-system-x86_64', '-cpu', 'max', '-rtc', 'base=utc']
        arguments = base + network_arguments(True)
        profile = validate_network_arguments(arguments, True)
        self.assertEqual(native_network_profile(True), profile)
        self.assertEqual('8086:100e', profile['PciVendorDevice'])
        self.assertEqual('Static', profile['AddressConfiguration'])
        self.assertFalse(profile['HostProxy']); self.assertFalse(profile['HostForwarding'])
        self.assertEqual(['-netdev', NAT_ARGUMENT, '-device', NIC_ARGUMENT], network_arguments(True))
        self.assertFalse(validate_network_arguments(base + network_arguments(False), False)['Enabled'])

    def test_wrong_nic_proxy_forwarding_offline_duplicate_cpu_and_rtc_are_rejected(self):
        valid = ['-cpu', 'max', '-rtc', 'base=utc', *network_arguments(True)]
        invalid = [valid + ['-net', 'none'], valid + ['-nic', 'user'],
                   valid + ['-device', 'usb-net,netdev=NativeNet'],
                   valid + ['-netdev', NAT_ARGUMENT], valid + ['-cpu', 'max'],
                   [item.replace('e1000,', 'e1000e,') for item in valid],
                   [item.replace('dns=10.0.2.3', 'dns=127.0.0.1') for item in valid],
                   [item.replace('base=utc', 'base=localtime') for item in valid],
                   [item + ',hostfwd=tcp::8080-:80' if item == NAT_ARGUMENT else item for item in valid],
                   ['-cpu', 'max', '-rtc', 'base=utc', '-net', 'none']]
        for arguments in invalid:
            with self.subTest(arguments=arguments), self.assertRaises(AssertionError):
                validate_network_arguments(arguments, True)

    def test_presentation_requires_latest_exact_snapshot_url_scroll_and_real_scanout(self):
        def marker(snapshot=12, address=HOME, sequence=6, ticks=100, scroll=0, window=3):
            return f'XFE_BROWSER_BACKGROUND_PRESENTED|Window|{window}|Generation|2|Address|{address}|SnapshotSequence|{snapshot}|PresentationSequence|{sequence}|PresentationTicks|{ticks}|ScrollY|{scroll}\n'
        valid = marker()
        value = presentation_identity(matching_presentation(valid, remote(), 3))
        self.assertEqual(12, value['SnapshotSequence']); self.assertEqual(100, value['PresentationTicks'])
        for serial in ('', marker(snapshot=13), marker(address='https://example.com/'), marker(sequence=0),
                       marker(ticks=0), marker(scroll=5), marker(window=4), valid + marker(snapshot=13)):
            with self.subTest(serial=serial), self.assertRaises(AssertionError):
                matching_presentation(serial, remote(), 3)
        with self.assertRaises(AssertionError): matching_presentation(valid, remote(), 3, len(valid))

    def test_screenshot_requires_unchanged_snapshot_and_monotonic_presentation_counters(self):
        initial = {'WindowId': 3, 'Generation': 2, 'Address': HOME, 'SnapshotSequence': 12,
                   'PresentationSequence': 6, 'PresentationTicks': 100, 'ScrollY': 0}
        later = {**initial, 'PresentationSequence': 7, 'PresentationTicks': 110}
        self.assertTrue(stable_screenshot_presentation(initial, later)['StableSnapshot'])
        for field, value in (('WindowId', 4), ('Generation', 3), ('Address', 'about:Start'),
                             ('SnapshotSequence', 13), ('PresentationSequence', 5), ('PresentationTicks', 99), ('ScrollY', 5)):
            with self.subTest(field=field), self.assertRaises(AssertionError):
                stable_screenshot_presentation(initial, {**later, field: value})

    def test_already_open_browser_requires_actual_navigation_start_offset(self):
        with temporary_directory() as temporary, self.assertRaisesRegex(AssertionError, 'start offset'):
            run_bing(temporary, lambda: '', lambda: '', lambda *_: None,
                     lambda *_: None, lambda *_: None, lambda *_: None, lambda *_: None,
                     open_browser=False)

    def test_powershell_launcher_checks_actual_receipt_hashes_and_snapshot_bindings(self):
        """Execute only Open.ps1's real read-only Bing verifier against small guard fixtures.

        These fixtures are neither guest evidence nor boot proofs. No launcher,
        image copy, host network request, display or VM is invoked by this test.
        """
        source = (ROOT / 'Tools/IndependentKernel/Open.ps1').read_text(encoding='utf-8-sig')
        helpers = source[source.index('function Get-IndependentHash'):source.index('$sourceImage=')]
        guard = source[source.index('$bingBinding='):source.index('foreach($capture in $documentWorkflow.Screenshots.')]
        with temporary_directory() as temporary:
            root = Path(temporary)
            cases = []
            for name in ('valid', 'tampered-png', 'tampered-log', 'wrong-body', 'wrong-snapshot',
                         'unstable', 'missing-generation', 'escaped-png', 'no-glyphs', 'unchecked-tls', 'parity-overclaim'):
                directory = root / name; directory.mkdir()
                image = directory / 'Browser-Bing-Native-1.png'
                Image.new('RGB', (2, 2), '#101010').save(image)
                raw = directory / 'BingRequestEvidence.log'; raw.write_text('Verifier unit-test fixture only', encoding='utf-8')
                sha = lambda path: hashlib.sha256(path.read_bytes()).hexdigest()
                presentation = {'WindowId': 3, 'Generation': 2, 'Address': HOME, 'SnapshotSequence': 12,
                                'PresentationSequence': 6, 'PresentationTicks': 100, 'ScrollY': 0}
                capture = {'Path': image.name, 'Sha256': sha(image), 'BodySha256': 'a' * 64,
                           'PresentationCorrelation': {'Before': dict(presentation), 'After': dict(presentation), 'StableSnapshot': True},
                           'TextPixels': {'Passed': True}}
                record = {'Kind': 'XfeIndependentNativeBingEvidence', 'Passed': True, 'NativeHttpsBingLoaded': True,
                          'AllUserRequirementsMet': False, 'FullCssParityVerified': False, 'CompleteJavaScriptCompatibility': False,
                          'HostHttpRequests': False, 'HostBrowserScreenshots': False, 'HostJavaScriptInjection': False,
                          'VmStartedByThisTool': False, 'RequestedUrl': HOME, 'Responses': [response()['Value']],
                          'RemotePage': remote(), 'Presentation': presentation, 'Screenshots': [capture],
                          'RawGuestEvidence': {'Path': raw.name, 'Sha256': sha(raw), 'Packets': 2, 'PresentationMarkers': 1}}
                if name == 'tampered-png': image.write_bytes(b'changed PNG')
                elif name == 'tampered-log': raw.write_text('changed log', encoding='utf-8')
                elif name == 'wrong-body': capture['BodySha256'] = 'b' * 64
                elif name == 'wrong-snapshot': capture['PresentationCorrelation']['Before']['SnapshotSequence'] = 13
                elif name == 'unstable': capture['PresentationCorrelation']['StableSnapshot'] = False
                elif name == 'missing-generation':
                    del capture['PresentationCorrelation']['Before']['Generation']; del capture['PresentationCorrelation']['After']['Generation']
                elif name == 'escaped-png': capture['Path'] = '../outside.png'
                elif name == 'no-glyphs': capture['TextPixels']['Passed'] = False
                elif name == 'unchecked-tls': record['Responses'][0]['CertificateValidated'] = False
                elif name == 'parity-overclaim': record['AllUserRequirementsMet'] = True
                receipt = directory / 'BingEvidence.json'; receipt.write_text(json.dumps(record), encoding='utf-8')
                binding = {'Path': receipt.name, 'Sha256': sha(receipt), 'Passed': True, 'NativeHttpsBingLoaded': True}
                cases.append({'Name': name, 'Root': str(directory), 'Binding': binding, 'Expected': name == 'valid'})
            fixtures = root / 'Cases.json'; fixtures.write_text(json.dumps(cases), encoding='utf-8')
            script = root / 'Check.ps1'
            script.write_text("param([string]$Cases)\n$ErrorActionPreference='Stop'\n"
                "Import-Module (Join-Path $PSHOME 'Modules/Microsoft.PowerShell.Utility/Microsoft.PowerShell.Utility.psd1')\n" + helpers +
                "\n$check={\n" + guard + "\n}\n$results=@(foreach($case in (Get-Content -Raw -Encoding UTF8 $Cases | ConvertFrom-Json)){\n"
                "$desktopInput=[pscustomobject]@{BingEvidence=$case.Binding};$guestEvidenceRoot=$case.Root;$passed=$false;$reason=$null\n"
                "try{& $check;$passed=$true}catch{$reason=$_.Exception.Message}\n"
                "[pscustomobject]@{Name=$case.Name;Passed=$passed;Expected=$case.Expected;Reason=$reason}\n"
                "});ConvertTo-Json -Depth 5 -InputObject $results\n", encoding='utf-8-sig')
            completed = subprocess.run(['powershell.exe', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-File', str(script), '-Cases', str(fixtures)],
                                       cwd=ROOT, check=True, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=30)
            results = json.loads(completed.stdout)
            self.assertEqual(len(cases), len(results))
            for result in results:
                with self.subTest(case=result['Name']):
                    self.assertEqual(result['Expected'], result['Passed'], result.get('Reason'))


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