Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451"""Exact-build compatibility inspection and closed-game update reconciliation.
No executable is ever modified. A reviewed profile is necessary but not sufficient:the executable, stock scripts, import hooks, WNB signature/relative targets andowned dependency chain must agree. Static agreement is not live ABI qualification."""import hashlibimport jsonimport osfrom pathlib import Pathimport reimport structimport time
from . import supportfrom .debug_journal import durable_jsonfrom .pe_image import PEImage
PROFILE_PATH = support.ROOT / 'native/compatibility.json'REGISTRATION = ( 'BA F8 00 00 00 48 8D 4D F0 41 B8 10 00 00 00 E8 ?? ?? ?? ?? ' '48 8B 7D F0 33 D2 4C 8B 45 F8 48 8B CF E8 ?? ?? ?? ?? ' '48 85 FF 74 ?? E8 ?? ?? ?? ?? 48 8D 15 ?? ?? ?? ?? ' '48 8B C8 E8 ?? ?? ?? ?? 4C 8D 05 ?? ?? ?? ?? ' '89 45 10 48 8D 55 10 48 8B CF E8 ?? ?? ?? ?? ' '48 8B F8 EB ?? 48 8B FB E8 ?? ?? ?? ?? 48 8B D7 48 8B C8 E8 ?? ?? ?? ??')STRING_HELPER = ( '8B 15 ?? ?? ?? ?? 33 C9 41 B9 0E 00 00 00 89 54 24 ?? ' '41 B8 01 00 00 00 E8 ?? ?? ?? ?? 44 8B 44 24 ?? 48 8B C8 ' '48 8B 15 ?? ?? ?? ?? 48 89 44 24 ?? E8 ?? ?? ?? ??')OLD_STRING_HELPER = STRING_HELPER.replace( '41 B8 01 00 00 00 E8', '41 B8 01 00 00 00 48 C7 44 24 20 00 00 00 00 E8')CALL_OFFSETS = dict(alloc=15, memset=33, name_pool=43, add_name=58, function_ctor=80, script_system=93, register_global=104)REGISTRAR_PROLOGUE = bytes.fromhex('48 89 6C 24 10 48 89 74 24 18 57 48 83 EC 20')REQUIRED_IMPORTS = { 'xinput9_1_0.dll': ['XInputGetState'], 'ws2_32.dll': [2], # bind is imported by ordinal; PathInput explicitly supports ordinal 2. 'user32.dll': ['CreateWindowExA', 'GetForegroundWindow', 'SetCursorPos', 'ClipCursor', 'SetCapture', 'SetForegroundWindow', 'GetAsyncKeyState'],}
def sha(data): return hashlib.sha256(data).hexdigest()
def relative(image, site, displacement, length): return site + length + struct.unpack('<i', image.read_rva(site + displacement, 4))[0]
def pattern_matches(data, pattern): expression = b''.join(b'.' if part == '??' else re.escape(bytes([int(part, 16)])) for part in pattern.split()) return [m.start() for m in re.finditer(expression, data, re.DOTALL)]
def native_bindings(image): """Mirror the pinned WNB resolver and demand full, unambiguous agreement.""" sites = image.find(REGISTRATION) if len(sites) < 32: raise ValueError('compatibility: insufficient WNB registration sites') targets = {} for name, offset in CALL_OFFSETS.items(): values = {relative(image, site + offset, 1, 5) for site in sites} if len(values) != 1: raise ValueError('compatibility: ambiguous WNB target ' + name) target = values.pop() if not any(s.characteristics & 0x20000000 and s.rva <= target < s.rva + s.virtual_size for s in image.sections): raise ValueError('compatibility: WNB call target is not executable') targets[name] = target if image.read_rva(targets['register_global'], 15) != REGISTRAR_PROLOGUE: raise ValueError('compatibility: WNB registration hook prologue changed') logs = [] for site in sites: name = relative(image, site + 48, 3, 7) if image.read_rva(name, 11) == b'LogChannel\0': logs.append(relative(image, site + 63, 3, 7)) if len(logs) != 1: raise ValueError('compatibility: LogChannel registration must be unique') log = logs[0] code = image.read_rva(log, 0xc0) if pattern_matches(code, OLD_STRING_HELPER): raise ValueError('compatibility: old string helper would prevent the reviewed fallback') matches = pattern_matches(code, STRING_HELPER) if len(matches) != 1: raise ValueError('compatibility: updated string helper must be unique') anchor = log + matches[0] helpers = {name: relative(image, anchor + offset, displacement, length) for name, offset, displacement, length in ( ('empty_length', 0, 2, 6), ('buffer_alloc', 24, 1, 5), ('empty_string', 37, 3, 7), ('buffer_copy', 49, 1, 5))} # Match WNB's first image-relative LEA in the native's first 0x40 bytes. opcode = None for i in range(0x40): if code[i] in (0x48, 0x4c) and code[i+1] == 0x8d and code[i+2] & 0xc7 == 5: candidate = relative(image, log + i, 3, 7) if 0 <= candidate < image.size_of_image: opcode = candidate break if opcode is None: raise ValueError('compatibility: opcode-table reference missing') helpers['opcode_table'] = opcode for name, address in helpers.items(): executable = name in ('buffer_alloc', 'buffer_copy') if not any(s.rva <= address < s.rva + s.virtual_size and bool(s.characteristics & 0x20000000) == executable for s in image.sections): raise ValueError('compatibility: invalid helper target ' + name) return dict(registration_count=len(sites), targets=targets, log_channel=log, helpers=helpers)
def profile_for(data): profiles = json.loads(PROFILE_PATH.read_text()) digest = sha(data) matches = [p for p in profiles['profiles'] if p['executable_sha256'] == digest] if len(matches) != 1: raise ValueError('unsupported_game_build: no reviewed exact executable profile; do not rewrite receipts') return matches[0]
def verify_receipt(receipt): """A changed hash alone cannot authorize this new dependency chain.""" profiles = json.loads(PROFILE_PATH.read_text())['profiles'] executable = receipt.get('executable', {}) known = next((p for p in profiles if p['executable_sha256'] == executable.get('sha256')), None) if not known: if receipt.get('compatibility'): raise ValueError('unsupported_game_build: compatibility receipt has no reviewed profile') legacy = {h for p in profiles for h in p['previous_executable_sha256']} if executable.get('sha256') not in legacy: raise ValueError('unsupported_game_build: unknown legacy executable hash') return None # Only the explicitly recorded, historically qualified build. if receipt.get('compatibility', {}).get('profile') != known['id']: raise ValueError('compatibility: executable hash alone is insufficient; use reconcile-game-update') checked, profile = inspect_game() expected_dir = support.manage.GAME / 'bin/x64_dx12' if len(receipt['files']) != len(profile['native_files']): raise ValueError('compatibility: native dependency set changed') actual = {Path(e['path']).name: e['sha256'] for e in receipt['files']} if actual != profile['native_files'] or any(Path(e['path']) != expected_dir / Path(e['path']).name for e in receipt['files']): raise ValueError('compatibility: native chain does not match reviewed profile') return checked
def inspect_game(game=None): game = Path(game or support.manage.GAME) executable = game / 'bin/x64_dx12/witcher3.exe' raw = executable.read_bytes() profile = profile_for(raw) image = PEImage(raw) if len(raw) != profile['executable_size'] or image.timestamp != profile['pe_timestamp']: raise ValueError('compatibility: executable metadata differs') imports = image.imports() for dll, names in REQUIRED_IMPORTS.items(): if not set(names) <= set(imports.get(dll, [])): raise ValueError('compatibility: required import missing in ' + dll) bindings = native_bindings(image) if bindings != profile['bindings']: raise ValueError('compatibility: native binding targets differ from reviewed profile') scripts = game / 'content/content0/scripts' manifest = {p.relative_to(scripts).as_posix(): support.manage.digest(p) for p in sorted(scripts.rglob('*.ws'))} digest = sha(json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode()) if len(manifest) != profile['stock_script_count'] or digest != profile['stock_scripts_sha256']: raise ValueError('compatibility: stock script interface tree differs') steam = (game.parent.parent / 'appmanifest_292030.acf').read_text() # Metadata corroborates the exact executable; it never authorizes an unknown hash. for key, expected in [('appid', '292030'), ('buildid', profile['steam_build']), ('TargetBuildID', profile['steam_build']), ('StateFlags', '4')]: values = re.findall(r'"' + key + r'"\s*"([^"]*)"', steam, re.IGNORECASE) if values != [expected]: raise ValueError('compatibility: Steam build is mismatched, incomplete or ambiguous') evidence = support.ROOT / profile['qualification'] qualification = dict(report=profile['qualification'], live_qualified=False) if evidence.is_file(): summary = json.loads(evidence.read_text()) qualification['live_qualified'] = ( summary.get('passed') is True and summary.get('executable_sha256') == profile['executable_sha256'] and summary.get('native_files') == profile['native_files']) return dict(ok=True, profile=profile['id'], executable_sha256=sha(raw), steam_build=profile['steam_build'], bindings=bindings, stock_script_count=len(manifest), stock_scripts_sha256=digest, qualification=qualification), profile
def closed_boundary(): from . import debugger, native support.manage.ensure_game_closed() debugger.reject_legacy() pending = debugger.starting_status() blocked_startup = False if pending: startup = pending.get('startup', {}) broker = pending.get('broker') recovered = (startup.get('phase') == 'recovered' and startup.get('cleanup_required') is False and startup.get('audits_final') is True and startup.get('recovery_result', {}).get('ok') is True) blocked_startup = (not recovered or pending.get('state') == 'starting' or bool(broker and debugger.identity(broker['pid']) == broker)) if debugger.active_owner() or blocked_startup: raise RuntimeError('compatibility: active owner/startup blocks reconciliation') if native.STATE.exists() and json.loads(native.STATE.read_text()).get('uncertain'): raise RuntimeError('compatibility: unresolved native delivery blocks reconciliation')
def file_audit(): return dict(saves=support.manage.files(support.manage.DOCS / 'gamesaves'), mods=support.manage.files(support.manage.GAME / 'mods'), settings={name: support.manage.digest(support.manage.DOCS / name) for name in ('user.settings', 'dx12user.settings', 'input.settings')})
def sync_directory(path): fd = os.open(path, os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd)
def durable_bytes(path, data): with path.open('xb') as stream: stream.write(data) stream.flush() os.fsync(stream.fileno()) sync_directory(path.parent)
def replace_adapter(target, data, folder, manifest, expected): """Record ownership before writing, so even an interrupted short write is recoverable.""" if sha(data) != expected: raise RuntimeError('compatibility: candidate bytes changed before staging') item = next(b for b in manifest['backups'] if b['path'] == str(target)) previous = item['before'] if expected == item['after'] else item['after'] before = target.lstat() identity_before = (before.st_dev, before.st_ino) if target.is_symlink() or support.manage.digest(target) != previous: raise RuntimeError('compatibility: target changed before staging') stages = manifest.setdefault('staged', []) temporary = target.with_name(target.name + '.' + folder.name + '-' + str(len(stages)) + '.devkit-next') staged = dict(path=str(temporary), device=None, inode=None) stages.append(staged) # Persist the intent before creation. A crash before recording the inode # leaves an explicitly unresolved stage, never a falsely successful cleanup. durable_json(folder / 'transaction.json', manifest) with temporary.open('xb') as stream: identity = os.fstat(stream.fileno()) staged.update(device=identity.st_dev, inode=identity.st_ino) sync_directory(temporary.parent) durable_json(folder / 'transaction.json', manifest) stream.write(data) stream.flush() os.fsync(stream.fileno()) if support.manage.digest(temporary) != expected: raise RuntimeError('compatibility: staged adapter bytes differ') current = target.lstat() if (target.is_symlink() or (current.st_dev, current.st_ino) != identity_before or support.manage.digest(target) != previous): raise RuntimeError('compatibility: target changed before replacement') temporary.replace(target) sync_directory(target.parent)
def plan(): from . import lifecycle checked, profile = inspect_game() receipt_path = support.STATE_ROOT / 'native/deployment.json' receipt = json.loads(receipt_path.read_text()) bridge = json.loads((support.STATE_ROOT / 'deployment.json').read_text()) from . import background background.verify() for path, digest in lifecycle.owned_scripts(bridge): if not path.is_file() or support.manage.digest(path) != digest: raise RuntimeError('Installed dependency differs from verified receipt: ' + str(path)) expected_dir = support.manage.GAME / 'bin/x64_dx12' files = receipt['files'] if len(files) != 3 or {Path(e['path']).name for e in files} != {'PathInput.asi', 'WitcherNativeBridge.asi', 'dinput8.dll'}: raise ValueError('compatibility: unexpected native dependency set') replacements = [] for entry in files: path = Path(entry['path']) if path != expected_dir / path.name or path.is_symlink() or support.manage.digest(path) != entry['sha256']: raise RuntimeError('Installed dependency differs from verified receipt: ' + str(path)) target_hash = profile['native_files'][path.name] if path.name == 'dinput8.dll' and entry['sha256'] != target_hash: raise ValueError('compatibility: unsupported ASI loader') if entry['sha256'] != target_hash: candidate = support.ROOT / 'native/build' / path.name if not candidate.is_file() or support.manage.digest(candidate) != target_hash: raise ValueError('compatibility: build the reviewed adapters with python3 native/build.py first') replacements.append(dict(path=str(path), source=str(candidate), before=entry['sha256'], after=target_hash)) old_executable = receipt.get('executable', {}) if old_executable.get('path') != str(expected_dir / 'witcher3.exe'): raise ValueError('compatibility: executable receipt path differs') if old_executable.get('sha256') not in profile['previous_executable_sha256'] + [profile['executable_sha256']]: raise ValueError('compatibility: unsupported executable receipt provenance') if old_executable['sha256'] == profile['executable_sha256']: if receipt.get('compatibility', {}).get('profile') != profile['id']: raise ValueError('compatibility: current executable pin lacks its checked reconciliation marker') return dict(checked, applied=False, replacements=replacements, previous_executable_sha256=old_executable['sha256'], executable_modified=False), receipt, profile
def reconcile(apply=False): """Caller holds installation lock, including the read-only plan.""" closed_boundary() result, receipt, profile = plan() if not apply: return result receipt_path = support.STATE_ROOT / 'native/deployment.json' if not result['replacements'] and receipt['executable']['sha256'] == result['executable_sha256']: verify_receipt(receipt) return dict(result, applied=False, already_reconciled=True) folder = support.ARTIFACTS / ('game-update-' + str(time.time_ns())) folder.mkdir() sync_directory(folder.parent) original = receipt_path.read_bytes() durable_bytes(folder / 'previous-native-receipt.json', original) manifest = dict(result, status='prepared', backups=[], baseline=file_audit()) for item in result['replacements']: name = Path(item['path']).name backup = Path(item['path']).read_bytes() if sha(backup) != item['before']: raise RuntimeError('compatibility: adapter changed before backup') durable_bytes(folder / name, backup) manifest['backups'].append(dict(item, backup=name)) updated = dict(receipt) updated['files'] = [dict(e, sha256=profile['native_files'][Path(e['path']).name]) for e in receipt['files']] updated['executable'] = dict(receipt['executable'], sha256=result['executable_sha256']) updated['compatibility'] = dict(profile=profile['id'], steam_build=profile['steam_build'], checks=result, evidence=profile['qualification'], rollback=str(folder)) durable_json(folder / 'next-native-receipt.json', updated) durable_json(folder / 'transaction.json', manifest) try: closed_boundary() fresh, _, _ = plan() if fresh != result or receipt_path.read_bytes() != original: raise RuntimeError('compatibility: installation changed during reconciliation') for item in result['replacements']: target = Path(item['path']) replace_adapter(target, Path(item['source']).read_bytes(), folder, manifest, item['after']) # Recheck the game and stock interfaces before accepting the new chain. closed_boundary() final, _ = inspect_game() if final['executable_sha256'] != result['executable_sha256']: raise RuntimeError('compatibility: game changed during reconciliation') durable_json(receipt_path, updated) from . import lifecycle verified = lifecycle.verify() if file_audit() != manifest['baseline']: raise RuntimeError('compatibility: saves/mods/settings changed; inspect retained baseline') durable_json(folder / 'transaction.json', dict(manifest, status='complete')) return dict(result, applied=True, rollback_artifacts=str(folder), verified=verified) except Exception as error: try: rollback(folder) except Exception as cleanup_error: try: durable_json(folder / 'transaction.json', dict( manifest, status='rollback_required', error=str(error), rollback_error=str(cleanup_error))) except Exception as evidence_error: raise RuntimeError('compatibility: reconciliation and recovery evidence persistence failed; inspect ' + str(folder) + '; original: ' + str(error) + '; rollback: ' + str(cleanup_error) + '; evidence: ' + str(evidence_error)) from error raise RuntimeError('compatibility: reconciliation failed and rollback is blocked; inspect ' + str(folder) + '; original: ' + str(error) + '; rollback: ' + str(cleanup_error)) from error raise
def staged_files(folder, manifest): """Validate every stage before rollback writes, then revalidate at cleanup.""" result = [] for index, staged in enumerate(manifest.get('staged', [])): temporary = Path(staged['path']) expected_names = {name + '.' + folder.name + '-' + str(index) + '.devkit-next' for name in ('PathInput.asi', 'WitcherNativeBridge.asi')} if temporary.parent != support.manage.GAME / 'bin/x64_dx12' or temporary.name not in expected_names: raise ValueError('compatibility: invalid staged path') if not temporary.exists() and not temporary.is_symlink(): continue stat = temporary.lstat() if (temporary.is_symlink() or not temporary.is_file() or (stat.st_dev, stat.st_ino) != (staged['device'], staged['inode'])): raise RuntimeError('compatibility: staged file identity changed or unresolved; rollback refused') data = temporary.read_bytes() retained = folder / ('staged-' + str(index) + '.bin') if retained.exists() and retained.read_bytes() != data: raise RuntimeError('compatibility: retained staged evidence differs') result.append((staged, temporary, data, retained)) return result
def rollback(folder): """Restore only hash-owned adapter bytes/receipt; never roll back Steam.""" closed_boundary() folder = Path(folder).resolve() if folder.parent != support.ARTIFACTS.resolve() or not folder.name.startswith('game-update-'): raise ValueError('compatibility: rollback must name a retained game-update artifact') manifest = json.loads((folder / 'transaction.json').read_text()) old = (folder / 'previous-native-receipt.json').read_bytes() receipt_path = support.STATE_ROOT / 'native/deployment.json' current = receipt_path.read_bytes() current_receipt = json.loads(current) expected_receipt = json.loads((folder / 'next-native-receipt.json').read_text()) if current != old and current_receipt != expected_receipt: raise RuntimeError('compatibility: receipt changed after transaction; rollback refused') staged_files(folder, manifest) for item in manifest['backups']: path = Path(item['path']) if item['backup'] not in ('PathInput.asi', 'WitcherNativeBridge.asi'): raise ValueError('compatibility: invalid rollback backup name') if path != support.manage.GAME / 'bin/x64_dx12' / item['backup'] or path.is_symlink(): raise ValueError('compatibility: invalid rollback target') if support.manage.digest(folder / item['backup']) != item['before']: raise RuntimeError('compatibility: rollback backup changed') actual = support.manage.digest(path) if actual not in (item['before'], item['after']): raise RuntimeError('compatibility: externally changed adapter; rollback refused') if actual == item['after']: identity = path.lstat() if not any(Path(s['path']).name.startswith(path.name + '.' + folder.name + '-') and (s['device'], s['inode']) == (identity.st_dev, identity.st_ino) for s in manifest.get('staged', [])): raise RuntimeError('compatibility: adapter replacement identity changed; rollback refused') for item in manifest['backups']: target = Path(item['path']) if support.manage.digest(target) != item['before']: replace_adapter(target, (folder / item['backup']).read_bytes(), folder, manifest, item['before']) for staged, temporary, data, retained in staged_files(folder, manifest): # Preserve even a partial write as evidence, rather than claiming its bytes # are the approved adapter or deleting an unexpected file without a copy. if not retained.exists(): durable_bytes(retained, data) stat = temporary.lstat() if (stat.st_dev, stat.st_ino) != (staged['device'], staged['inode']) or temporary.read_bytes() != data: raise RuntimeError('compatibility: staged file changed during recovery') temporary.unlink() sync_directory(temporary.parent) durable_json(receipt_path, json.loads(old)) durable_json(folder / 'transaction.json', dict(manifest, status='rolled_back')) return dict(ok=True, rolled_back=True, executable_modified=False, note='Updated Steam executable retained; previous receipt may intentionally block launch.')