64 lines
4.1 KiB
Python
64 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Automatic wrapper dispatch for pinned spike-spk extraction; never runs game code."""
|
|
import hashlib, json, pathlib, tempfile, zipfile, sys
|
|
import spike_extract, zip_extract, tar_extract
|
|
|
|
def run(request):
|
|
if request['protocol']!=1: raise ValueError('unsupported protocol')
|
|
source=pathlib.Path(request['input_dir']);output=pathlib.Path(request['output_dir']);settings=request['settings']
|
|
packages=[p for p in sorted(source.rglob('*')) if p.is_file() and not p.is_symlink() and (p.name.lower().endswith('.spk') or p.name.lower().endswith('.spk.zip'))]
|
|
if settings.get('input_path'):packages=[p for p in packages if p.relative_to(source).as_posix()==settings['input_path']]
|
|
if len(packages)!=1: raise ValueError('automatic wrapper dispatch requires exactly one package in the input snapshot')
|
|
package=packages[0];rel=package.relative_to(source).as_posix()
|
|
with package.open('rb') as f:digest=hashlib.file_digest(f,'sha256').hexdigest()
|
|
with package.open('rb') as f:magic=f.read(8)
|
|
inventory=[]
|
|
with tempfile.TemporaryDirectory(prefix='wrapper-',dir=output.parent) as work:
|
|
work=pathlib.Path(work);inputs=source
|
|
if zipfile.is_zipfile(package):
|
|
with zipfile.ZipFile(package) as z:
|
|
inventory=[{'entry':e.filename,'bytes':e.file_size,'crc32':f'{e.CRC:08x}'} for e in z.infolist()]
|
|
first=next((e for e in z.infolist() if e.filename.endswith('.000')),None)
|
|
if first is None: raise ValueError('ZIP has no first split-package part')
|
|
with z.open(first) as f:magic=f.read(8)
|
|
if magic[:6]!=b'LUKS\xba\xbe':
|
|
inputs=work/'zip';inputs.mkdir()
|
|
selected=work/'selected';selected.mkdir()
|
|
import shutil
|
|
shutil.copyfile(package,selected/package.name)
|
|
zip_extract.run({**request,'input_dir':str(selected),'output_dir':str(inputs)})
|
|
if magic[:6]==b'LUKS\xba\xbe':
|
|
import luks_extract
|
|
inputs=work/'decrypted';inputs.mkdir()
|
|
luks_extract.run({**request,'output_dir':str(inputs),'settings':{**settings,'input_path':rel}})
|
|
elif magic[:4] not in (b'SPKS',b'hsqs') and magic[:2]!=b'\x1f\x8b': raise ValueError('unsupported SPIKE package wrapper')
|
|
if magic[:2]==b'\x1f\x8b':
|
|
parts=work/'gzip';parts.mkdir()
|
|
prefix,payload,offset=tar_extract.split_prefix(package,parts,request['workspace_bytes']//2)
|
|
bootstrap=output/'bootstrap';bootstrap.mkdir()
|
|
result=tar_extract.run({**request,'output_dir':str(bootstrap)},prefix)
|
|
inventory=[{'entry':'bootstrap.tar.gz','offset':0,'bytes':offset}, {'entry':'payload.spk','offset':offset,'bytes':payload.stat().st_size}]
|
|
if payload.stat().st_size:
|
|
result=spike_extract.run({**request,'input_dir':str(parts)})
|
|
inputs=parts
|
|
else:
|
|
result=spike_extract.run({**request,'input_dir':str(inputs)})
|
|
# Preserve every recovered inner wrapper file. It is needed to reproduce
|
|
# extraction independently of credentials and outer-wrapper reimport.
|
|
if inputs!=source:
|
|
for path in sorted(inputs.rglob('*')):
|
|
if not path.is_file() or path.is_symlink():continue
|
|
dest=output/'wrapper-payload'/path.relative_to(inputs);dest.parent.mkdir(parents=True,exist_ok=True)
|
|
import shutil
|
|
shutil.copyfile(path,dest)
|
|
evidence={'schema':1,'source':rel,'source_sha256':digest,'wrapper_magic':magic.hex(),'outer_inventory':inventory,'payload_inventory_complete':True,'wrapper_retained':True}
|
|
(output/'wrapper-evidence.json').write_text(json.dumps(evidence,indent=2))
|
|
result['files']=[p.relative_to(output).as_posix() for p in sorted(output.rglob('*')) if p.is_file()]
|
|
return result
|
|
if __name__=='__main__':
|
|
request=json.loads(pathlib.Path(sys.argv[1]).read_text())
|
|
try:result=run(request)
|
|
except Exception as error:
|
|
pathlib.Path(request['result_file']).write_text(json.dumps({'protocol':1,'error':str(error)}));raise
|
|
pathlib.Path(request['result_file']).write_text(json.dumps(result))
|