PoC: Stale HTTP/2 control frames leaked to new upstream on proxy_v2 reinit (nginx PR #1135)

poc_stale_frames.py

#!/usr/bin/env python3
"""
PoC: Stale HTTP/2 control frames leaked to new upstream on reinit.

Demonstrates that without the ctx->out = NULL fix in
ngx_http_proxy_v2_reinit_request(), PING ACK / SETTINGS ACK frames
queued for a failed upstream connection are sent to the next upstream
during a proxy_next_upstream retry.

Setup:
  upstream1 (port 9081): Completes HTTP/2 handshake, sends PING,
                         then GOAWAY with last_stream_id=0 to force retry.
  upstream2 (port 9082): Accepts connection, logs all received HTTP/2 frames.
  nginx (port 9080):     proxy_pass to upstream{} with proxy_http_version 2.
"""

import os
import socket
import struct
import subprocess
import sys
import tempfile
import threading
import time

# HTTP/2 frame types
DATA          = 0x0
HEADERS       = 0x1
RST_STREAM    = 0x3
SETTINGS      = 0x4
PING          = 0x6
GOAWAY        = 0x7
WINDOW_UPDATE = 0x8

# Flags
ACK = 0x1

H2_PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"

FRAME_NAMES = {
    0x0: "DATA", 0x1: "HEADERS", 0x2: "PRIORITY", 0x3: "RST_STREAM",
    0x4: "SETTINGS", 0x5: "PUSH_PROMISE", 0x6: "PING", 0x7: "GOAWAY",
    0x8: "WINDOW_UPDATE", 0x9: "CONTINUATION",
}


def make_frame(ftype, flags, stream_id, payload=b""):
    length = len(payload)
    hdr = struct.pack("!I", length)[1:]  # 3 bytes
    hdr += struct.pack("!BB", ftype, flags)
    hdr += struct.pack("!I", stream_id & 0x7FFFFFFF)
    return hdr + payload


def parse_frames(data):
    """Parse all HTTP/2 frames from raw bytes (after stripping preface)."""
    frames = []
    off = 0
    while off + 9 <= len(data):
        length = struct.unpack("!I", b"\x00" + data[off:off+3])[0]
        ftype = data[off+3]
        flags = data[off+4]
        sid = struct.unpack("!I", data[off+5:off+9])[0] & 0x7FFFFFFF
        payload = data[off+9:off+9+length]
        if off + 9 + length > len(data):
            break  # incomplete frame
        frames.append((ftype, flags, sid, payload))
        off += 9 + length
    return frames


def upstream1(port):
    """Fake upstream: SETTINGS + PING + GOAWAY(0) to force retry."""
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("127.0.0.1", port))
    srv.listen(1)
    srv.settimeout(10)

    try:
        conn, _ = srv.accept()
    except socket.timeout:
        srv.close()
        return

    # Read nginx's connection preface + request
    data = b""
    while len(data) < len(H2_PREFACE) + 9:
        chunk = conn.recv(4096)
        if not chunk:
            break
        data += chunk

    # Send ALL frames in a single write so nginx processes them in one
    # pass through process_header, before the output filter can flush ctx->out.
    resp = b""
    # Server SETTINGS (empty, no settings to change)
    resp += make_frame(SETTINGS, 0, 0)
    # SETTINGS ACK (ack nginx's settings)
    resp += make_frame(SETTINGS, ACK, 0)
    # PING with opaque data
    ping_data = b"\xDE\xAD\xBE\xEF\xCA\xFE\xBA\xBE"
    resp += make_frame(PING, 0, 0, ping_data)
    # GOAWAY: last_stream_id=0, error_code=0 (NO_ERROR)
    # last_stream_id=0 < nginx's stream 1, so nginx must retry
    goaway_payload = struct.pack("!II", 0, 0)  # last_stream_id=0, error=0
    resp += make_frame(GOAWAY, 0, 0, goaway_payload)
    conn.sendall(resp)
    time.sleep(0.1)
    conn.close()
    srv.close()


upstream2_frames = []


def upstream2(port):
    """Second upstream: log all received frames, send a minimal 200 response."""
    global upstream2_frames
    upstream2_frames = []

    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("127.0.0.1", port))
    srv.listen(1)
    srv.settimeout(10)

    try:
        conn, _ = srv.accept()
    except socket.timeout:
        srv.close()
        return

    conn.settimeout(2)

    # Read all data nginx sends
    data = b""
    while True:
        try:
            chunk = conn.recv(4096)
            if not chunk:
                break
            data += chunk
        except socket.timeout:
            break

    # Strip H2 preface if present
    if data.startswith(H2_PREFACE):
        raw = data[len(H2_PREFACE):]
    else:
        raw = data

    upstream2_frames = parse_frames(raw)

    # We don't need to send a valid response for the PoC; nginx will
    # eventually time out, but we have the frames we need.
    conn.close()
    srv.close()


def write_nginx_conf(tmpdir, nginx_port, up1_port, up2_port):
    conf = f"""\
daemon off;
master_process off;
worker_processes 1;
error_log {tmpdir}/error.log debug;
pid {tmpdir}/nginx.pid;

events {{
    worker_connections 64;
}}

http {{
    access_log off;

    upstream backend {{
        server 127.0.0.1:{up1_port};
        server 127.0.0.1:{up2_port};
    }}

    server {{
        listen {nginx_port};

        location / {{
            proxy_pass http://backend;
            proxy_http_version 2;
            proxy_next_upstream error timeout invalid_header;
            proxy_connect_timeout 2s;
            proxy_read_timeout 2s;
        }}
    }}
}}
"""
    path = os.path.join(tmpdir, "nginx.conf")
    with open(path, "w") as f:
        f.write(conf)
    return path


def main():
    nginx_bin = os.path.join(os.path.dirname(__file__), "objs", "nginx")
    if not os.path.isfile(nginx_bin):
        print(f"ERROR: nginx binary not found at {nginx_bin}", file=sys.stderr)
        sys.exit(1)

    nginx_port = 9080
    up1_port = 9081
    up2_port = 9082

    tmpdir = tempfile.mkdtemp(prefix="nginx_poc_")

    # Ensure temp dirs exist for nginx
    for d in ("client_body_temp", "proxy_temp", "fastcgi_temp",
              "uwsgi_temp", "scgi_temp", "logs"):
        os.makedirs(os.path.join(tmpdir, d), exist_ok=True)

    conf_path = write_nginx_conf(tmpdir, nginx_port, up1_port, up2_port)

    # Start upstream servers
    t1 = threading.Thread(target=upstream1, args=(up1_port,), daemon=True)
    t2 = threading.Thread(target=upstream2, args=(up2_port,), daemon=True)
    t1.start()
    t2.start()
    time.sleep(0.2)

    # Start nginx
    nginx_proc = subprocess.Popen(
        [nginx_bin, "-c", conf_path, "-p", tmpdir],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )
    time.sleep(0.5)

    # Send a request through nginx
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(5)
        s.connect(("127.0.0.1", nginx_port))
        s.sendall(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")
        try:
            s.recv(4096)
        except socket.timeout:
            pass
        s.close()
    except Exception as e:
        print(f"Request failed: {e}")

    # Wait for upstreams to finish
    t1.join(timeout=5)
    t2.join(timeout=5)

    # Kill nginx
    nginx_proc.terminate()
    nginx_proc.wait(timeout=5)

    # Analyze frames received by upstream2
    print("\n=== Frames received by upstream2 ===")

    stale_ping_ack = False
    stale_settings_ack_count = 0
    expected_settings_ack = False

    for ftype, flags, sid, payload in upstream2_frames:
        name = FRAME_NAMES.get(ftype, f"UNKNOWN({ftype})")
        flag_str = ""
        if flags & ACK:
            flag_str = " [ACK]"
        print(f"  {name}{flag_str}  stream={sid}  len={len(payload)}")

        if ftype == PING and (flags & ACK):
            stale_ping_ack = True
            print(f"    *** PING ACK data: {payload.hex()}")

        if ftype == SETTINGS and (flags & ACK):
            stale_settings_ack_count += 1

        if ftype == SETTINGS and not (flags & ACK):
            # nginx's own SETTINGS -> one SETTINGS ACK is expected later
            expected_settings_ack = True

    print()

    # The normal flow: nginx sends SETTINGS (its own), then eventually
    # SETTINGS ACK after receiving upstream2's SETTINGS. A SETTINGS ACK
    # arriving *before* upstream2 sends anything is stale.
    # A PING ACK without upstream2 ever sending PING is always stale.

    bug_found = False

    if stale_ping_ack:
        print("BUG: upstream2 received PING ACK but never sent a PING!")
        print("     This is a stale frame leaked from the failed upstream1 connection.")
        bug_found = True

    # More than 1 SETTINGS ACK is suspicious (one is expected for nginx's own settings response)
    if stale_settings_ack_count > 1:
        print(f"BUG: upstream2 received {stale_settings_ack_count} SETTINGS ACK frames")
        print("     (expected at most 1). Extra ones are stale from upstream1.")
        bug_found = True

    if not bug_found:
        print("OK: No stale control frames detected on upstream2.")
        print("    (The fix in reinit_request is working correctly.)")

    # Show relevant debug log lines
    error_log = os.path.join(tmpdir, "error.log")
    if os.path.exists(error_log):
        with open(error_log) as f:
            lines = f.readlines()
        relevant = [l.rstrip() for l in lines
                    if "goaway" in l.lower()
                    or "ping ack" in l.lower()
                    or "reinit" in l.lower()
                    or "next upstream" in l.lower()
                    or "settings ack" in l.lower()]
        if relevant:
            print("\n=== Relevant nginx debug log lines ===")
            for l in relevant[:20]:
                print(f"  {l}")

    print(f"\nFull debug log: {error_log}")
    print(f"Temp dir: {tmpdir}")

    sys.exit(1 if bug_found else 0)


if __name__ == "__main__":
    main()