Added the CDJ-900NXS firmware patcher.

Takes the official C900NXS.UPD (1.31, checked by SHA-256), lets you choose
mods in a list (arrows, space, enter; numbers where the terminal cannot pass
keys), patches the application with the emulator checkout's
mods/patch_cdj900nxs.py, packs it and writes a new C900NXS.UPD labelled 1.32.

Only the application is replaced: the boot loader and recovery records,
MAIN's S0/S7 records, DRIV and PANL are copied byte for byte and checked again
on the finished file, which is also unpacked the boot loader's way and
compared with the patched application. --stock differs from the official file
in the version label and its CRC only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Thiemo TizianiandClaude Opus 5.5 committed 2026-10-17 19:05:49 +02:00
1 parent e8134e446c
commit cfa0e85854
5 files changed
+730

No files matched your search

+4
View File
@@ -0,0 +1,4 @@
out/
*.UPD
__pycache__/
.pytest_cache/
+72
View File
@@ -0,0 +1,72 @@
# CDJ-900NXS firmware patcher
Builds a new `C900NXS.UPD` from the official CDJ-900NXS firmware 1.31 with
the mods you choose, for the player's own USB update.
```sh
python cdj900nxs_patcher.py C900NXS.UPD # choose in a list
python cdj900nxs_patcher.py C900NXS.UPD --mods beatjump,cuecolours,hotcuekeys
python cdj900nxs_patcher.py C900NXS.UPD --stock # the original, to go back
python cdj900nxs_patcher.py --list
```
In the list: **arrow keys** move, **space** ticks or unticks, **a** ticks all
or none, **enter** builds, **q** quits. Terminals that cannot pass single
keys (Git Bash's mintty, a pipe) get a numbered list instead. The result goes
to `out/<date>-<mods>/C900NXS.UPD` with a `report.txt` of every check.
## What it guarantees
- **Only the original file is accepted**: SHA-256
`e96bca7d1ac983985992a8e5fa7813e1ebf04bef514e0c6cfd0a8375f667f7f9`, and the
1.31 application inside it is checked too.
- **Only the main application is replaced.** The boot loader and recovery
loader (MAIN's records below flash 0x40000), MAIN's S0 header and S7 entry
records, and the DRIV and PANL sections are copied from the original byte
for byte. Before anything is written the finished file is read back and
checked: every section CRC and S-record checksum; DRIV and PANL equal to the
original; MAIN's header equal but for its version label, application-only
mode kept; every boot loader and recovery record and byte equal; no record
past 0x7E0000; the packed application's checksum; and the application,
unpacked the way the boot loader does it, equal to the patched one.
- **Labelled 1.32**, so that the player's updater does not skip the update as
the version it already has. The application still reports 1.31 (as
arnoldasr's builds do). `--stock` gives the original firmware with only that
label changed: three bytes differ from the official file (the digit and the
section CRC), exactly as arnoldasr's back-to-stock file, which installed on
his test player.
Checked in the emulator: both the `--stock` file and one with every mod were
written into an emulated flash and booted through Pioneer's own boot loader,
which unpacked and started them.
## Mods
From the emulator checkout `../CDJ900NXSVirtual` (`--repo` for another place):
`mods/patch_cdj900nxs.py` and the mods next to it.
| mod | what it does |
|---|---|
| `code` | arnoldasr: BPM with two decimals, two-line phase meter, RGB waveform |
| `utility` | arnoldasr: UTILITY > PHASE METER switch (needs `code`) |
| `overview` | arnoldasr: overview waveform colour gradient |
| `beatjump` | TRACK SEARCH jumps 1/2/4/8 beats, UTILITY > BEAT JUMP |
| `cuecolours` | memory cues in their rekordbox colours, hot cues shown |
| `hotcuekeys` | UTILITY > HOT CUE KEYS; QUANTIZE short = hot key mode |
arnoldasr's mods need his repository checked out in
`../CDJ900NXSVirtual/mods/cdj900nxs/upstream` (it has no licence, so it is
not copied here); without it they are greyed out.
## Risk
Nothing here talks to a player; you copy the file to a FAT32 USB stick and run
the update yourself. Our mods have only run in the emulator. Recovery from an
application that crashes at start-up has not been tested by anyone. Keep the
`--stock` file at hand.
## Tests
```sh
CDJ900NXS_UPD=/path/to/C900NXS.UPD python -m pytest test_patcher.py
```
+280
View File
@@ -0,0 +1,280 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
"""CDJ-900NXS firmware patcher: the official 1.31 update with the mods you
choose, as a new C900NXS.UPD for the player's own USB update.
python cdj900nxs_patcher.py C900NXS.UPD # choose in a list, write out/...
python cdj900nxs_patcher.py C900NXS.UPD --mods beatjump,cuecolours
python cdj900nxs_patcher.py C900NXS.UPD --stock # unmodified, to go back
python cdj900nxs_patcher.py --list
* Only the original update is accepted (its SHA-256 is checked).
* Only the main application is replaced. The boot loader and recovery
loader (flash 0x0-0x3FFFF), the S-record header and entry, and the DRIV
and PANL sections are copied from the original byte for byte, and checked
again on the finished file before it is written.
* The new file says 1.32 in its MAIN header so that the player's updater
does not skip it as the version it already has; the application itself
still reports 1.31 (as arnoldasr's builds do). Only that label differs.
* Nothing here talks to a player. You copy the file to a USB stick and run
the update on the player yourself, at your own risk: nothing of this has
been tested on hardware except the relabelled stock update (by arnoldasr),
and recovery from an application that crashes at start-up is untested.
The mods come from the CDJ900NXSVirtual checkout next to this folder
(--repo): mods/patch_cdj900nxs.py, which also needs arnoldasr's checkout in
mods/cdj900nxs/upstream for his mods (code, utility, overview).
"""
import argparse
import os
import sys
import time
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import upd # noqa: E402
DEFAULT_REPO = os.path.join(os.path.dirname(HERE), 'CDJ900NXSVirtual')
UPSTREAM_MODS = ('code', 'utility', 'overview')
NEEDS = {'utility': 'code'} # the PHASE METER switch chooses the code mod's meter
LABEL = '1.32'
def load_repo(repo):
"""patch_cdj900nxs and the LZSS packer from the emulator checkout."""
mods_dir = os.path.join(repo, 'mods')
fw_dir = os.path.join(repo, 'scripts', 'firmware')
if not os.path.isfile(os.path.join(mods_dir, 'patch_cdj900nxs.py')):
raise SystemExit('error: no mods/patch_cdj900nxs.py in %s (use --repo)' % repo)
for d in (mods_dir, fw_dir):
if d not in sys.path:
sys.path.insert(0, d)
import patch_cdj900nxs
import lzss_encode
return patch_cdj900nxs, lzss_encode
def available(patcher, repo):
"""(name, description, usable, why not) for every mod."""
upstream = os.path.isdir(os.environ.get('CDJ900NXS_MODS_SRC')
or os.path.join(repo, 'mods', 'cdj900nxs', 'upstream'))
out = []
for name, what in patcher.MODS.items():
ok = upstream or name not in UPSTREAM_MODS
out.append((name, what, ok, '' if ok else "needs arnoldasr's checkout in mods/cdj900nxs/upstream"))
return out
# --- choosing ------------------------------------------------------------------
def _keys():
"""A function returning 'up', 'down', 'space', 'enter', 'all', 'quit' per
key press, or None when the terminal cannot be read key by key."""
if not sys.stdin.isatty():
return None
if os.name == 'nt':
import msvcrt
def read():
c = msvcrt.getwch()
if c in ('\x00', '\xe0'):
return {'H': 'up', 'P': 'down'}.get(msvcrt.getwch())
return {' ': 'space', '\r': 'enter', 'a': 'all', 'q': 'quit', '\x1b': 'quit',
'\x03': 'quit'}.get(c.lower())
return read
import termios
import tty
def read():
fd = sys.stdin.fileno()
old = termios.tcgetattr(fd)
try:
tty.setraw(fd)
c = sys.stdin.read(1)
if c == '\x1b':
if sys.stdin.read(1) == '[':
return {'A': 'up', 'B': 'down'}.get(sys.stdin.read(1))
return 'quit'
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old)
return {' ': 'space', '\r': 'enter', '\n': 'enter', 'a': 'all', 'q': 'quit',
'\x03': 'quit'}.get(c.lower())
return read
def _fix_needs(chosen, mods):
for name, need in NEEDS.items():
if name in chosen and need not in chosen and any(m[0] == need and m[2] for m in mods):
chosen.add(need)
def choose(mods):
"""The interactive list: arrows move, space ticks, enter goes on."""
read = _keys()
if read is None:
return choose_by_number(mods)
if os.name == 'nt':
os.system('') # VT sequences on in the Windows console
chosen = {m[0] for m in mods if m[2]}
at = 0
lines = len(mods) + 3
def draw(first):
if not first:
sys.stdout.write('\x1b[%dA' % lines)
sys.stdout.write('\x1b[2K Mods (arrows move, space ticks, a = all/none, enter goes on, q quits)\n')
for i, (name, what, ok, why) in enumerate(mods):
mark = '[x]' if name in chosen else ('[ ]' if ok else ' - ')
cursor = '>' if i == at else ' '
text = '%s %s %-11s %s' % (cursor, mark, name, what if ok else why)
sys.stdout.write('\x1b[2K' + ('\x1b[7m%s\x1b[0m' % text if i == at else text)[:200] + '\n')
need = ', '.join('%s needs %s' % kv for kv in NEEDS.items())
sys.stdout.write('\x1b[2K (%s)\n\x1b[2K\n' % need)
sys.stdout.flush()
draw(True)
while True:
k = read()
if k == 'up':
at = (at - 1) % len(mods)
elif k == 'down':
at = (at + 1) % len(mods)
elif k == 'space' and mods[at][2]:
chosen ^= {mods[at][0]}
_fix_needs(chosen, mods)
elif k == 'all':
usable = {m[0] for m in mods if m[2]}
chosen = set() if chosen == usable else usable
elif k == 'enter':
break
elif k == 'quit':
raise SystemExit('nothing written')
draw(False)
return [m[0] for m in mods if m[0] in chosen]
def choose_by_number(mods):
"""For terminals that cannot be read key by key (mintty, pipes)."""
chosen = {m[0] for m in mods if m[2]}
while True:
print('\n Mods (all usable ones are ticked):')
for i, (name, what, ok, why) in enumerate(mods, 1):
mark = '[x]' if name in chosen else ('[ ]' if ok else ' - ')
print(' %d %s %-11s %s' % (i, mark, name, what if ok else why))
answer = input(' Numbers to tick/untick (e.g. 1 3), enter to go on, q to quit: ').strip()
if not answer:
break
if answer.lower() == 'q':
raise SystemExit('nothing written')
for tok in answer.replace(',', ' ').split():
if tok.isdigit() and 1 <= int(tok) <= len(mods) and mods[int(tok) - 1][2]:
chosen ^= {mods[int(tok) - 1][0]}
_fix_needs(chosen, mods)
return [m[0] for m in mods if m[0] in chosen]
# --- building ------------------------------------------------------------------
def build(stock, names, patcher, lzss, label=LABEL, log=print):
"""(update bytes, application, list of checks)."""
if names:
try:
app = patcher.patch(stock.application, names)
except SystemExit as e:
raise upd.UpdError(str(e))
log('packing the application (LZSS, about a minute and a half)...')
t = time.time()
stream = lzss.pack(app)
log(' %d -> %d bytes in %.0f s' % (len(app), len(stream), time.time() - t))
else:
import struct
app = stock.application
size = struct.unpack_from('<I', stock.flash, upd.APP_START)[0]
stream = stock.flash[upd.APP_START + 4:upd.APP_START + 4 + size]
raw = upd.build(stock, app, stream, label)
checks = upd.verify(raw, stock, app, label)
return raw, app, checks
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument('stock', nargs='?', help='the official C900NXS.UPD (firmware 1.31)')
ap.add_argument('--out', help='new folder for the built C900NXS.UPD (default out/<date>-<mods>)')
ap.add_argument('--mods', help='comma-separated mods instead of the list')
ap.add_argument('--stock', dest='back', action='store_true',
help='no mods: the original application labelled %s, to go back to stock' % LABEL)
ap.add_argument('--repo', default=DEFAULT_REPO, help='the CDJ900NXSVirtual checkout (default %(default)s)')
ap.add_argument('--list', action='store_true', help='list the mods and exit')
args = ap.parse_args(argv)
patcher, lzss = load_repo(args.repo)
mods = available(patcher, args.repo)
if args.list:
for name, what, ok, why in mods:
print('%-11s %s%s' % (name, what, '' if ok else ' (%s)' % why))
return 0
if not args.stock:
ap.error('give the official C900NXS.UPD')
print('CDJ-900NXS firmware patcher')
raw = open(args.stock, 'rb').read()
try:
stock = upd.Stock(raw)
except upd.UpdError as e:
raise SystemExit('error: %s' % e)
print(' %s: the original 1.31 update (SHA-256 %s...), checked' % (os.path.basename(args.stock),
upd.STOCK_UPD_SHA256[:16]))
print(' boot loader and recovery: %d records below 0x%X, kept as they are'
% (len(stock.boot_records()), upd.APP_START))
if args.back:
names = []
elif args.mods is not None:
names = [n for n in args.mods.replace(' ', ',').split(',') if n]
unknown = [n for n in names if n not in patcher.MODS]
if unknown:
raise SystemExit('error: unknown mod(s) %s; --list shows them' % ', '.join(unknown))
blocked = [m[0] for m in mods if m[0] in names and not m[2]]
if blocked:
raise SystemExit('error: %s: %s' % (', '.join(blocked), mods[[m[0] for m in mods].index(blocked[0])][3]))
for name, need in NEEDS.items():
if name in names and need not in names:
raise SystemExit('error: %s needs %s' % (name, need))
else:
names = choose(mods)
if not names:
print(' no mod chosen: use --stock for the unmodified update')
return 1
print(' mods: %s' % (', '.join(names) or 'none (stock application)'))
try:
out_raw, app, checks = build(stock, names, patcher, lzss)
except upd.UpdError as e:
raise SystemExit('error: %s; nothing written' % e)
folder = args.out or os.path.join(HERE, 'out', time.strftime('%Y%m%d-%H%M%S') + '-' +
('-'.join(names) if names else 'stock'))
if os.path.exists(os.path.join(folder, 'C900NXS.UPD')):
raise SystemExit('error: %s already holds a C900NXS.UPD; nothing written' % folder)
os.makedirs(folder, exist_ok=True)
path = os.path.join(folder, 'C900NXS.UPD')
with open(path, 'wb') as f:
f.write(out_raw)
report = ['CDJ-900NXS firmware patcher', 'source: %s (SHA-256 %s)' % (args.stock, upd.STOCK_UPD_SHA256),
'mods: %s' % (', '.join(names) or 'none'), 'C900NXS.UPD SHA-256 %s' % upd.sha256(out_raw),
'application SHA-256 %s, %d bytes' % (upd.sha256(app), len(app)),
'MAIN label %s (the application still reports 1.31)' % LABEL, 'checks:'] + \
[' - ' + c for c in checks]
with open(os.path.join(folder, 'report.txt'), 'w', encoding='utf-8') as f:
f.write('\n'.join(report) + '\n')
print(' checked:')
for c in checks:
print(' - ' + c)
print(' written: %s' % path)
print(' SHA-256 %s' % upd.sha256(out_raw))
print(' Copy it to the root of a FAT32 USB stick and run the update from the player.')
print(' Untested on hardware: use at your own risk; keep this folder\'s report.txt.')
return 0
if __name__ == '__main__':
sys.exit(main())
+95
View File
@@ -0,0 +1,95 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for the patcher: python -m pytest test_patcher.py
Offline: S-records, the boot loader's LZSS against our packer, the version
label guard. With the official update (CDJ900NXS_UPD, or C900NXS.UPD in this
folder): the stock file rebuilt byte for byte, the relabelled stock differing
in three bytes only, and every tampering with the boot area caught."""
import os
import random
import struct
import sys
import pytest
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import upd # noqa: E402
import cdj900nxs_patcher as cli # noqa: E402
STOCK = os.environ.get('CDJ900NXS_UPD') or os.path.join(HERE, 'C900NXS.UPD')
need_stock = pytest.mark.skipif(not os.path.isfile(STOCK), reason='needs the official C900NXS.UPD')
def test_record_round_trip():
rec = upd.Record(2, 0x040020, bytes(range(32)))
line = rec.encode()
assert line.startswith(b'S2') and line.endswith(b'\r\n')
raw = bytes.fromhex(line[2:-2].decode())
assert raw[0] == len(raw) - 1 and sum(raw) & 0xFF == 0xFF
def test_lzss_matches_the_boot_loader():
_, lzss = cli.load_repo(cli.DEFAULT_REPO)
rnd = random.Random(7)
data = bytes(rnd.choice(b' \0ABAB\xff') for _ in range(20000)) + b' ' * 5000 + bytes(4000)
assert upd.unpack_lzss(lzss.pack(data)) == data
def test_region_checksum_and_padding():
reg = upd.pack_region(b'abc')
assert len(reg) % 32 == 0
size = struct.unpack_from('<I', reg)[0]
assert struct.unpack_from('<H', reg, 4 + size)[0] == sum(reg[:4 + size]) & 0xFFFF
def test_relabel_touches_the_label_only():
h = b'CDJ-900NXS MAINVer1.31\x00 0'
assert upd.relabel(h, '1.32') == b'CDJ-900NXS MAINVer1.32\x00 0'
with pytest.raises(upd.UpdError):
upd.relabel(h, '1.3')
@need_stock
def test_stock_rebuilt_byte_for_byte():
raw = open(STOCK, 'rb').read()
stock = upd.Stock(raw)
size = struct.unpack_from('<I', stock.flash, upd.APP_START)[0]
stream = stock.flash[upd.APP_START + 4:upd.APP_START + 4 + size]
assert upd.build(stock, stock.application, stream, None) == raw
@need_stock
def test_relabelled_stock_differs_in_three_bytes():
raw = open(STOCK, 'rb').read()
stock = upd.Stock(raw)
patcher, lzss = cli.load_repo(cli.DEFAULT_REPO)
out, app, checks = cli.build(stock, [], patcher, lzss)
assert len(out) == len(raw) and sum(a != b for a, b in zip(out, raw)) == 3
assert app == stock.application and any('boot loader' in c for c in checks)
@need_stock
def test_other_files_refused():
raw = bytearray(open(STOCK, 'rb').read())
raw[-3] ^= 1
with pytest.raises(upd.UpdError):
upd.Stock(bytes(raw))
@need_stock
def test_boot_area_tampering_caught():
raw = open(STOCK, 'rb').read()
stock = upd.Stock(raw)
size = struct.unpack_from('<I', stock.flash, upd.APP_START)[0]
stream = stock.flash[upd.APP_START + 4:upd.APP_START + 4 + size]
good = upd.build(stock, stock.application, stream)
# rebuild MAIN with one boot loader byte changed, CRCs made right again
sections = upd.parse(good)
main = sections[1]
recs = list(main.records)
i = next(n for n, r in enumerate(recs) if r.kind == 2 and r.address < upd.APP_START)
recs[i] = upd.Record(2, recs[i].address, bytes([recs[i].data[0] ^ 1]) + recs[i].data[1:])
bad = upd.join([sections[0].blob, upd.encode_section(main.header, recs), sections[2].blob])
with pytest.raises(upd.UpdError, match='boot loader'):
upd.verify(bad, stock, stock.application)
+279
View File
@@ -0,0 +1,279 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""The CDJ-900NXS update file (C900NXS.UPD, firmware 1.31): read, rebuild with
a new MAIN application, and check that nothing else changed.
The file: three decimal section lengths, one per line, then the sections
DRIV, MAIN and PANL. Each section is a 32-byte header ("CDJ-900NXS MAINVer1.31"
...; its last byte selects the updater's mode, "0" = application only), then
Motorola S-records, then a CRC-16/XMODEM of all that, little-endian.
MAIN's S-records fill the player's NOR flash from 0: below 0x40000 the boot
loader and the recovery loader, from 0x40000 the application as one packed
region ([u32 length][LZSS stream][u16 additive checksum of length and stream]).
The boot loader checks that sum and unpacks the region to 0x08000000. The
updater in application-only mode writes 0x40000-0x7E0000 only.
This module never changes the boot loader or recovery records, DRIV or PANL:
build_main() copies them byte for byte and verify() checks it again on the
finished file.
"""
import binascii
import hashlib
import re
import struct
STOCK_UPD_SHA256 = 'e96bca7d1ac983985992a8e5fa7813e1ebf04bef514e0c6cfd0a8375f667f7f9'
STOCK_APP_SHA256 = '1ac3e30cde292865413950fe159663b3958e85939ed0a5d0122f796809a5f8ef'
STOCK_APP_SIZE = 16519056
APP_START = 0x40000 # the application's packed region in flash
APP_END = 0x7E0000 # what the updater may write, end exclusive
RECORD_DATA = 32 # data bytes per new S2 record, as the stock file
ADDRESS_BYTES = {0: 2, 1: 2, 2: 3, 3: 4, 5: 2, 6: 3, 7: 4, 8: 3, 9: 2}
HEADER = re.compile(rb'CDJ-900NXS (DRIV|MAIN|PANL)Ver(\d\.\d\d)')
class UpdError(Exception):
pass
def sha256(data):
return hashlib.sha256(data).hexdigest()
def crc(data):
return binascii.crc_hqx(data, 0)
class Record:
__slots__ = ('kind', 'address', 'data')
def __init__(self, kind, address, data):
self.kind, self.address, self.data = kind, address, bytes(data)
def __eq__(self, other):
return (self.kind, self.address, self.data) == (other.kind, other.address, other.data)
def encode(self):
width = ADDRESS_BYTES[self.kind]
body = bytes([width + len(self.data) + 1]) + self.address.to_bytes(width, 'big') + self.data
body += bytes([~sum(body) & 0xFF])
return b'S%d' % self.kind + body.hex().upper().encode() + b'\r\n'
class Section:
def __init__(self, blob):
self.blob = blob
if len(blob) < 36:
raise UpdError('a section is truncated')
if crc(blob[:-2]) != struct.unpack_from('<H', blob, len(blob) - 2)[0]:
raise UpdError('a section fails its CRC')
self.header = blob[:32]
m = HEADER.match(self.header)
if not m:
raise UpdError('unexpected section header %r' % self.header)
self.name, self.version = m.group(1).decode(), m.group(2).decode()
text = blob[32:-2]
if not text.endswith(b'\r\n'):
raise UpdError('%s: the S-records do not end in a line break' % self.name)
self.records = []
for n, line in enumerate(text[:-2].split(b'\r\n'), 1):
if not re.fullmatch(rb'S[0-9][0-9A-F]+', line):
raise UpdError('%s: S-record %d is malformed' % (self.name, n))
kind = int(line[1:2])
if kind not in ADDRESS_BYTES:
raise UpdError('%s: S-record %d has type %d' % (self.name, n, kind))
raw = bytes.fromhex(line[2:].decode())
width = ADDRESS_BYTES[kind]
if len(raw) != raw[0] + 1 or len(raw) < width + 2 or sum(raw) & 0xFF != 0xFF:
raise UpdError('%s: S-record %d fails its length or checksum' % (self.name, n))
self.records.append(Record(kind, int.from_bytes(raw[1:1 + width], 'big'), raw[1 + width:-1]))
if self.records[0].kind != 0 or self.records[-1].kind not in (7, 8, 9):
raise UpdError('%s: no S-record header or end' % self.name)
def data_records(self):
return [r for r in self.records if r.kind in (1, 2, 3)]
def flash(self, size=APP_END):
"""The bytes the records put into flash below `size` (0xFF elsewhere)."""
image = bytearray(b'\xff' * size)
for r in self.data_records():
if r.address + len(r.data) > size:
raise UpdError('%s: a record reaches past 0x%X' % (self.name, size))
image[r.address:r.address + len(r.data)] = r.data
return bytes(image)
def encode_section(header, records):
body = header + b''.join(r.encode() for r in records)
return body + struct.pack('<H', crc(body))
def parse(raw):
"""The three sections of an update file, every length and checksum checked."""
pos, sizes = 0, []
for _ in range(3):
end = raw.index(b'\r\n', pos)
if not raw[pos:end].isdigit():
raise UpdError('the file does not start with three section lengths')
sizes.append(int(raw[pos:end]))
pos = end + 2
if pos + sum(sizes) != len(raw):
raise UpdError('the section lengths do not add up to the file')
sections = []
for size in sizes:
sections.append(Section(raw[pos:pos + size]))
pos += size
if [s.name for s in sections] != ['DRIV', 'MAIN', 'PANL']:
raise UpdError('the sections are not DRIV, MAIN, PANL')
return sections
def join(sections_blobs):
return b''.join(b'%d\r\n' % len(b) for b in sections_blobs) + b''.join(sections_blobs)
# --- the packed application -------------------------------------------------
def unpack_lzss(stream):
"""The boot loader's LZSS (0xA0000900): a 4096-byte ring of 4078 spaces and
18 zeros, written from 4078; flag bits LSB first, 1 = literal; a match is
2 bytes, 12-bit ring position and length 3-18."""
ring = bytearray(b' ' * 4078 + b'\0' * 18)
r, out, pos, n = 4078, bytearray(), 0, len(stream)
while pos < n:
flags = stream[pos]
pos += 1
for bit in range(8):
if pos >= n:
break
if flags >> bit & 1:
b = stream[pos]
pos += 1
out.append(b)
ring[r] = b
r = (r + 1) & 0xFFF
else:
if pos + 2 > n:
raise UpdError('the packed application ends inside a match')
lo, hi = stream[pos], stream[pos + 1]
pos += 2
at, count = lo | (hi & 0xF0) << 4, (hi & 0x0F) + 3
for k in range(count):
b = ring[(at + k) & 0xFFF]
out.append(b)
ring[r] = b
r = (r + 1) & 0xFFF
return bytes(out)
def region(flash):
"""(application, packed region bytes) from the flash image."""
size = struct.unpack_from('<I', flash, APP_START)[0]
end = APP_START + 4 + size
if end + 2 > APP_END:
raise UpdError('the packed application runs past 0x%X' % APP_END)
stored = struct.unpack_from('<H', flash, end)[0]
if stored != sum(flash[APP_START:end]) & 0xFFFF:
raise UpdError('the packed application fails its additive checksum')
return unpack_lzss(flash[APP_START + 4:end]), flash[APP_START:end + 2]
def pack_region(stream):
"""[length][stream][additive checksum], padded with 0xFF to 32 bytes."""
out = struct.pack('<I', len(stream)) + stream
out += struct.pack('<H', sum(out) & 0xFFFF)
return out + b'\xff' * (-len(out) % RECORD_DATA)
# --- the stock file -----------------------------------------------------------
class Stock:
"""The official update, checked to be exactly the 1.31 file."""
def __init__(self, raw):
if sha256(raw) != STOCK_UPD_SHA256:
raise UpdError('this is not the original CDJ-900NXS 1.31 update (SHA-256 %s)' % sha256(raw))
self.raw = raw
self.sections = parse(raw)
main = self.sections[1]
if main.version != '1.31' or main.header[31:32] != b'0':
raise UpdError('MAIN is not 1.31 in application-only mode')
kinds = {r.kind for r in main.records}
if not kinds <= {0, 2, 7}:
raise UpdError('MAIN has unexpected record types %s' % sorted(kinds))
self.flash = main.flash()
self.application, _ = region(self.flash)
if len(self.application) != STOCK_APP_SIZE or sha256(self.application) != STOCK_APP_SHA256:
raise UpdError('the 1.31 application inside does not match')
def boot_records(self):
"""Boot loader and recovery: every data record below the application."""
recs = self.sections[1].data_records()
low = [r for r in recs if r.address < APP_START]
if any(r.address + len(r.data) > APP_START for r in low):
raise UpdError('a stock record crosses into the application')
return low
def relabel(header, version):
old = HEADER.match(header).group(0)
new = old[:-4] + version.encode()
if not re.fullmatch(r'\d\.\d\d', version):
raise UpdError('the version label must look like 1.32')
out = header.replace(old, new, 1)
if out[:len(old) - 4] != header[:len(old) - 4] or out[len(old):] != header[len(old):]:
raise UpdError('relabelling would change more than the version')
return out
def build(stock, application, packed_stream, label='1.32'):
"""The new update: stock DRIV and PANL, MAIN with the stock header (version
label only changed), the stock S0, the stock boot loader and recovery
records, the new application region, and the stock S7 end record."""
main = stock.sections[1]
reg = pack_region(packed_stream)
if APP_START + len(reg) > APP_END:
raise UpdError('the packed application needs %d bytes, the flash has %d'
% (len(reg), APP_END - APP_START))
records = [main.records[0]] + stock.boot_records()
records += [Record(2, APP_START + off, reg[off:off + RECORD_DATA]) for off in range(0, len(reg), RECORD_DATA)]
records.append(main.records[-1])
blob = encode_section(relabel(main.header, label) if label else main.header, records)
return join([stock.sections[0].blob, blob, stock.sections[2].blob])
def verify(raw, stock, application, label='1.32'):
"""Every check on a finished file; returns the list of what was checked."""
done = []
sections = parse(raw)
done.append('section lengths, CRC-16 of every section, every S-record checksum')
if sections[0].blob != stock.sections[0].blob or sections[2].blob != stock.sections[2].blob:
raise UpdError('DRIV or PANL differs from stock')
done.append('DRIV and PANL byte for byte as stock')
main, smain = sections[1], stock.sections[1]
if main.header != (relabel(smain.header, label) if label else smain.header):
raise UpdError('the MAIN header differs beyond its version label')
if main.header[31:32] != b'0':
raise UpdError('the MAIN header no longer selects application-only mode')
done.append('MAIN header as stock but for the label %s; application-only mode kept' % (label or '1.31'))
if main.records[0] != smain.records[0] or main.records[-1] != smain.records[-1]:
raise UpdError('the S0 header or S7 entry record changed')
done.append('S0 header record and S7 entry record as stock')
low = [r for r in main.data_records() if r.address < APP_START]
if low != stock.boot_records():
raise UpdError('the boot loader or recovery records changed')
flash = main.flash()
if flash[:APP_START] != stock.flash[:APP_START]:
raise UpdError('the boot loader or recovery bytes changed')
done.append('boot loader and recovery (flash 0x0-0x3FFFF): every record and byte as stock')
if any(r.address < APP_START < r.address + len(r.data) for r in main.data_records()):
raise UpdError('a record crosses into the boot area')
done.append('no record reaches past 0x%X or into the boot area' % APP_END)
app, reg = region(flash)
done.append('packed application: additive checksum')
if app != application:
raise UpdError('the update does not unpack to the patched application')
done.append('unpacked with the boot loader\'s LZSS: exactly the patched application (SHA-256 %s)'
% sha256(app)[:16])
done.append('flash used by the application: %d of %d bytes' % (len(reg), APP_END - APP_START))
return done