XFESystem

"""Source-bound guest Bing acceptance. This module performs no host HTTP requests."""
import hashlib
import json
import math
from pathlib import Path
import re
import time
from urllib.parse import urljoin, urlsplit

from PIL import Image

RESPONSE = 'XFE_NATIVE_WEB_RESPONSE'
REMOTE = 'XFE_REMOTE_PAGE_PROOF'
HOME = 'https://www.bing.com/'
NAT_ARGUMENT = 'user,id=NativeNet,net=10.0.2.0/24,host=10.0.2.2,dns=10.0.2.3,dhcpstart=10.0.2.15'
NIC_ARGUMENT = 'e1000,netdev=NativeNet'
TITLE = re.compile(r'bing|必应', re.I)
PRESENTED = re.compile(r'^XFE_BROWSER_BACKGROUND_PRESENTED\|Window\|(\d+)\|Generation\|(\d+)\|Address\|([^|\r\n]+)'
                       r'\|SnapshotSequence\|(\d+)\|PresentationSequence\|(\d+)\|PresentationTicks\|(\d+)\|ScrollY\|([\d.]+)\r?$', re.M)
WINDOW = re.compile(r'^XFE_WINDOW_STATE\|(\d+)\|(\w+)\|(-?\d+)\|(-?\d+)\|(\d+)\|(\d+)\|([01])\|([01])\|([01])\r?$', re.M)
BROWSER_STATE = re.compile(r'XFE_BROWSER_STATE_BEGIN\r?\n(\d+)\|([^\r\n]+)\r?\n(.*?)\r?\nXFE_BROWSER_STATE_END', re.S)


def native_network_profile(enabled):
    """Describe the actual QEMU profile, without claiming guest network success."""
    return {'Kind': 'XfeIndependentNativeNetworkProfile', 'Enabled': bool(enabled),
            'Backend': 'QEMU user NAT' if enabled else 'None', 'Device': 'e1000' if enabled else None,
            'PciVendorDevice': '8086:100e' if enabled else None,
            'GuestAddress': '10.0.2.15' if enabled else None, 'Gateway': '10.0.2.2' if enabled else None,
            'DnsServer': '10.0.2.3' if enabled else None, 'AddressConfiguration': 'Static' if enabled else None,
            'NetdevArgument': NAT_ARGUMENT if enabled else None, 'DeviceArgument': NIC_ARGUMENT if enabled else None,
            'HostProxy': False, 'HostForwarding': False, 'Rtc': 'base=utc', 'Cpu': 'max',
            'EntropyPolicy': 'Guest CPUID plus bounded RDSEED/RDRAND with carry check; unavailable fails closed'}


def network_arguments(enabled):
    return ['-netdev', NAT_ARGUMENT, '-device', NIC_ARGUMENT] if enabled else ['-net', 'none']


def validate_network_arguments(arguments, enabled):
    expected = network_arguments(enabled)
    options = ('-net', '-netdev', '-nic', '-device')
    actual = []
    for index, argument in enumerate(arguments):
        if argument not in options:
            continue
        assert index + 1 < len(arguments), 'Missing QEMU network argument value'
        value = arguments[index + 1]
        actual.extend((argument, value))
    assert actual == expected, 'QEMU network arguments differ from the source-bound profile'
    for key, value in (('-cpu', 'max'), ('-rtc', 'base=utc')):
        indices = [index for index, item in enumerate(arguments) if item == key]
        assert len(indices) == 1 and indices[0] + 1 < len(arguments) and arguments[indices[0] + 1] == value, 'Missing or ambiguous CPU/UTC profile'
    return native_network_profile(enabled)


def window_geometry(log, application):
    end = log.rfind('XFE_DESKTOP_STATE_END')
    begin = log.rfind('XFE_DESKTOP_ACTIVE|', 0, end)
    assert begin >= 0 and end > begin, 'No complete current native desktop state'
    matches = [item for item in WINDOW.finditer(log[begin:end]) if item[2] == application and item[7] == '0']
    assert matches, 'No visible actual ' + application + ' window geometry'
    match = matches[-1]
    return int(match[1]), [int(match[index]) for index in (3, 4, 5, 6)]


def json_blocks(log, marker, after=0):
    expression = re.compile(re.escape(marker) + r'_BEGIN\r?\n(.*?)\r?\n' + re.escape(marker) + r'_END', re.S)
    result = []
    for match in expression.finditer(log, after):
        try:
            value = json.loads(match[1])
        except (ValueError, TypeError):
            continue  # An incomplete/rejected packet never counts as evidence.
        if isinstance(value, dict):
            result.append({'Value': value, 'SerialOffset': match.start(), 'EndOffset': match.end(), 'Raw': match[0]})
    return result


def _url(value):
    parsed = urlsplit(value)
    assert parsed.scheme == 'https' and parsed.hostname and not parsed.username and not parsed.password, 'A Bing hop must use HTTPS without credentials'
    assert parsed.port in (None, 443), 'Unexpected HTTPS port'
    return parsed


def _same_url(left, right):
    try:
        a, b = _url(left), _url(right)
    except (AssertionError, TypeError, ValueError, AttributeError):
        return False
    return (a.hostname.lower(), a.path or '/', a.query) == (b.hostname.lower(), b.path or '/', b.query)


def validate_response_chain(packets, initial=HOME):
    """Validate actual guest response packets; redirects cannot become implicit success."""
    expected, chain, seen = initial, [], set()
    for packet in packets:
        value = packet['Value']
        if not _same_url(value.get('RequestUri', ''), expected):
            continue
        uri = _url(value['RequestUri'])
        assert value.get('NativeTransport') is True, 'A host transport cannot qualify'
        assert value.get('CertificateValidated') is True, 'Every HTTPS hop requires validated TLS'
        assert _same_url(value['ResponseUri'], value['RequestUri']), 'The hop response address differs from its request'
        request, headers = value.get('RequestHeaders', ''), value.get('ResponseHeaders', '')
        target = (uri.path or '/') + ('?' + uri.query if uri.query else '')
        assert request.splitlines() and request.splitlines()[0] == 'GET ' + target + ' HTTP/1.1', 'Missing actual HTTP request line'
        host = re.search(r'^Host:\s*([^\r\n]+)', request, re.M | re.I)
        assert host and host[1].strip().lower() in (uri.hostname.lower(), uri.hostname.lower() + ':443'), 'HTTP Host does not match TLS URL'
        status = value.get('StatusCode')
        assert type(status) is int and re.match(r'^HTTP/1\.[01]\s+' + str(status) + r'(?:\s|\r?$)', headers), 'Status differs from raw response headers'
        assert value.get('RequestId') is not None, 'Missing guest request identity'
        chain.append(packet)
        if status in (301, 302, 303, 307, 308):
            location = value.get('Location')
            assert isinstance(location, str) and location, 'Redirect has no Location'
            raw_location = re.search(r'^Location:\s*([^\r\n]+)', headers, re.M | re.I)
            assert raw_location and raw_location[1].strip() == location, 'Redirect differs from raw Location'
            expected = urljoin(value['RequestUri'], location)
            _url(expected)
            assert expected not in seen and len(chain) <= 5, 'Redirect cycle or bound exceeded'
            seen.add(expected)
            continue
        assert status == 200, 'Bing did not return HTTP 200'
        assert uri.hostname.lower() == 'bing.com' or uri.hostname.lower().endswith('.bing.com'), 'The final page is outside Bing'
        assert value.get('ContentType', '').split(';')[0].strip().lower() in ('text/html', 'application/xhtml+xml'), 'The response is not HTML'
        assert value.get('HasHtml') is True and value.get('HasHead') is True, 'No actual HTML/head was received'
        assert TITLE.search(value.get('HtmlTitle', '')), 'The real HTML title is not Bing'
        assert type(value.get('BodyBytes')) is int and value['BodyBytes'] >= 128, 'No bounded actual HTML body'
        assert re.fullmatch(r'[0-9a-fA-F]{64}', value.get('BodySha256', '')), 'Missing actual HTML fingerprint'
        return chain
    raise AssertionError('No complete validated native HTTPS Bing response chain')


def _rect(value):
    values = [float(value[name]) for name in ('X', 'Y', 'Width', 'Height')]
    assert all(math.isfinite(number) for number in values) and values[2] > 0 and values[3] > 0, 'Invalid actual element geometry'
    return values


def validate_remote_page(page, response):
    assert _same_url(page['Address'], response['ResponseUri']), 'Remote DOM belongs to another address'
    assert page.get('BodySha256', '').lower() == response['BodySha256'].lower(), 'Rendered HTML does not match the HTTP body'
    assert page.get('CertificateValidated') is True, 'The actual rendered response must retain validated TLS'
    assert type(page.get('SnapshotSequence')) is int and page['SnapshotSequence'] > 0, 'Missing actual immutable paint snapshot identity'
    assert TITLE.search(page.get('Title', '')) and not re.search(r'failed|unavailable|\berror\b|错误|不可用', page.get('Title', ''), re.I), 'A failure page cannot qualify'
    width, height = float(page['ViewportWidth']), float(page['ViewportHeight'])
    assert width > 0 and height > 0 and math.isfinite(width + height)
    searches = [item for item in page.get('SearchInputs', []) if item.get('Name') == 'q']
    rectangles = [_rect(item['Bounds']) for item in searches]
    assert any(x < width and y < height and x + w > 0 and y + h > 0 for x, y, w, h in rectangles), 'No actual Bing search input geometry'
    assert page.get('TextRuns'), 'No actual remote text geometry'
    loaded = {}
    for item in page.get('LoadedStyles', []):
        if item.get('Characters', 0) <= 0:
            continue
        # A real redirect receipt keeps both the DOM-requested stylesheet URI
        # and its validated final URI. Unrelated final resources cannot stand
        # in for the page's declared request. Older unredirected receipts may
        # use Address alone, without inventing an implicit redirect.
        requested = item.get('RequestedAddress', item['Address'])
        final = item.get('FinalAddress', item['Address'])
        _url(requested); _url(final)
        assert _same_url(item['Address'], final), 'The stylesheet final address differs from its actual resource receipt'
        loaded[requested] = item
    for declared in page.get('DeclaredStyleLinks', []):
        _url(declared)
        assert declared in loaded, 'A declared external stylesheet did not load: ' + declared
    return {'SearchInputs': searches, 'LoadedStyles': list(loaded.values()),
            'InlineStyleCount': page.get('InlineStyleCount'), 'DeclaredStyleLinks': page.get('DeclaredStyleLinks', []),
            'CssParityVerified': False, 'Meaning': 'Actual resource receipts and element geometry; full Windows CSS/JS parity is not inferred.'}


def native_navigation_failure(serial, after, window_id):
    """Only a fresh complete state for the current native Bing window qualifies."""
    matches = [item for item in BROWSER_STATE.finditer(serial, after) if int(item[1]) == window_id]
    if not matches:
        return None
    item = matches[-1]
    try:
        address = _url(item[2])
    except (AssertionError, TypeError, ValueError):
        return None
    if address.hostname.lower() != 'bing.com' and not address.hostname.lower().endswith('.bing.com'):
        return None
    content = item[3].strip()
    if not content.startswith('Navigation failed:'):
        return None
    return {'WindowId': window_id, 'Address': item[2], 'Text': content,
            'SerialOffset': item.start(), 'Raw': item[0]}


def matching_presentation(serial, page, window_id, after=0):
    """The latest paint for this window must be this exact immutable DOM snapshot."""
    matches = [item for item in PRESENTED.finditer(serial, after) if int(item[1]) == window_id]
    assert matches, 'No actual native browser presentation'
    item = matches[-1]
    assert _same_url(item[3], page['Address']) and int(item[4]) == page['SnapshotSequence'], 'The latest presented browser snapshot differs from the remote DOM'
    assert int(item[5]) > 0 and int(item[6]) > 0, 'No actual scanout presentation identity/time'
    assert abs(float(item[7]) - float(page.get('ScrollY', 0))) <= .001, 'Rendered remote scroll position differs'
    return item


def presentation_identity(item):
    return {'WindowId': int(item[1]), 'Generation': int(item[2]), 'Address': item[3],
            'SnapshotSequence': int(item[4]), 'PresentationSequence': int(item[5]),
            'PresentationTicks': int(item[6]), 'ScrollY': float(item[7])}


def stable_screenshot_presentation(before, after):
    for name in ('WindowId', 'Generation', 'Address', 'SnapshotSequence', 'ScrollY'):
        assert before[name] == after[name], 'Browser presentation changed during screenshot: ' + name
    assert after['PresentationSequence'] >= before['PresentationSequence'] and after['PresentationTicks'] >= before['PresentationTicks'], 'Browser presentation counters regressed during screenshot'
    return {'Before': before, 'After': after, 'StableSnapshot': True,
            'Meaning': 'Actual QMP framebuffer capture bracketed by the same immutable native browser snapshot; this is not physical display timing.'}


def _color(value):
    value = value.strip().lower()
    if re.fullmatch(r'#?[0-9a-f]{6}', value):
        value = value.lstrip('#')
        return tuple(int(value[index:index+2], 16) for index in (0, 2, 4))
    match = re.fullmatch(r'rgba?\(\s*(\d+(?:\.\d+)?)\s*[, ]\s*(\d+(?:\.\d+)?)\s*[, ]\s*(\d+(?:\.\d+)?)(?:\s*[,/]\s*[\d.]+)?\s*\)', value)
    assert match, 'Unsupported diagnostic text color: ' + value
    return tuple(round(float(number)) for number in match.groups())


def remote_text_pixels(frame, page, viewport, scale=1):
    """Require nonuniform ink at visible actual text geometry in the guest PNG."""
    image = frame.convert('RGB'); samples = []
    for run in page.get('TextRuns', [])[:128]:
        if len(run.get('Text', '').strip()) < 2:
            continue
        x, y, width, height = _rect(run['Bounds'])
        left, top = max(0, x), max(0, y)
        right, bottom = min(viewport[2], x + width), min(viewport[3], y + height)
        if right <= left or bottom <= top:
            continue
        bounds = [round((viewport[0] + left) * scale), round((viewport[1] + top) * scale),
                  round((viewport[0] + right) * scale), round((viewport[1] + bottom) * scale)]
        region = image.crop(bounds); color = _color(run['Color'])
        ink = [(px, py) for py in range(region.height) for px in range(region.width)
               if all(abs(region.getpixel((px, py))[channel] - color[channel]) <= 14 for channel in range(3))]
        area = region.width * region.height
        columns = len({point[0] for point in ink}); rows = len({point[1] for point in ink})
        passed = len(ink) >= 12 and columns >= 6 and rows >= 3 and len(ink) < area * .75
        samples.append({'Text': run['Text'][:96], 'NativeBounds': bounds, 'InkPixels': len(ink), 'InkColumns': columns, 'InkRows': rows, 'Passed': passed})
    return {'Passed': sum(item['Passed'] for item in samples) >= 2, 'Samples': samples,
            'Meaning': 'Nonuniform text ink inside real DOM rectangles in guest framebuffer PNG; no host browser image or synthetic acceptance pixels.'}


def run_bing(proof_directory, log, healthy, wait, screenshot, key, type_text, move, *, timeout=180, open_browser=True, start_offset=None):
    """Open the product's default Browser through QMP. Caller owns QEMU and cleanup."""
    proof_directory = Path(proof_directory)
    started = time.monotonic(); initial_log = healthy()
    assert open_browser or type(start_offset) is int, 'An already-open browser requires the actual input start offset'
    after = len(initial_log) if start_offset is None else start_offset
    assert 0 <= after <= len(initial_log), 'Invalid actual navigation evidence offset'
    result = {'Kind': 'XfeIndependentNativeBingEvidence', 'Passed': False, 'AllUserRequirementsMet': False,
              'NativeHttpsBingLoaded': False, 'FullCssParityVerified': False, 'CompleteJavaScriptCompatibility': False,
              'HostHttpRequests': False, 'HostBrowserScreenshots': False, 'HostJavaScriptInjection': False,
              'VmStartedByThisTool': False, 'Input': 'Actual QMP keyboard/mouse only', 'RequestedUrl': HOME,
              'TimeoutSeconds': timeout, 'Screenshots': []}
    output = proof_directory / 'BingEvidence.json'

    def persist():
        result['ElapsedHostSeconds'] = time.monotonic() - started
        output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')

    persist()
    try:
        # The desktop profile starts with its menu closed. Preserve the product
        # default navigation: do not replace it with an offline/failure fixture.
        if open_browser:
            key('ctrl-esc'); key('down'); key('down'); key('ret')
        wait('XFE_DESKTOP_ACTIVE|Browser', after)
        end = time.monotonic() + timeout
        last_error = 'No native response packet'
        accepted = None
        while time.monotonic() < end:
            serial = healthy()
            try:
                window_id, _ = window_geometry(serial, 'Browser')
            except AssertionError:
                window_id = None  # The complete desktop-state packet may still be arriving.
            failure = native_navigation_failure(serial, after, window_id) if window_id is not None else None
            if failure is not None:
                result['NavigationFailure'] = {key: value for key, value in failure.items() if key != 'Raw'}
                raise AssertionError('Actual native Bing navigation failed: ' + failure['Text'])
            try:
                packets = json_blocks(serial, RESPONSE, after)
                chain = validate_response_chain(packets)
                final = chain[-1]['Value']
                remote_packets = json_blocks(serial, REMOTE, after)
                for remote in reversed(remote_packets):
                    page = remote['Value']
                    css = validate_remote_page(page, final)
                    window_id, bounds = window_geometry(serial, 'Browser')
                    presented = matching_presentation(serial, page, window_id, after)
                    accepted = (chain, remote, css, presented, bounds)
                    break
                assert accepted, 'No matching rendered remote DOM readback'
                break
            except (AssertionError, KeyError, TypeError, ValueError) as error:
                last_error = str(error)
            time.sleep(.1)
        assert accepted, 'Actual native Bing acceptance timed out: ' + last_error
        chain, remote, css, presented, bounds = accepted
        result['Responses'] = [packet['Value'] for packet in chain]
        result['RemotePage'] = remote['Value']; result['CssResources'] = css
        result['Presentation'] = presentation_identity(presented)
        viewport = [bounds[0] + 2, bounds[1] + 67, bounds[2] - 4, bounds[3] - 96]
        assert abs(viewport[2] - float(remote['Value']['ViewportWidth'])) <= 1 and abs(viewport[3] - float(remote['Value']['ViewportHeight'])) <= 1
        move(1200, 710)
        for attempt in range(1, 5):
            serial = healthy(); window_id, current_bounds = window_geometry(serial, 'Browser')
            assert current_bounds == bounds, 'The browser geometry changed before its actual screenshot'
            before = presentation_identity(matching_presentation(serial, remote['Value'], window_id, after))
            name = 'Browser-Bing-Native-' + str(attempt)
            screenshot(name)
            serial = healthy(); current_id, current_bounds = window_geometry(serial, 'Browser')
            assert current_id == window_id and current_bounds == bounds, 'The browser geometry changed during its actual screenshot'
            presentation = stable_screenshot_presentation(before,
                presentation_identity(matching_presentation(serial, remote['Value'], window_id, after)))
            path = proof_directory / (name + '.png')
            with Image.open(path) as frame:
                assert frame.width >= viewport[0] + viewport[2] and frame.height >= viewport[1] + viewport[3]
                ink = remote_text_pixels(frame, remote['Value'], viewport)
            result['Screenshots'].append({'Path': path.name, 'Sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
                'BodySha256': remote['Value']['BodySha256'], 'PresentationCorrelation': presentation, 'TextPixels': ink})
            if ink['Passed']:
                break
            time.sleep(.25)
        assert result['Screenshots'][-1]['TextPixels']['Passed'], 'The actual Bing text was not painted at its remote DOM geometry'
        result.update(Passed=True, NativeHttpsBingLoaded=True,
                      Meaning='Validated guest HTTPS 200, real Bing HTML/DOM, actual native presentation and framebuffer screenshot; full CSS/JS parity remains unverified.')
    except Exception as error:
        result['Failure'] = str(error)
        raise
    finally:
        # Raw packets survive timeout and partial failure, without dumping the
        # entire HTML body or making another network request on the host.
        try:
            serial = log()
            packets = json_blocks(serial, RESPONSE, after) + json_blocks(serial, REMOTE, after)
            raw = proof_directory / 'BingRequestEvidence.log'
            presented_packets = [{'Raw': match[0], 'SerialOffset': match.start()} for match in PRESENTED.finditer(serial, after)]
            failure_packets = [{'Raw': match[0], 'SerialOffset': match.start()} for match in BROWSER_STATE.finditer(serial, after)
                               if match[3].strip().startswith('Navigation failed:')]
            raw.write_text('\n'.join(packet['Raw'] for packet in sorted(packets + presented_packets + failure_packets, key=lambda item: item['SerialOffset'])), encoding='utf-8')
            result['RawGuestEvidence'] = {'Path': raw.name, 'Sha256': hashlib.sha256(raw.read_bytes()).hexdigest(), 'Packets': len(packets)}
            result['RawGuestEvidence']['PresentationMarkers'] = len(presented_packets)
            result['RawGuestEvidence']['NavigationFailures'] = len(failure_packets)
        except Exception as error:
            result['RawGuestEvidenceFailure'] = str(error)
            result['Passed'] = False
        persist()
    return result