bbfs_audit.py

#!/usr/bin/env python3
"""
bbfs_audit.py - Big-block-correct FlashFS extractor + full-chain EDC auditor.

fstool.py's extract path resolves big-block file payloads from
FlashFileSystemEntry.block_number as a direct linear index WITHOUT adding
base_cluster (0xAE0), so extracted big-block files come from wrong offsets.
This tool walks the chain correctly (entry/blockmap are base-relative on disk;
absolute cluster = relative + base), extracts clean bytes, and recomputes the
26-bit EDC on every sub-page of the chain so kernel-visible ECC errors surface.

Image model (64 MB big-block "bigonsmall"): 512+16 sub-pages, 32 sub-pages per
0x4000 cluster (lil-block), raw cluster stride 0x4200. Root block = one cluster:
blockmap on even sub-pages, 32-byte directory entries on odd sub-pages.

Usage:
  python3 bbfs_audit.py <image> [-f FILE] [--base 0xAE0] [--list]
                        [--no-edc] [--spare] [--sha-only]
"""

from __future__ import annotations

import argparse
import hashlib
import struct
import sys
from pathlib import Path

CLUSTER_CLEAN = 0x4000
CLUSTER_RAW = 0x4200          # 32 sub-pages * 528
SUBPAGES = 32
PAGE_DATA = 0x200
PAGE_RAW = 0x210

FS_ROOT_TYPES = (0x2C, 0x30)
STATUS_BAD = 0x1FF0
STATUS_RESERVED = 0x1FFB
STATUS_FREE = 0x1FFE
STATUS_EOF = 0x1FFF

EDC_POLY = 0x6954559


def parse_spare(s: bytes) -> dict:
    # Big-block on-disk order (matches retail/xebuild, verified byte-for-byte):
    #   s[0]=bad marker, s[1..2]=block id, s[5]=seq b0, s[3]=seq b1,
    #   s[4]=seq b2, s[6]=seq b3, s[7..8]=fs_size(LE), s[9]=page_count,
    #   s[12]&0x3F=block type.
    return {
        "bad": s[0],
        "block_id": s[1] | ((s[2] & 0x0F) << 8),
        "seq": s[5] | (s[3] << 8) | (s[4] << 16) | (s[6] << 24),
        "fs_size": s[7] | (s[8] << 8),
        "page_count": s[9],
        "type": s[12] & 0x3F,
    }


def calc_edc(page: bytearray) -> tuple[int, int, int, int]:
    """Port of gxbuild3 Driver::calculate_edc over one 528-byte sub-page."""
    sp = PAGE_DATA
    page[sp + 0xC] &= 0x3F
    page[sp + 0xD] = 0
    page[sp + 0xE] = 0
    page[sp + 0xF] = 0
    val = 0
    v = 0
    idx = 0
    for i in range(0x1066):
        if i == 0x1000:
            idx = sp
        if (i & 31) == 0:
            w = int.from_bytes(page[idx:idx + 4], "little")
            v = (~w) & 0xFFFFFFFF
            idx += 4
        val ^= v & 1
        v >>= 1
        if val & 1:
            val ^= EDC_POLY
        val >>= 1
    val = (~val) & 0xFFFFFFFF
    s12 = (page[sp + 0xC] & 0x3F) | ((val << 6) & 0xC0)
    return s12, (val >> 2) & 0xFF, (val >> 10) & 0xFF, (val >> 18) & 0xFF


class BigFs:
    def __init__(self, data: bytes, base: int):
        self.data = data
        self.base = base
        self.nclusters = len(data) // CLUSTER_RAW

    def subpage(self, cluster: int, sp: int) -> bytes:
        off = cluster * CLUSTER_RAW + sp * PAGE_RAW
        return self.data[off:off + PAGE_RAW]

    def cluster_spare0(self, cluster: int) -> bytes:
        off = cluster * CLUSTER_RAW + PAGE_DATA
        return self.data[off:off + 16]

    def cluster_clean(self, cluster: int) -> bytes:
        out = bytearray()
        base = cluster * CLUSTER_RAW
        for sp in range(SUBPAGES):
            off = base + sp * PAGE_RAW
            out += self.data[off:off + PAGE_DATA]
        return bytes(out)

    def find_root(self) -> tuple[int, int]:
        best = None
        best_seq = -1
        for c in range(self.nclusters):
            s = self.cluster_spare0(c)
            if len(s) < 16:
                break
            p = parse_spare(s)
            if p["type"] in FS_ROOT_TYPES and p["seq"] != 0 and p["bad"] == 0xFF:
                if p["seq"] > best_seq or (p["seq"] == best_seq and best is not None and c > best):
                    best, best_seq = c, p["seq"]
        if best is None:
            raise SystemExit("no FS root found")
        return best, best_seq

    def read_root(self, root_cluster: int) -> bytes:
        return self.cluster_clean(root_cluster)

    @staticmethod
    def map_offset(rel: int) -> int:
        return (rel // 256) * 1024 + (rel % 256) * 2

    def mapval(self, root: bytes, rel: int) -> int:
        off = BigFs.map_offset(rel)
        if off + 2 > len(root):
            return STATUS_EOF
        return struct.unpack_from(">H", root, off)[0]

    def entries(self, root: bytes) -> list[dict]:
        out = []
        for e in range(256):
            off = ((e // 16) * 2 + 1) * 512 + (e % 16) * 32
            if off + 32 > len(root):
                break
            raw = root[off:off + 32]
            name = raw[0:0x16]
            deleted = bool(name) and name[0] == 0x05
            if deleted:
                name = b"_" + name[1:]
            name = name.split(b"\x00")[0].decode("ascii", "replace")
            if not name:
                break
            block_number = struct.unpack_from(">H", raw, 0x16)[0]
            length = struct.unpack_from(">I", raw, 0x18)[0]
            out.append({"name": name, "rel": block_number, "len": length,
                        "deleted": deleted})
        return out

    def chain(self, root: bytes, rel_start: int) -> list[int]:
        chain = []
        cur = rel_start
        seen = set()
        while cur < STATUS_BAD and cur not in seen:
            seen.add(cur)
            chain.append(cur)
            nxt = self.mapval(root, cur)
            if nxt >= STATUS_BAD:
                break
            cur = nxt
        return chain

    def extract(self, root: bytes, entry: dict) -> bytes:
        buf = bytearray()
        for rel in self.chain(root, entry["rel"]):
            buf += self.cluster_clean(rel + self.base)
            if len(buf) >= entry["len"]:
                break
        return bytes(buf[:entry["len"]])

    def audit_edc(self, chain_rel: list[int]) -> dict:
        total = bad = 0
        first_bad = None
        for rel in chain_rel:
            c = rel + self.base
            for sp in range(SUBPAGES):
                raw = self.subpage(c, sp)
                if len(raw) < PAGE_RAW:
                    continue
                stored = raw[0x20C:0x210]
                page = bytearray(raw)
                s12, s13, s14, s15 = calc_edc(page)
                total += 1
                if bytes((s12, s13, s14, s15)) != stored:
                    bad += 1
                    if first_bad is None:
                        first_bad = (c, sp, stored.hex(" "),
                                     bytes((s12, s13, s14, s15)).hex(" "))
        return {"pages": total, "bad": bad, "first_bad": first_bad}


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("image", type=Path)
    ap.add_argument("-f", "--file", default="xam.xex")
    ap.add_argument("--base", default="0xAE0")
    ap.add_argument("--list", action="store_true", help="list entries and exit")
    ap.add_argument("--no-edc", action="store_true", help="skip the EDC audit")
    ap.add_argument("--spare", action="store_true", help="dump per-cluster chain spare")
    ap.add_argument("-o", "--output", type=Path, default=None,
                    help="write the extracted file here")
    args = ap.parse_args()

    base = int(args.base, 0)
    data = args.image.read_bytes()
    fs = BigFs(data, base)
    root_cluster, root_seq = fs.find_root()
    root = fs.read_root(root_cluster)
    print(f"{args.image.name}: root cluster=0x{root_cluster:X} seq=0x{root_seq:X} base=0x{base:X}")

    ents = fs.entries(root)
    if args.list:
        for e in ents:
            print(f"  {'[del] ' if e['deleted'] else ''}{e['name']:24} "
                  f"rel=0x{e['rel']:04X} abs=0x{e['rel']+base:X} len=0x{e['len']:X}")
        return 0

    target = next((e for e in ents if e["name"] == args.file and not e["deleted"]), None)
    if target is None:
        print(f"  entry '{args.file}' NOT FOUND")
        return 2
    chain_rel = fs.chain(root, target["rel"])
    payload = fs.extract(root, target)
    print(f"  {args.file}: rel=0x{target['rel']:X} abs=0x{target['rel']+base:X} "
          f"len=0x{target['len']:X} chain={len(chain_rel)} clusters")
    print(f"    first16={payload[:16].hex(' ')}")
    print(f"    sha256={hashlib.sha256(payload).hexdigest()}")
    if args.output:
        args.output.write_bytes(payload)
        print(f"    wrote {args.output}")

    if args.spare:
        print("    chain cluster spare (abs, type, seq, fs_size, pc, bad):")
        for rel in chain_rel:
            s = fs.cluster_spare0(rel + base)
            p = parse_spare(s)
            print(f"      0x{rel+base:04X} type=0x{p['type']:02X} seq=0x{p['seq']:X} "
                  f"fs_size=0x{p['fs_size']:X} pc=0x{p['page_count']:X} bad=0x{p['bad']:02X}")

    if not args.no_edc:
        res = fs.audit_edc(chain_rel)
        verdict = "OK" if res["bad"] == 0 else f"INVALID ({res['bad']})"
        print(f"    EDC over {res['pages']} sub-pages: {verdict}")
        if res["first_bad"]:
            c, sp, stored, comp = res["first_bad"]
            print(f"      first bad: cluster=0x{c:X} sub={sp} stored={stored} computed={comp}")
    return 0


if __name__ == "__main__":
    sys.exit(main())