"""Read-only code/blueprint resource search with bounded local decompression."""
import concurrent.futures
import json
import re
import struct
import zlib
from collections import Counter
from pathlib import Path
from cryptography.hazmat.primitives.ciphers import Cipher,algorithms,modes
root=Path(r'D:/SteamLibrary/steamapps/common/Elite Dangerous/Products/elite-dangerous-odyssey-64/Win64')
key=json.loads(Path('artifacts/ship-model-reference/local-asset-public-key.json').read_text())
n,e=int(key['n']),key['e']
candidate=re.compile(rb'code|blueprint|component|uidata|gamedata',re.I)
pattern=re.compile(rb'GalaxyMapVisualisation|GalaxyMapView|tradeData|noData|unclaimed|alreadyPopulated|localPlayerClaimed|starclass',re.I)
out=Path('artifacts/galaxy-query-research/palette-resources')
def scan(directory):
count=0;types=Counter();names=[];found=[]
for path in directory.iterdir():
if not path.is_file():continue
with path.open('rb') as stream:
header=stream.read(262)
if header[:4]!=b'FREA':continue
count+=1
raw=pow(int.from_bytes(header[6:262],'big'),e,n).to_bytes(256,'big')
separator=raw.find(b'\0',2)
if separator<10:continue
message=raw[separator+1:];metadata=message[36:]
strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',metadata)]
types[strings[1] if len(strings)>1 else '(no type in signed prefix)']+=1
names.append(dict(key=path.name,strings=strings))
if not candidate.search(metadata):continue
encrypted=stream.read(65536);encrypted=encrypted[:len(encrypted)//16*16]
aeskey=bytes(value^path.name.lower().encode()[i%40] for i,value in enumerate(message[4:36]))
decrypt=Cipher(algorithms.AES(aeskey),modes.ECB()).decryptor()
prefix=(metadata+decrypt.update(encrypted)+decrypt.finalize())[:struct.unpack_from('<I',message)[0]]
blocks=[prefix];descriptions=[]
for match in re.finditer(rb'\x78[\x01\x9c\xda]',prefix):
try:body=zlib.decompressobj().decompress(prefix[match.start():],262144)
except zlib.error:continue
if len(body)<16:continue
blocks.append(body);descriptions.append(dict(offset=match.start(),bytes=len(body)))
if len(blocks)>=5:break
hits=[]
for index,body in enumerate(blocks):
matches=[dict(offset=match.start(),term=match.group().decode()) for match in pattern.finditer(body)]
if matches:
(out/(path.name+f'.search{index}')).write_bytes(body)
hits.append(dict(block=index,matches=matches,strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',body)][:300]))
if hits:found.append(dict(path=str(path),strings=strings,blocks=descriptions,hits=hits))
return count,types,names,found
counts=Counter();names=[];rows=[];count=0
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
for observed,types,metadata,found in pool.map(scan,sorted(path for path in root.iterdir() if path.is_dir())):
count+=observed;counts.update(types);names.extend(metadata);rows.extend(found)
print(f'signed={count} paletteCandidates={len(rows)}',flush=True)
(out/'signed-metadata.json').write_text(json.dumps(dict(scope='Original signed header names; no game access',count=count,types=dict(counts),headers=names),indent=2),encoding='utf-8')
(out/'signed-palette-search.json').write_text(json.dumps(dict(scope='Original code/blueprint prefixes up to64KiB, inflated blocks up to256KiB',count=count,types=dict(counts),resources=rows),indent=2),encoding='utf-8')
print(json.dumps(dict(count=count,matches=len(rows),types=counts.most_common(25))))
"""Read-only code/blueprint resource search with bounded local decompression."""
import concurrent.futures
import json
import re
import struct
import zlib
from collections import Counter
from pathlib import Path
from cryptography.hazmat.primitives.ciphers import Cipher,algorithms,modes
root=Path(r'D:/SteamLibrary/steamapps/common/Elite Dangerous/Products/elite-dangerous-odyssey-64/Win64')
key=json.loads(Path('artifacts/ship-model-reference/local-asset-public-key.json').read_text())
n,e=int(key['n']),key['e']
candidate=re.compile(rb'code|blueprint|component|uidata|gamedata',re.I)
pattern=re.compile(rb'GalaxyMapVisualisation|GalaxyMapView|tradeData|noData|unclaimed|alreadyPopulated|localPlayerClaimed|starclass',re.I)
out=Path('artifacts/galaxy-query-research/palette-resources')
def scan(directory):
count=0;types=Counter();names=[];found=[]
for path in directory.iterdir():
if not path.is_file():continue
with path.open('rb') as stream:
header=stream.read(262)
if header[:4]!=b'FREA':continue
count+=1
raw=pow(int.from_bytes(header[6:262],'big'),e,n).to_bytes(256,'big')
separator=raw.find(b'\0',2)
if separator<10:continue
message=raw[separator+1:];metadata=message[36:]
strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',metadata)]
types[strings[1] if len(strings)>1 else '(no type in signed prefix)']+=1
names.append(dict(key=path.name,strings=strings))
if not candidate.search(metadata):continue
encrypted=stream.read(65536);encrypted=encrypted[:len(encrypted)//16*16]
aeskey=bytes(value^path.name.lower().encode()[i%40] for i,value in enumerate(message[4:36]))
decrypt=Cipher(algorithms.AES(aeskey),modes.ECB()).decryptor()
prefix=(metadata+decrypt.update(encrypted)+decrypt.finalize())[:struct.unpack_from('<I',message)[0]]
blocks=[prefix];descriptions=[]
for match in re.finditer(rb'\x78[\x01\x9c\xda]',prefix):
try:body=zlib.decompressobj().decompress(prefix[match.start():],262144)
except zlib.error:continue
if len(body)<16:continue
blocks.append(body);descriptions.append(dict(offset=match.start(),bytes=len(body)))
if len(blocks)>=5:break
hits=[]
for index,body in enumerate(blocks):
matches=[dict(offset=match.start(),term=match.group().decode()) for match in pattern.finditer(body)]
if matches:
(out/(path.name+f'.search{index}')).write_bytes(body)
hits.append(dict(block=index,matches=matches,strings=[m.group().decode('ascii') for m in re.finditer(rb'[ -~]{4,}',body)][:300]))
if hits:found.append(dict(path=str(path),strings=strings,blocks=descriptions,hits=hits))
return count,types,names,found
counts=Counter();names=[];rows=[];count=0
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
for observed,types,metadata,found in pool.map(scan,sorted(path for path in root.iterdir() if path.is_dir())):
count+=observed;counts.update(types);names.extend(metadata);rows.extend(found)
print(f'signed={count} paletteCandidates={len(rows)}',flush=True)
(out/'signed-metadata.json').write_text(json.dumps(dict(scope='Original signed header names; no game access',count=count,types=dict(counts),headers=names),indent=2),encoding='utf-8')
(out/'signed-palette-search.json').write_text(json.dumps(dict(scope='Original code/blueprint prefixes up to64KiB, inflated blocks up to256KiB',count=count,types=dict(counts),resources=rows),indent=2),encoding='utf-8')
print(json.dumps(dict(count=count,matches=len(rows),types=counts.most_common(25))))
