Init
This commit is contained in:
Dejvino
Executable
+338
@@ -0,0 +1,338 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Long-running checksum cracker for Lofi-12 XT cmtd+0x08.
|
||||
|
||||
Strategy: 3 known (image, checksum) pairs let us test each hypothesis fast
|
||||
with early exit, plus detect CONSTANT-XOR-masked CRCs (stored = crc ^ mask).
|
||||
|
||||
Phases:
|
||||
A (seconds): zlib-speed hashes (crc32/adler/word-sums/fnv) x spans x field-modes.
|
||||
B (long): generic table-driven CRC32 parameter search
|
||||
(poly x init x xorout x refin x span x field-mode),
|
||||
multiprocessed, checkpointed, resumable.
|
||||
|
||||
Usage (long run):
|
||||
python3 tools/lofi_checksum_crack.py --jobs $(nproc) --out /tmp/opencode/ck-results.txt
|
||||
|
||||
Quick smoke test:
|
||||
python3 tools/lofi_checksum_crack.py --quick-only
|
||||
python3 tools/lofi_checksum_crack.py --jobs 2 --limit 40 --out /tmp/opencode/ck-smoke.txt
|
||||
|
||||
Stdlib only. Safe: read-only on stock .bin files.
|
||||
"""
|
||||
import argparse, binascii, itertools, json, multiprocessing as mp
|
||||
import os, struct, sys, time, zlib
|
||||
|
||||
BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
IMAGES = [
|
||||
('1.1.156', 'Lofi-12XT_v1.1.156/Lofi-12XT_v1.1.156/Lofi-12 XT.bin'),
|
||||
('1.2.179', 'Lofi-12XT_v1.2.179/Lofi-12XT_v1.2.179/Lofi-12 XT.bin'),
|
||||
('1.5.205', 'Lofi-12XT_v1.5.205/Lofi-12XT_v1.5.205/Lofi-12 XT.bin'),
|
||||
]
|
||||
|
||||
POLYS = [ # normal (non-reflected) form
|
||||
0x04C11DB7, # IEEE / PKZIP
|
||||
0x1EDC6F41, # CRC-32C (Castagnoli)
|
||||
0x741B8CD7, # CRC-32K (Koopman)
|
||||
0x1A2B3E55, # spare / nonstandard probes
|
||||
0x814141AB, # CRC-32Q
|
||||
0x000000AF, # tiny-poly probe (catches nibble-CRC schemes fast-fail)
|
||||
]
|
||||
INITS = [0x00000000, 0xFFFFFFFF]
|
||||
XOROUTS = [0x00000000, 0xFFFFFFFF]
|
||||
REFINS = [True, False]
|
||||
FIELDMODES = ['asis', 'zeroed', 'ff'] # how cmtd+0x08 bytes are treated during hashing
|
||||
SPANS = ['full', 'from_0x0C', 'from_0x30', 'from_0x54', 'payload_concat',
|
||||
'headers_only', 'sect_headers_mixed']
|
||||
|
||||
def load_images():
|
||||
blobs = []
|
||||
for ver, rel in IMAGES:
|
||||
p = os.path.join(BASE, rel)
|
||||
d = open(p, 'rb').read()
|
||||
assert d[:4] == b'cmtd' and d[0x30:0x34] == b'mmtd', p
|
||||
ck = struct.unpack('<I', d[8:12])[0]
|
||||
blobs.append((ver, d, ck))
|
||||
return blobs
|
||||
|
||||
def span_bufs(d: bytes):
|
||||
nsect = struct.unpack('<I', d[0x50:0x54])[0]
|
||||
off = 0x54
|
||||
pay = bytearray()
|
||||
hdrs = bytearray()
|
||||
for _ in range(nsect):
|
||||
assert d[off:off+4] == b'sect'
|
||||
ln = struct.unpack('<I', d[off+8:off+12])[0]
|
||||
hdrs += d[off:off+12]
|
||||
pay += d[off+12:off+12+ln]
|
||||
off += 12 + ln
|
||||
z = bytearray(d); z[8:12] = b'\0\0\0\0'
|
||||
f = bytearray(d); f[8:12] = b'\xff\xff\xff\xff'
|
||||
return {
|
||||
'full': [d, bytes(z), bytes(f)],
|
||||
'from_0x0C': [d[0x0C:], bytes(z)[0x0C:], bytes(f)[0x0C:]],
|
||||
'from_0x30': [d[0x30:], d[0x30:], d[0x30:]],
|
||||
'from_0x54': [d[0x54:], d[0x54:], d[0x54:]],
|
||||
'payload_concat': [bytes(pay), bytes(pay), bytes(pay)],
|
||||
'headers_only': [d[:0x54], bytes(z)[:0x54], bytes(f)[:0x54]],
|
||||
'sect_headers_mixed': [bytes(hdrs), bytes(hdrs), bytes(hdrs)],
|
||||
}
|
||||
_FIELDMODE_IDX = {'asis': 0, 'zeroed': 1, 'ff': 2}
|
||||
|
||||
# ---------- generic CRC32 (table-driven, pure python) ----------
|
||||
_crc_tables = {}
|
||||
def crc_table(poly, refin):
|
||||
key = (poly, refin)
|
||||
t = _crc_tables.get(key)
|
||||
if t is not None:
|
||||
return t
|
||||
t = []
|
||||
if refin:
|
||||
rpoly = int(f'{poly:032b}'[::-1], 2)
|
||||
for i in range(256):
|
||||
c = i
|
||||
for _ in range(8):
|
||||
c = (c >> 1) ^ rpoly if c & 1 else c >> 1
|
||||
t.append(c & 0xFFFFFFFF)
|
||||
else:
|
||||
for i in range(256):
|
||||
c = i << 24
|
||||
for _ in range(8):
|
||||
c = ((c << 1) ^ poly) & 0xFFFFFFFF if c & 0x80000000 else (c << 1) & 0xFFFFFFFF
|
||||
t.append(c)
|
||||
_crc_tables[key] = t
|
||||
return t
|
||||
|
||||
def crc_generic(buf: bytes, poly, init, refin, xorout):
|
||||
tab = crc_table(poly, refin)
|
||||
crc = init
|
||||
if refin:
|
||||
for b in buf:
|
||||
crc = tab[(crc ^ b) & 0xFF] ^ (crc >> 8)
|
||||
else:
|
||||
for b in buf:
|
||||
crc = tab[((crc >> 24) ^ b) & 0xFF] ^ ((crc << 8) & 0xFFFFFFFF)
|
||||
return (crc ^ xorout) & 0xFFFFFFFF
|
||||
|
||||
def fnv1a32(buf: bytes):
|
||||
h = 0x811C9DC5
|
||||
for b in buf:
|
||||
h = ((h ^ b) * 0x01000193) & 0xFFFFFFFF
|
||||
return h
|
||||
|
||||
def wordsum_le(buf: bytes):
|
||||
s = 0
|
||||
for i in range(0, len(buf) - 3, 4):
|
||||
(w,) = struct.unpack('<I', buf[i:i+4])
|
||||
s = (s + w) & 0xFFFFFFFF
|
||||
return s
|
||||
|
||||
# ---------- phase A: fast hashes ----------
|
||||
def phase_a(blobs):
|
||||
print('=== Phase A: fast hashes (crc32/adler/fnv/wordsum) ===', flush=True)
|
||||
cands = []
|
||||
for span in SPANS:
|
||||
bufs = [(ver, span_bufs(d)[span], ck) for ver, d, ck in blobs]
|
||||
tests = {
|
||||
'crc32': lambda b: binascii.crc32(b) & 0xFFFFFFFF,
|
||||
'crc32_bswap': lambda b: struct.unpack('>I', struct.pack('<I', binascii.crc32(b) & 0xFFFFFFFF))[0],
|
||||
'adler32': lambda b: zlib.adler32(b) & 0xFFFFFFFF,
|
||||
'fnv1a32': fnv1a32,
|
||||
'wordsum_le': wordsum_le,
|
||||
'sum_bytes': lambda b: sum(b) & 0xFFFFFFFF,
|
||||
}
|
||||
for aname, fn in tests.items():
|
||||
for fmode in _FIELDMODE_IDX:
|
||||
idx = _FIELDMODE_IDX[fmode]
|
||||
try:
|
||||
vals = [(ver, ck, fn(b[idx])) for ver, b, ck in bufs]
|
||||
except Exception as e:
|
||||
print(f' {aname} x {span} x {fmode}: error {e}')
|
||||
continue
|
||||
if all(v == ck for _, ck, v in vals):
|
||||
print(f' *** EXACT MATCH: {aname} x {span} x {fmode}')
|
||||
cands.append((aname, span, fmode, 0))
|
||||
masks = [ck ^ v for _, ck, v in vals]
|
||||
if masks[0] == masks[1] == masks[2]:
|
||||
print(f' --- xor-mask candidate: {aname} x {span} x {fmode} '
|
||||
f'mask={masks[0]:08X} (masked CRC, needs 1 confirmation)')
|
||||
cands.append((aname, span, fmode, masks[0]))
|
||||
if not cands:
|
||||
print('Phase A: no exact or xor-mask candidates.', flush=True)
|
||||
return cands
|
||||
|
||||
# ---------- phase B: generic CRC search ----------
|
||||
def all_params():
|
||||
for poly, init, xorout, refin, span, fmode in itertools.product(
|
||||
POLYS, INITS, XOROUTS, REFINS, SPANS, FIELDMODES):
|
||||
yield (poly, init, xorout, refin, span, fmode)
|
||||
|
||||
_G_BLOBS = None
|
||||
def _init_worker(blobs_packed):
|
||||
global _G_BLOBS
|
||||
_G_BLOBS = blobs_packed # list of (ver, span->bufs, ck); pickled once per worker
|
||||
|
||||
def _worker(param):
|
||||
poly, init, xorout, refin, span, fmode = param
|
||||
idx = _FIELDMODE_IDX[fmode]
|
||||
try:
|
||||
r = []
|
||||
for ver, sbufs, ck in _G_BLOBS:
|
||||
v = crc_generic(sbufs[span][idx], poly, init, refin, xorout)
|
||||
r.append((ver, ck, v))
|
||||
if r[0][2] == r[0][1] and r[1][2] == r[1][1] and r[2][2] == r[2][1]:
|
||||
return ('EXACT', param, 0)
|
||||
m0, m1, m2 = r[0][1] ^ r[0][2], r[1][1] ^ r[1][2], r[2][1] ^ r[2][2]
|
||||
if m0 == m1 == m2:
|
||||
return ('MASK', param, m0)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def phase_b(blobs, jobs, out, limit=None, resume_from=0):
|
||||
params = list(all_params())
|
||||
if limit:
|
||||
params = params[:limit]
|
||||
total = len(params)
|
||||
print(f'=== Phase B: generic CRC search: {total} combos, jobs={jobs} ===', flush=True)
|
||||
# pack span buffers once (pickle to workers a single time)
|
||||
packed = [(ver, span_bufs(d), ck) for ver, d, ck in blobs]
|
||||
done = resume_from
|
||||
t0 = time.time()
|
||||
found = []
|
||||
with mp.Pool(jobs, initializer=_init_worker, initargs=(packed,)) as pool:
|
||||
CH = 8
|
||||
for i, res in enumerate(pool.imap_unordered(_worker, params, chunksize=CH), start=1):
|
||||
if i <= done:
|
||||
continue
|
||||
if res is not None:
|
||||
kind, param, mask = res
|
||||
poly, init, xorout, refin, span, fmode = param
|
||||
line = (f'{kind} poly={poly:08X} init={init:08X} xorout={xorout:08X} '
|
||||
f'refin={int(refin)} span={span} field={fmode} mask={mask:08X}')
|
||||
print(f' *** {line}', flush=True)
|
||||
found.append(line)
|
||||
with open(out, 'a') as fh:
|
||||
fh.write(line + '\n')
|
||||
if i % 50 == 0 or i == total:
|
||||
el = time.time() - t0
|
||||
rate = i / max(el, 1e-6)
|
||||
print(f' [{i}/{total}] {rate:.1f} combos/s elapsed={el:.0f}s', flush=True)
|
||||
with open(out + '.progress', 'w') as fh:
|
||||
fh.write(json.dumps({'done': i, 'total': total,
|
||||
'elapsed': el, 'found': found}) + '\n')
|
||||
el = time.time() - t0
|
||||
print(f'Phase B done: {total} combos in {el:.0f}s, {len(found)} candidates.', flush=True)
|
||||
return found
|
||||
|
||||
_G_SEED_BUFS = None
|
||||
_G_SEED_CKS = None
|
||||
|
||||
def _init_seed_worker(bufs, cks):
|
||||
global _G_SEED_BUFS, _G_SEED_CKS
|
||||
_G_SEED_BUFS = bufs
|
||||
_G_SEED_CKS = cks
|
||||
|
||||
def _seed_scan(task):
|
||||
lo, hi = task
|
||||
b1, b2, b3 = _G_SEED_BUFS
|
||||
s1, s2, s3 = _G_SEED_CKS
|
||||
hits = []
|
||||
for seed in range(lo, hi):
|
||||
if (binascii.crc32(b1, seed) & 0xFFFFFFFF) == s1:
|
||||
if ((binascii.crc32(b2, seed) & 0xFFFFFFFF) == s2 and
|
||||
(binascii.crc32(b3, seed) & 0xFFFFFFFF) == s3):
|
||||
hits.append(seed)
|
||||
return (lo, hi, hits)
|
||||
|
||||
def _seed_worker(args):
|
||||
lo, hi, span, fmode = args
|
||||
idx = _FIELDMODE_IDX[fmode]
|
||||
b1, b2, b3 = _G_SEED[0][span][idx], _G_SEED[1][span][idx], _G_SEED[2][span][idx]
|
||||
s1, s2, s3 = _G_SEED[3]
|
||||
hits = []
|
||||
for seed in range(lo, hi):
|
||||
c1 = binascii.crc32(b1, seed) & 0xFFFFFFFF
|
||||
if c1 != s1:
|
||||
# xor-mask path: derive mask from image 1, confirm on 2+3
|
||||
# (costs 2 more CRCs only on the rare near-hit; here c1!=s1 always
|
||||
# so check mask constancy cheaply only every step? skip: exact-only
|
||||
# in seed phase for speed; mask search lives in Phase B)
|
||||
continue
|
||||
c2 = binascii.crc32(b2, seed) & 0xFFFFFFFF
|
||||
c3 = binascii.crc32(b3, seed) & 0xFFFFFFFF
|
||||
if c2 == s2 and c3 == s3:
|
||||
hits.append(f'EXACT seed={seed:08X} span={span} field={fmode}')
|
||||
return hits
|
||||
|
||||
def phase_c(blobs, jobs, out, seed_max=1 << 24, seed_span='full', seed_chunk=4096):
|
||||
print(f'=== Phase C: seeded CRC32-IEEE brute force: seeds 0..{seed_max} '
|
||||
f'span={seed_span} jobs={jobs} ===', flush=True)
|
||||
print(' (C-speed crc32; exact-match only; this is the hours-long phase)', flush=True)
|
||||
# NOTE: 'zeroed' = cmtd+0x08 treated as 00s during hashing (standard scheme);
|
||||
# 'asis' re-check is cheap relative to seed space, so do zeroed first.
|
||||
t0 = time.time()
|
||||
found = []
|
||||
for fmode in ('zeroed', 'asis'):
|
||||
idx = _FIELDMODE_IDX[fmode]
|
||||
bufs = [span_bufs(d)[seed_span][idx] for _, d, _ in blobs]
|
||||
cks = [ck for _, _, ck in blobs]
|
||||
found += _phase_c_loop(bufs, cks, seed_span, fmode, seed_max, seed_chunk, jobs, out, t0)
|
||||
return found
|
||||
|
||||
def _phase_c_loop(bufs, cks, span, fmode, seed_max, chunk, jobs, out, t0):
|
||||
b1, b2, b3 = bufs
|
||||
s1, s2, s3 = cks
|
||||
found = []
|
||||
# shard seed space across workers: each task scans [lo,hi)
|
||||
tasks = [(lo, min(lo + chunk, seed_max)) for lo in range(0, seed_max, chunk)]
|
||||
total = len(tasks)
|
||||
done = 0
|
||||
with mp.Pool(jobs, initializer=_init_seed_worker, initargs=(bufs, cks)) as pool:
|
||||
for lo, hi, hits in pool.imap_unordered(_seed_scan, tasks, chunksize=4):
|
||||
done += 1
|
||||
for seed in hits:
|
||||
line = f'EXACT seed={seed:08X} span={span} field={fmode}'
|
||||
print(f' *** {line}', flush=True)
|
||||
found.append(line)
|
||||
with open(out, 'a') as fh:
|
||||
fh.write(line + '\n')
|
||||
if done % max(1, total // 20) == 0 or done == total:
|
||||
el = time.time() - t0
|
||||
print(f' Phase C [{done}/{total} chunks] seeds~{done*chunk}/{seed_max} '
|
||||
f'elapsed={el:.0f}s', flush=True)
|
||||
el = time.time() - t0
|
||||
print(f'Phase C done: {seed_max} seeds in {el:.0f}s, {len(found)} hits.', flush=True)
|
||||
return found
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('--jobs', type=int, default=max(1, (os.cpu_count() or 4) - 1))
|
||||
ap.add_argument('--out', default='/tmp/opencode/ck-results.txt')
|
||||
ap.add_argument('--quick-only', action='store_true')
|
||||
ap.add_argument('--skip-phase-a', action='store_true')
|
||||
ap.add_argument('--skip-phase-b', action='store_true')
|
||||
ap.add_argument('--skip-phase-c', action='store_true')
|
||||
ap.add_argument('--limit', type=int, default=None, help='test only first N combos (smoke test)')
|
||||
ap.add_argument('--resume-from', type=int, default=0)
|
||||
ap.add_argument('--seed-max', type=int, default=1 << 24,
|
||||
help='seed brute-force range 0..SEED_MAX (default 2^24 ≈ hours)')
|
||||
ap.add_argument('--seed-span', default='full', choices=SPANS)
|
||||
ap.add_argument('--seed-chunk', type=int, default=4096)
|
||||
a = ap.parse_args()
|
||||
blobs = load_images()
|
||||
for ver, d, ck in blobs:
|
||||
print(f'{ver}: size={len(d)} stored_ck={ck:08X}', flush=True)
|
||||
if not a.skip_phase_a:
|
||||
phase_a(blobs)
|
||||
if a.quick_only:
|
||||
print('quick-only: stopping before Phase B/C.', flush=True)
|
||||
return
|
||||
open(a.out, 'a').write(f'# run {time.ctime()} jobs={a.jobs} limit={a.limit} seed_max={a.seed_max}\n')
|
||||
if not a.skip_phase_b:
|
||||
phase_b(blobs, a.jobs, a.out, limit=a.limit, resume_from=a.resume_from)
|
||||
if not a.skip_phase_c and not a.limit:
|
||||
phase_c(blobs, a.jobs, a.out, seed_max=a.seed_max,
|
||||
seed_span=a.seed_span, seed_chunk=a.seed_chunk)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user