XFESystem

"""Pure offline resource-ownership and release negative controls; no native pass claimed."""
import copy
import unittest

from NativeDependentSlotEvidence import ClosedCycleOwnershipTracker, checked_address, parse_symbols, symbol_address
from TestEditorStability import (closed_cycle_release_observation, closed_cycle_physical_budget,
                                require_no_closed_cycle_handle_growth)


TLS_MT = 0xabc000


def owner(generic, index, base, target):
    return {'ClosedT': generic, 'ModuleIndex': 0, 'TypeIndex': index, 'TypeBase': base,
            'Target': target, 'RootKind': 'SchedulerThread', 'ThreadAddress': 0x1000,
            'ThreadId': 5, 'RegistryIndex': 3}


def add_tls(value, generic='INode', index=1, address=40, target=0x40000, base=0xb000):
    runtime = value['RuntimeTlsPools']; root = runtime['ThreadRootSnapshots'][0]
    row = root['TypeStorageRows'][0]
    while len(row['DataWords']) <= index: row['DataWords'].append(0)
    row['Length'] = len(row['DataWords']); row['DataWords'][index] = base
    field = owner(generic, index, base, target); root['PoolFields'].append(copy.deepcopy(field))
    slot = {'SlotAddress': address, 'Target': target, 'Dependent': 0, 'ClearedTombstone': False,
            'TargetObject': {'EEType': TLS_MT}}
    value['Slots'].append(slot)
    runtime['TlsArrayBodies'].append({'SlotAddress': address, 'TlsArray': target, 'Body': {'Length': 27},
        'ThreadOwners': [field], 'PoolRegistryOwners': [{'ClosedT': generic, 'Index': 1, 'HashCode': 7,
            'DependentHandle': address, 'SlotTarget': target, 'SlotDependent': 0}]})
    value['AllocatedTypedSlots'] = value['AliveCounterSum'] = len(value['Slots'])
    return value


def baseline():
    root = {'RootKind': 'SchedulerThread', 'ThreadAddress': 0x1000, 'ThreadId': 5, 'RegistryIndex': 3,
            'TlsRoot': 0x8000, 'ModuleStorage': {'Address': 0x8000, 'Length': 1, 'DataWords': [0x9000]},
            'TypeStorageRows': {0: {'Address': 0x9000, 'Length': 2, 'DataWords': [0, 0]}}, 'PoolFields': []}
    value = {'SupportedTlsArrayEEType': TLS_MT, 'AllocatedTypedSlots': 2, 'AliveCounterSum': 2,
        'Slots': [{'SlotAddress': 10, 'Target': 0x111, 'Dependent': 0x222, 'ClearedTombstone': False, 'TargetObject': {'EEType': 1}},
                  {'SlotAddress': 20, 'Target': 0, 'Dependent': 0, 'ClearedTombstone': True, 'TargetObject': {'Null': True}}],
        'RuntimeTlsPools': {'DescriptorCount': 2,
            'Descriptors': [{'ClosedT': 'Char', 'ModuleIndex': 0, 'TypeIndex': 0},
                            {'ClosedT': 'INode', 'ModuleIndex': 0, 'TypeIndex': 1}],
            'ThreadRootSnapshots': [root, {'RootKind': 'BootstrapFallback', 'ThreadId': None, 'TlsRoot': 0, 'PoolFields': []}],
            'TlsArrayBodies': []}}
    return add_tls(value, 'Char', 0, 30, 0x30000, 0xa000)


class NativeOwnershipGuards(unittest.TestCase):
    def qualify_change(self, changed):
        tracker = ClosedCycleOwnershipTracker(); tracker.qualify(baseline())
        return tracker.qualify(changed)

    def test_same_thread_same_t_first_nil_to_root_qualifies_but_raw_counter_stays_false(self):
        initial = baseline(); changed = add_tls(copy.deepcopy(initial))
        tracker = ClosedCycleOwnershipTracker(); tracker.qualify(initial)
        result = tracker.qualify(changed)
        self.assertTrue(result['Passed'])
        self.assertEqual(1, len(result['ProvenFirstUseRuntimeTlsInitializations']))
        self.assertEqual(2, result['FixedNonTlsSlots'])
        self.assertFalse(result['WarmBaselineRolled'])
        with self.assertRaises(AssertionError): require_no_closed_cycle_handle_growth(4, 3)

    def test_fake_tls_type_missing_root_or_wrong_cwt_never_qualifies(self):
        for corruption in ('type', 'root', 't', 'handle', 'target', 'free_entry'):
            value = add_tls(baseline()); body = value['RuntimeTlsPools']['TlsArrayBodies'][-1]
            if corruption == 'type': value['Slots'][-1]['TargetObject']['EEType'] = 0xBAD
            if corruption == 'root': body['ThreadOwners'] = []
            if corruption == 't': body['PoolRegistryOwners'][0]['ClosedT'] = 'Char'
            if corruption == 'handle': body['PoolRegistryOwners'][0]['DependentHandle'] = 999
            if corruption == 'target': body['PoolRegistryOwners'][0]['SlotTarget'] = 999
            if corruption == 'free_entry': body['PoolRegistryOwners'][0]['HashCode'] = -1
            with self.subTest(corruption=corruption), self.assertRaises(AssertionError): self.qualify_change(value)

    def test_dead_or_changed_thread_identity_is_rejected(self):
        for change in ('remove', 'id', 'address', 'new'):
            value = baseline(); roots = value['RuntimeTlsPools']['ThreadRootSnapshots']
            if change == 'remove': roots.pop(0)
            if change == 'id': roots[0]['ThreadId'] = 6
            if change == 'address': roots[0]['ThreadAddress'] = 0x9999
            if change == 'new': roots.append({'RootKind': 'SchedulerThread', 'ThreadAddress': 0x9999,
                'ThreadId': 9, 'RegistryIndex': 4, 'TlsRoot': 0, 'PoolFields': []})
            with self.subTest(change=change), self.assertRaises(AssertionError): self.qualify_change(value)

    def test_existing_type_base_target_or_native_slot_replacement_fails(self):
        for change in ('base', 'target', 'slot'):
            value = baseline(); root = value['RuntimeTlsPools']['ThreadRootSnapshots'][0]; body = value['RuntimeTlsPools']['TlsArrayBodies'][0]
            if change == 'base':
                root['TypeStorageRows'][0]['DataWords'][0] = 0xc000
                root['PoolFields'][0]['TypeBase'] = body['ThreadOwners'][0]['TypeBase'] = 0xc000
            if change == 'target':
                root['PoolFields'][0]['Target'] = body['ThreadOwners'][0]['Target'] = body['TlsArray'] = value['Slots'][-1]['Target'] = 0x99999
                body['PoolRegistryOwners'][0]['SlotTarget'] = 0x99999
            if change == 'slot':
                value['Slots'][-1]['SlotAddress'] = body['SlotAddress'] = body['PoolRegistryOwners'][0]['DependentHandle'] = 99
            with self.subTest(change=change), self.assertRaises(AssertionError): self.qualify_change(value)

    def test_new_tombstone_or_non_tls_slot_or_changed_non_tls_pair_fails(self):
        for change in ('tombstone', 'new', 'pair', 'delete'):
            value = baseline()
            if change == 'tombstone': value['Slots'].append({'SlotAddress': 99, 'Target': 0, 'Dependent': 0, 'ClearedTombstone': True})
            if change == 'new': value['Slots'].append({'SlotAddress': 99, 'Target': 0x999, 'Dependent': 0x888, 'ClearedTombstone': False})
            if change == 'pair': value['Slots'][0]['Dependent'] = 0x999
            if change == 'delete': value['Slots'].pop(0)
            value['AllocatedTypedSlots'] = value['AliveCounterSum'] = len(value['Slots'])
            with self.subTest(change=change), self.assertRaises(AssertionError): self.qualify_change(value)

    def test_new_tls_target_alias_or_duplicate_t_thread_fails(self):
        value = add_tls(baseline(), target=0x30000)
        with self.assertRaises(AssertionError): self.qualify_change(value)
        value = baseline(); body = copy.deepcopy(value['RuntimeTlsPools']['TlsArrayBodies'][0])
        value['RuntimeTlsPools']['TlsArrayBodies'].append(body)
        with self.assertRaises(AssertionError): self.qualify_change(value)

    def test_storage_array_resize_preserving_every_prior_field_is_allowed(self):
        initial = baseline(); initial['RuntimeTlsPools']['ThreadRootSnapshots'][0]['TypeStorageRows'][0]['Length'] = 1
        initial['RuntimeTlsPools']['ThreadRootSnapshots'][0]['TypeStorageRows'][0]['DataWords'] = [0xa000]
        changed = add_tls(copy.deepcopy(initial)); root = changed['RuntimeTlsPools']['ThreadRootSnapshots'][0]
        root['TlsRoot'] = root['ModuleStorage']['Address'] = 0x18000
        root['ModuleStorage'].update(Length=2, DataWords=[0x19000, 0])
        root['TypeStorageRows'][0]['Address'] = 0x19000
        tracker = ClosedCycleOwnershipTracker(); tracker.qualify(initial)
        self.assertTrue(tracker.qualify(changed)['Passed'])

    def test_unsafe_resize_or_disappearing_old_field_is_rejected(self):
        changed = baseline(); root = changed['RuntimeTlsPools']['ThreadRootSnapshots'][0]
        root['ModuleStorage']['DataWords'][0] = 0x19000
        root['TypeStorageRows'][0].update(Address=0x19000, Length=1, DataWords=[0])
        root['PoolFields'] = []
        with self.assertRaises(AssertionError): self.qualify_change(changed)

    def test_recorded_historical_plus442_cycle_growth_is_not_excluded_as_runtime_tls(self):
        # These totals were read from the historical real Files-only log before
        # its build directory was unexpectedly removed. This synthetic slot
        # negative control cannot claim the lost original serial was replayed.
        for total in (1630, 2071, 2513):
            value = baseline()
            for index in range(total - 1188):
                value['Slots'].append({'SlotAddress': 100 + index, 'Target': 10000 + index,
                    'Dependent': 0x999, 'ClearedTombstone': False})
            value['AllocatedTypedSlots'] = value['AliveCounterSum'] = len(value['Slots'])
            with self.subTest(recorded_total=total), self.assertRaises(AssertionError): self.qualify_change(value)

    def test_read_boundaries_and_dynamic_symbol_resolution_fail_closed(self):
        for address, count in ((0, 1), (0xffff800000000001, 1), (0xffff800000000000, 4097)):
            with self.assertRaises(AssertionError): checked_address(address, count)
        symbols = parse_symbols('ffffffff81234560 d PoolCell\nffffffff83456780 d OtherCell\n')
        self.assertEqual(0xffffffff81234560, symbol_address(symbols, 'PoolCell'))
        with self.assertRaises(AssertionError): symbol_address(symbols, 'Missing')


def release_packet(page, reason='Release'):
    return ('XFE_BROWSER_RELEASE|HandlesBefore|100|HandlesAfter|0\n'
            'XFE_BROWSER_WEAK_RELEASE|TablesBefore|8|EntriesBefore|20|TablesAfter|0|EntriesAfter|0'
            '|Created|30|Disposed|30|OwnerHandlesCreated|8|OwnerHandlesDisposed|8'
            f'|Page|{page}|Reason|{reason}\n')


def release_fixture():
    log = ''; entries = []; retired = []
    for index, name in enumerate(('Hello.cs', 'Hello.js', 'Readme.md')):
        preview_start = len(log); log += release_packet(100 + index * 2)
        preview_end = len(log); log += 'EDITOR_ACK\n'
        entry = {'SerialStart': 0, 'Path': name, 'EditorId': 3 + index * 2,
                 'Preview': {'BrowserId': 4 + index * 2, 'CloseSerialStart': preview_start,
                             'ClosedToEditor': {'SerialOffset': preview_end}}}
        if index < 2:
            entry['CloseSerialStart'] = len(log); log += release_packet(101 + index * 2)
            retired.append({'WindowId': entry['EditorId'], 'SerialOffset': len(log)})
            log += f'XFE_EDITOR_BACKGROUND|Retired|Window|{entry["EditorId"]}|Generation|1|Thread|5\n'
        entries.append(entry)
    return log, entries, retired


class ReleaseAndPhysicalGuards(unittest.TestCase):
    def test_all_five_corresponding_owner_packets_must_release_dom_and_weak_resources(self):
        log, entries, retired = release_fixture()
        proof = closed_cycle_release_observation(log, entries, retired, len(log), set())
        self.assertEqual(5, len(proof['ExpectedOwnerReleases']))
        self.assertTrue(proof['Passed'])

    def test_missing_partial_malformed_or_unreleased_weak_packet_is_rejected(self):
        log, entries, retired = release_fixture()
        for bad in (log.replace('|Disposed|30', '|Disposed|29', 1),
                    log.replace('|OwnerHandlesDisposed|8', '|OwnerHandlesDisposed|7', 1),
                    log.replace('|TablesAfter|0', '|TablesAfter|1', 1),
                    log.replace('XFE_BROWSER_WEAK_RELEASE', 'BROKEN', 1),
                    log.replace('|Reason|Release\n', '|Reason|Release', 1)):
            with self.subTest(bad=bad[-100:]), self.assertRaises(AssertionError):
                closed_cycle_release_observation(bad, entries, retired, len(bad), set())

    def test_historical_duplicate_or_wrong_close_interval_packet_is_rejected(self):
        log, entries, retired = release_fixture()
        with self.assertRaises(AssertionError): closed_cycle_release_observation(log, entries, retired, len(log), {100})
        with self.assertRaises(AssertionError): closed_cycle_release_observation(log.replace('|Page|102|', '|Page|100|'), entries, retired, len(log), set())
        entries[0]['Preview']['ClosedToEditor']['SerialOffset'] = entries[0]['Preview']['CloseSerialStart']
        with self.assertRaises(AssertionError): closed_cycle_release_observation(log, entries, retired, len(log), set())

    def test_extra_replacement_packet_must_release_but_does_not_replace_expected_close(self):
        log, entries, retired = release_fixture()
        suffix = release_packet(100, 'Replace')
        proof = closed_cycle_release_observation(log + suffix, entries, retired, len(log + suffix), set())
        self.assertEqual(6, len(proof['AllReleasePackets']))
        bad = log + suffix.replace('|EntriesAfter|0', '|EntriesAfter|1')
        with self.assertRaises(AssertionError): closed_cycle_release_observation(bad, entries, retired, len(bad), set())

    def test_closed_physical_budget_keeps_original64mib_fixed_baseline(self):
        def observation(free): return {'ConsecutiveNativeSamples': [{'RawFreePages': free}, {'RawFreePages': free}]}
        self.assertTrue(closed_cycle_physical_budget([observation(50000), observation(33616)])['Passed'])
        self.assertFalse(closed_cycle_physical_budget([observation(50000), observation(33615)])['Passed'])
        self.assertFalse(closed_cycle_physical_budget([observation(50000), observation(50000), observation(30000)])['Passed'])


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