CSR OTAU binary parser

csrotau.py

#!/usr/bin/env python3

# CSR OTAU binary parser
# https://developer.qualcomm.com/qfile/34081/csr102x_otau_overview.pdf
# For use with test and demonstration only. This is obviously not official and
# is not affiliated with Qualcomm.

import io
import os
import sys

MAGIC_HEADER = b'APPUHDR2'
MAGIC_PART = b'PARTDATA'
MAGIC_FOOTER = b'APPUPFTR'

def read_otau(otau_path, out_path=None):
    with open(otau_path, 'rb') as otau:
        magic = otau.read(8)
        if magic != MAGIC_HEADER:
            raise RuntimeError('Not an OTAU file.')
        print_otau_header(otau)
        seq = 0
        while True:
            magic = otau.read(8)
            if len(magic) == 0:
                print('EOF')
                break
            if magic == MAGIC_PART:
                print()
                ptype, pnum, psize = parse_partition_header(otau)
                if out_path is not None:
                    out_file_path = os.path.join(out_path, f'{os.path.basename(otau_path)}.s{seq}.n{pnum}.t{ptype}')
                    print(f'Dumping partition {pnum} to', out_file_path)
                    with open(out_file_path, 'wb') as part:
                        part.write(otau.read(psize))
                else:
                    otau.seek(psize, io.SEEK_CUR)
                seq += 1
            elif magic == MAGIC_FOOTER:
                footer_size = int.from_bytes(otau.read(4), 'big')
                if out_path is not None:
                    print(f'Dumping footer')
                    with open(f'{os.path.basename(otau_path)}.footer', 'wb') as footer:
                        footer.write(otau.read(footer_size))
                else:
                    otau.seek(footer_size, io.SEEK_CUR)
            else:
                raise RuntimeError(f'Unexpected magic {repr(magic)}')

def print_otau_header(otau):
    hsize = int.from_bytes(otau.read(4), 'big')
    begin_pos = otau.tell()
    model = otau.read(8)
    ver = int.from_bytes(otau.read(4), 'big')
    n_compatatible_upgrades = int.from_bytes(otau.read(2), 'big')
    compatatible_upgrades = tuple(int.from_bytes(otau.read(4), 'big') for _ in range(n_compatatible_upgrades))
    ps_config_ver = int.from_bytes(otau.read(2), 'big')
    n_prev_ps_config_vers = int.from_bytes(otau.read(2), 'big')
    prev_ps_config_vers = tuple(int.from_bytes(otau.read(2), 'big') for _ in range(n_prev_ps_config_vers))
    end_pos = otau.tell()
    assert end_pos - begin_pos == hsize, f'Header size mismatch (got {end_pos - begin_pos}, expecting {hsize})'

    print('Model:', model)
    print('Upgrade version:', f'{ver >> 16}.{ver & 0xffff}')
    print('Compatible upgrades:')
    for cu in compatatible_upgrades:
        print(f'  - {cu >> 16}.{cu & 0xffff}')
    print('PS (Persistent Storage?) config ver:', ps_config_ver)
    print('Prev PS config vers:')
    for ppsv in prev_ps_config_vers:
        print(f'  - {ppsv}')

def parse_partition_header(otau):
    # exclude ptype and pnum fields
    psize = int.from_bytes(otau.read(4), 'big') - 4
    ptype = int.from_bytes(otau.read(2), 'big')
    pnum = int.from_bytes(otau.read(2), 'big')

    print('Partition #:', pnum)
    print('Partition type:', ptype)
    print('Payload size:', psize)
    return ptype, pnum, psize

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} <otau> [outdir]')
        sys.exit(1)

    read_otau(sys.argv[1], (sys.argv[2] if len(sys.argv) >= 3 else None))

Hi, thanks for your work. What's the license for this snippet? I'm working on Plutoberth/SonyHeadphonesClient#35 .

@Plutoberth You can just take it. It can be considered as public domain or under Unlicense.

Also thanks for sharing your project. I'm really interested in it. Fun fact: this script was created when I was looking at firmwares for my WH-1000XM2 but ended up not going further than extracting the files due to lack of time and interest.

Awesome, thanks! Pretty cool goal :)