Verstack Local
f73167de85
Package Apple Silicon VM exports and cached runtime switching with local cabinet controls and conagent provisioning. Preserve experimental SPIKE 2 emulation. Improve preparation concurrency, Ghidra checkpoints, signature importing, sound indexing and client-side spectrograms. Include regression tests and validation notes.
178 lines
10 KiB
Python
178 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Recover and round-trip Godot source with pinned GDRE tools; never run game scripts."""
|
|
import hashlib
|
|
import json
|
|
import mmap
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from media_extract import pck_entries, safe_name
|
|
from progress import report
|
|
|
|
REVISION = 2
|
|
SCRIPT_LIMIT = 8 * 1024**2
|
|
|
|
|
|
def sha(path):
|
|
with path.open('rb') as stream:
|
|
return hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
|
|
|
|
def checked_text(path):
|
|
if path.is_symlink() or not path.is_file() or path.stat().st_size > SCRIPT_LIMIT:
|
|
raise ValueError('Recovered script is missing or exceeds the 8 MiB limit')
|
|
text = path.read_text(encoding='utf-8')
|
|
if '\0' in text or not text.strip():
|
|
raise ValueError('Recovered script is empty or contains binary data')
|
|
return text
|
|
|
|
|
|
def references(text):
|
|
return sorted(set(re.findall(r'["\']((?:res://|uid://)[^"\'\r\n]+)["\']', text)))
|
|
|
|
|
|
class Recovery:
|
|
def __init__(self, request):
|
|
self.source = Path(request['input_dir'])
|
|
self.output = Path(request['output_dir'])
|
|
self.settings = request.get('settings') or {}
|
|
self.tool = Path(self.settings['tool']).resolve()
|
|
for name, expected in self.settings['expected_files'].items():
|
|
if PurePosixPath(name).name != name or sha(self.tool.parent / name) != expected:
|
|
raise ValueError('Godot decompiler installation differs from its pin')
|
|
if self.tool.name not in self.settings['expected_files']:
|
|
raise ValueError('Godot decompiler executable must be pinned')
|
|
self.rows = []; self.packages = []; self.used = 0
|
|
self.budget = request['workspace_bytes'] // 2
|
|
|
|
def invoke(self, args, work):
|
|
env = os.environ.copy()
|
|
for key, folder in [('XDG_DATA_HOME','data'),('XDG_CONFIG_HOME','config'),('XDG_CACHE_HOME','cache')]:
|
|
target = work / 'user' / folder; target.mkdir(parents=True, exist_ok=True); env[key] = str(target)
|
|
# Keep decompiler output bounded on disk. Only our fixed diagnostics
|
|
# become coverage errors; tool logs remain temporary.
|
|
log = work / 'decompiler.log'
|
|
with log.open('wb') as stream:
|
|
result = subprocess.run([str(self.tool), '--headless', *args], cwd=work, env=env,
|
|
stdin=subprocess.DEVNULL, stdout=stream, stderr=subprocess.STDOUT,
|
|
timeout=300)
|
|
if result.returncode:
|
|
raise ValueError('Godot decompiler could not process this input or bytecode version')
|
|
|
|
def validate(self, scripts, version, work):
|
|
original = work / 'validation-input'; original.mkdir()
|
|
compiled = work / 'validation-compiled'
|
|
returned = work / 'validation-returned'
|
|
for i, (_, path) in enumerate(scripts):
|
|
checked_text(path)
|
|
shutil.copyfile(path, original / f'script-{i:05}.gd')
|
|
if not scripts:return
|
|
self.invoke([f'--compile={original}/*.gd', f'--bytecode={version}', f'--output={compiled}'], work)
|
|
self.invoke([f'--decompile={compiled}/*.gdc', f'--bytecode={version}', f'--output={returned}'], work)
|
|
for i, (_, path) in enumerate(scripts):
|
|
if checked_text(path).strip() != checked_text(returned / f'script-{i:05}.gd').strip():
|
|
raise ValueError('Recovered script changed during the compile/decompile validation')
|
|
|
|
def publish(self, row, path, logical):
|
|
text = checked_text(path)
|
|
data = text.encode('utf-8'); self.used += len(data)
|
|
if self.used > self.budget:raise ValueError('Recovered scripts exceed the output budget')
|
|
dest = self.output / 'scripts' / safe_name(logical)
|
|
if dest.exists():raise ValueError('Multiple scripts resolve to the same output path')
|
|
dest.parent.mkdir(parents=True, exist_ok=True); dest.write_bytes(data)
|
|
if PurePosixPath(logical).name == 'spike_game_sound_ids.gd':
|
|
from sound_names import enrich
|
|
row = {**row, 'sound_catalog':enrich(text, self.source / row['source'], self.source)}
|
|
self.rows.append({**row, 'output':dest.relative_to(self.output).as_posix(),
|
|
'status':'recovered', 'output_sha256':hashlib.sha256(data).hexdigest(),
|
|
'resource_references':references(text), 'name_confidence':'confirmed_container_path'})
|
|
|
|
def package(self, path):
|
|
rel = path.relative_to(self.source).as_posix()
|
|
context = {'source':rel, 'source_sha256':sha(path), 'processing_revision':REVISION}
|
|
inventory = []
|
|
try:
|
|
with path.open('rb') as stream, mmap.mmap(stream.fileno(), 0, access=mmap.ACCESS_READ) as data:
|
|
_, _, major, minor, patch = struct.unpack_from('<5I', data)
|
|
version = f'{major}.{minor}.{patch}'
|
|
for name, offset, length, digest in pck_entries(data):
|
|
if not name.endswith(('.gd', '.gdc')):continue
|
|
if length > SCRIPT_LIMIT:raise ValueError('Godot script exceeds the 8 MiB limit')
|
|
body = data[offset:offset+length]
|
|
if hashlib.md5(body).digest() != digest:raise ValueError('Godot script checksum mismatch')
|
|
inventory.append({'entry':name, 'offset':offset, 'length':length,
|
|
'script_sha256':hashlib.sha256(body).hexdigest()})
|
|
if len(inventory)>10000:raise ValueError('Godot script count exceeds the limit')
|
|
self.packages.append({**context, 'engine_version':version, 'script_count':len(inventory)})
|
|
if not inventory:return
|
|
with tempfile.TemporaryDirectory(prefix='godot-', dir=self.output.parent) as tmp:
|
|
work = Path(tmp); recovered = work / 'recovered'
|
|
self.invoke([f'--recover={path}', '--scripts-only', f'--output={recovered}'], work)
|
|
scripts = [(row, recovered / PurePosixPath(row['entry']).with_suffix('.gd')) for row in inventory]
|
|
names=[str(PurePosixPath(row['entry']).with_suffix('.gd')) for row,_ in scripts]
|
|
if len(names)!=len(set(names)):raise ValueError('Duplicate recovered script path')
|
|
self.validate(scripts, version, work)
|
|
if self.used+sum(p.stat().st_size for _,p in scripts)>self.budget:
|
|
raise ValueError('Recovered scripts exceed the output budget')
|
|
for row, recovered_path in scripts:
|
|
self.publish({**context, **row, 'engine_version':version,
|
|
'validation':'compile-decompile-source-match'}, recovered_path,
|
|
f"{rel}/{PurePosixPath(row['entry']).with_suffix('.gd')}")
|
|
except (ValueError, OSError, UnicodeError, struct.error, subprocess.SubprocessError) as error:
|
|
message = str(error) if isinstance(error,ValueError) else 'Godot package recovery failed'
|
|
for row in inventory or [{}]:
|
|
self.rows.append({**context, **row, 'status':'failed', 'error':message})
|
|
|
|
def loose(self, path):
|
|
rel = path.relative_to(self.source).as_posix()
|
|
row = {'source':rel, 'source_sha256':sha(path), 'entry':rel, 'processing_revision':REVISION}
|
|
try:
|
|
if path.suffix == '.gd':
|
|
self.publish({**row, 'validation':'original-source'}, path, rel);return
|
|
version = self.settings.get('bytecode_version')
|
|
if not version:raise ValueError('Loose compiled script needs an explicit bytecode_version setting or its original PCK')
|
|
with tempfile.TemporaryDirectory(prefix='godot-loose-', dir=self.output.parent) as tmp:
|
|
work = Path(tmp); recovered = work / 'recovered'
|
|
self.invoke([f'--decompile={path}', f'--bytecode={version}', f'--output={recovered}'], work)
|
|
output = recovered / path.with_suffix('.gd').name
|
|
self.validate([(row, output)], version, work)
|
|
self.publish({**row,'engine_version':version,'validation':'compile-decompile-source-match'}, output,
|
|
str(PurePosixPath(rel).with_suffix('.gd')))
|
|
except (ValueError,OSError,UnicodeError,subprocess.SubprocessError) as error:
|
|
self.rows.append({**row,'status':'failed','error':str(error) if isinstance(error,ValueError) else 'Godot script recovery failed'})
|
|
|
|
def run(self):
|
|
inputs = [p for p in sorted(self.source.rglob('*')) if not p.is_symlink() and p.is_file() and p.suffix in ('.pck','.gd','.gdc')]
|
|
for i,path in enumerate(inputs):
|
|
report('Recovering Godot scripts',i,len(inputs),'inputs',path.relative_to(self.source).as_posix(),force=True)
|
|
if path.suffix == '.pck':self.package(path)
|
|
else:self.loose(path)
|
|
recovered = sum(r['status']=='recovered' for r in self.rows)
|
|
failed = len(self.rows)-recovered
|
|
evidence = {'schema':1,'processing_revision':REVISION,'tool_version':self.settings.get('version'),
|
|
'tool_files':self.settings['expected_files'],'packages':self.packages,'scripts':self.rows,
|
|
'summary':{'recovered':recovered,'failed':failed,'packages':len(self.packages)}}
|
|
(self.output/'godot-scripts-evidence.json').write_text(json.dumps(evidence,indent=2))
|
|
report('Godot scripts recovered',recovered,len(self.rows),'scripts',force=True)
|
|
return {'protocol':1,'layer':'derived','coverage':'partial' if failed else 'complete',
|
|
'warnings':[f'{failed} Godot scripts or packages could not be recovered; see the script recovery report.'] if failed else [],
|
|
'files':[p.relative_to(self.output).as_posix() for p in sorted(self.output.rglob('*')) if p.is_file()]}
|
|
|
|
|
|
def run(request):
|
|
if request['protocol']!=1:raise ValueError('Unsupported protocol')
|
|
return Recovery(request).run()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
request=json.loads(Path(sys.argv[1]).read_text())
|
|
try:result=run(request)
|
|
except Exception as error:
|
|
Path(request['result_file']).write_text(json.dumps({'protocol':1,'error':str(error) if isinstance(error,ValueError) else 'Godot recovery failed; check the tool installation and workspace'}));raise
|
|
Path(request['result_file']).write_text(json.dumps(result))
|