Handle HTTP/2 1xx interim responses from the origin

src/proxy/http2/Http2ConnectionState.cc

@@ -290,6 +290,42 @@ Http2ConnectionState::rcv_data_frame(const Http2Frame &frame)
  *   2. A HEADERS frame without the END_HEADERS flag set MUST be followed by a
  *      CONTINUATION frame
  */
+namespace
+{
+// An interim (1xx) response received on an outbound
+// (origin) HTTP/2 connection is not the final response. Detect it after the
+// header block is decoded so the caller can discard it and wait for the final
+// response, instead of merging it with the final response headers (which would
+// produce a duplicate :status pseudo-header that fails validation).
+bool
+is_outbound_interim_response(Http2Stream *stream)
+{
+  if (!stream->is_outbound_connection() || stream->trailing_header_is_possible()) {
+    return false;
+  }
+  const MIMEField *status_field = stream->get_receive_header()->field_find(PSEUDO_HEADER_STATUS);
+  if (status_field == nullptr) {
+    return false;
+  }
+  // :status is origin-controlled and HeaderValidator only checks that it is present, so
+  // bound the length before indexing. RFC 9110 15 requires exactly three digits.
+  auto value{status_field->value_get()};
+  return value.length() == 3 && value[0] == '1';
+}
+
+// Discard a decoded interim (1xx) response along with its encoded header block so the
+// following final response is decoded into a clean buffer. Freeing header_blocks here
+// avoids leaking it when the next HEADERS frame allocates a new buffer.
+void
+discard_interim_response(Http2Stream *stream)
+{
+  stream->reset_receive_headers();
+  ats_free(stream->header_blocks);
+  stream->header_blocks        = nullptr;
+  stream->header_blocks_length = 0;
+}
+} // namespace
+
 Http2Error
 Http2ConnectionState::rcv_headers_frame(const Http2Frame &frame)
 {
@@ -511,6 +547,22 @@ Http2ConnectionState::rcv_headers_frame(const Http2Frame &frame)
                         "recv data bad payload length");
     }
 
+    // Discard an interim (1xx) response from the
+    // origin and wait for the final response on this stream.
+    if (is_outbound_interim_response(stream)) {
+      // RFC 9113 8.1: a HEADERS frame with END_STREAM carrying an informational (1xx)
+      // status code is malformed. change_state() has already moved the stream toward
+      // closed, so the final response would have nowhere to go; reject as a stream error.
+      if (stream->receive_end_stream) {
+        return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_STREAM, Http2ErrorCode::HTTP2_ERROR_PROTOCOL_ERROR,
+                          "1xx interim response must not set END_STREAM");
+      }
+      Http2StreamDebug(this->session, stream_id, "received interim 1xx response from origin; awaiting final response");
+      discard_interim_response(stream);
+      this->session->interrupt_reading_frames();
+      return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_NONE);
+    }
+
     // Hard-enforce the global active-streams cap on inbound client streams.
     if (!stream->is_outbound_connection() && !stream->trailing_header_is_possible() && Http2::max_active_streams_policy_in == 1 &&
         Http2::max_active_streams_in > 0) {
@@ -1073,6 +1125,15 @@ Http2ConnectionState::rcv_continuation_frame(const Http2Frame &frame)
                         "continuation half close remote");
     case Http2StreamState::HTTP2_STREAM_STATE_IDLE:
       break;
+    case Http2StreamState::HTTP2_STREAM_STATE_HALF_CLOSED_LOCAL:
+      // On an outbound (origin) connection the response is
+      // received while the stream is half-closed (local); its header block may legitimately
+      // span CONTINUATION frames. The per-minute CONTINUATION flood limit still applies below.
+      if (!stream->is_outbound_connection()) {
+        return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_CONNECTION, Http2ErrorCode::HTTP2_ERROR_PROTOCOL_ERROR,
+                          "continuation bad state");
+      }
+      break;
     default:
       return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_CONNECTION, Http2ErrorCode::HTTP2_ERROR_PROTOCOL_ERROR,
                         "continuation bad state");
@@ -1141,6 +1202,22 @@ Http2ConnectionState::rcv_continuation_frame(const Http2Frame &frame)
                         "recv data bad payload length");
     }
 
+    // Discard an interim (1xx) response from the
+    // origin and wait for the final response on this stream.
+    if (is_outbound_interim_response(stream)) {
+      // RFC 9113 8.1: a HEADERS frame with END_STREAM carrying an informational (1xx)
+      // status code is malformed. change_state() has already moved the stream toward
+      // closed, so the final response would have nowhere to go; reject as a stream error.
+      if (stream->receive_end_stream) {
+        return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_STREAM, Http2ErrorCode::HTTP2_ERROR_PROTOCOL_ERROR,
+                          "1xx interim response must not set END_STREAM");
+      }
+      Http2StreamDebug(this->session, stream_id, "received interim 1xx response from origin; awaiting final response");
+      discard_interim_response(stream);
+      this->session->interrupt_reading_frames();
+      return Http2Error(Http2ErrorClass::HTTP2_ERROR_CLASS_NONE);
+    }
+
     // Hard-enforce the global active-streams cap on inbound client streams.
     if (!stream->is_outbound_connection() && !stream->trailing_header_is_possible() && Http2::max_active_streams_policy_in == 1 &&
         Http2::max_active_streams_in > 0) {
@@ -1152,14 +1229,22 @@ Http2ConnectionState::rcv_continuation_frame(const Http2Frame &frame)
       }
     }
 
-    // Set up the State Machine
-    SCOPED_MUTEX_LOCK(stream_lock, stream->mutex, this_ethread());
-    stream->mark_milestone(Http2StreamMilestone::START_TXN);
-    // This should be fine, need to verify whether we need to replace this with the
-    // "from_early_data" flag from the associated HEADERS frame.
-    stream->new_transaction(frame.is_from_early_data());
-    // Send request header to SM
-    stream->send_headers(*this);
+    // Set up the State Machine. An outbound stream and a trailing header block both
+    // already have a state machine attached, so only a new inbound request may start
+    // one; new_transaction() asserts that none is attached yet. This mirrors the
+    // equivalent branch in rcv_headers_frame().
+    if (!stream->is_outbound_connection() && !stream->trailing_header_is_possible()) {
+      SCOPED_MUTEX_LOCK(stream_lock, stream->mutex, this_ethread());
+      stream->mark_milestone(Http2StreamMilestone::START_TXN);
+      // This should be fine, need to verify whether we need to replace this with the
+      // "from_early_data" flag from the associated HEADERS frame.
+      stream->new_transaction(frame.is_from_early_data());
+      // Send request header to SM
+      stream->send_headers(*this);
+    } else {
+      // Propagate the response (or the trailer) to the existing state machine.
+      stream->send_headers(*this);
+    }
     // Give a chance to send response before reading next frame.
     this->session->interrupt_reading_frames();
   } else {

tests/gold_tests/h2/h2_interim_origin.py

@@ -0,0 +1,179 @@
+#!/usr/bin/env python3
+"""An HTTP/2 (TLS) origin that sends a 1xx interim response before the final 200.
+
+Proxy Verifier cannot emit interim/1xx responses, so this hand-frames HTTP/2 so we
+can exercise ATS origin-side handling of 1xx interim responses.
+
+Modes (chosen by --mode):
+  single     : 103 Early Hints, then 200          (the common CDN/framework preload case)
+  multi      : 103, 103, 100, then 200            (multiple sequential interims)
+  continue   : 100 Continue, then 200
+  cont       : a single 103 whose header block is split across HEADERS+CONTINUATION,
+               then 200                            (multi-frame interim)
+  none       : 200 only                           (control; must always pass)
+  endstream  : a 103 carrying END_STREAM           (malformed, RFC 9113 8.1)
+  finalsplit : no interim; the FINAL 200 header block spans HEADERS+CONTINUATION
+
+Mode names are used as remap path prefixes, and remap matches first-rule-wins on the
+path prefix. Do not name a mode so that it extends another mode's name, or requests for
+the longer path will be routed to the shorter mode's origin.
+"""
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import argparse
+import socket
+import ssl
+import struct
+import subprocess
+import sys
+import tempfile
+import threading
+
+BODY = b"interim-origin-body"
+
+
+def frame(ftype: int, flags: int, sid: int, payload: bytes) -> bytes:
+    return struct.pack(">I", len(payload))[1:] + bytes([ftype, flags]) + struct.pack(">I", sid) + payload
+
+
+def lit(name: str, value: str) -> bytes:
+    # HPACK literal header field without indexing, new name, no Huffman.
+    n = name.encode()
+    v = value.encode()
+    return b"\x00" + bytes([len(n)]) + n + bytes([len(v)]) + v
+
+
+def final_block() -> bytes:
+    return b"\x88" + lit("content-type", "text/plain")  # :status 200 (static idx 8)
+
+
+def interim_block(status: str) -> bytes:
+    return lit(":status", status) + lit("link", "</style.css>; rel=preload; as=style")
+
+
+def send_response(sock: ssl.SSLSocket, mode: str, sid: int) -> None:
+    if mode == "single":
+        sock.sendall(frame(0x1, 0x4, sid, interim_block("103")))
+    elif mode == "multi":
+        sock.sendall(frame(0x1, 0x4, sid, interim_block("103")))
+        sock.sendall(frame(0x1, 0x4, sid, interim_block("103")))
+        sock.sendall(frame(0x1, 0x4, sid, interim_block("100")))
+    elif mode == "continue":
+        sock.sendall(frame(0x1, 0x4, sid, interim_block("100")))
+    elif mode == "cont":
+        blk = interim_block("103")
+        half = len(blk) // 2
+        sock.sendall(frame(0x1, 0x0, sid, blk[:half]))  # HEADERS, no END_HEADERS
+        sock.sendall(frame(0x9, 0x4, sid, blk[half:]))  # CONTINUATION, END_HEADERS
+    elif mode == "endstream":
+        # RFC 9113 8.1 violation: an informational (1xx) response with END_STREAM.
+        sock.sendall(frame(0x1, 0x5, sid, interim_block("103")))  # HEADERS: END_HEADERS | END_STREAM
+        return
+    elif mode == "finalsplit":
+        # The FINAL response header block spans HEADERS+CONTINUATION, with no interim
+        # response at all. Legal HTTP/2 at any block size; nothing requires the block to
+        # exceed SETTINGS_MAX_FRAME_SIZE for a sender to split it.
+        blk = final_block()
+        half = len(blk) // 2
+        sock.sendall(frame(0x1, 0x0, sid, blk[:half]))  # HEADERS, no END_HEADERS
+        sock.sendall(frame(0x9, 0x4, sid, blk[half:]))  # CONTINUATION, END_HEADERS
+        sock.sendall(frame(0x0, 0x1, sid, BODY))  # DATA, END_STREAM
+        return
+    # mode "none": no interim
+    sock.sendall(frame(0x1, 0x4, sid, final_block()))  # final HEADERS, END_HEADERS
+    sock.sendall(frame(0x0, 0x1, sid, BODY))  # DATA, END_STREAM
+
+
+def handle(sock: ssl.SSLSocket, mode: str) -> None:
+    try:
+        sock.sendall(frame(0x4, 0x0, 0, b""))  # server SETTINGS
+        preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
+        buf = b""
+        preface_done = False
+        while True:
+            data = sock.recv(65535)
+            if not data:  # client closed the connection
+                return
+            buf += data
+            if not preface_done:
+                if len(buf) < len(preface):
+                    continue
+                buf = buf[len(preface):]
+                preface_done = True
+            while len(buf) >= 9:
+                ln = int.from_bytes(buf[0:3], "big")
+                if len(buf) < 9 + ln:
+                    break
+                ftype = buf[3]
+                flags = buf[4]
+                sid = int.from_bytes(buf[5:9], "big") & 0x7FFFFFFF
+                buf = buf[9 + ln:]
+                if ftype == 0x4 and not (flags & 0x1):  # client SETTINGS -> ACK it
+                    sock.sendall(frame(0x4, 0x1, 0, b""))
+                if ftype == 0x1:  # a request HEADERS -> respond on the same stream
+                    send_response(sock, mode, sid)
+    finally:
+        sock.close()
+
+
+def make_cert() -> tuple[str, str]:
+    cert = tempfile.NamedTemporaryFile(dir=".", suffix=".crt", delete=False).name
+    key = tempfile.NamedTemporaryFile(dir=".", suffix=".key", delete=False).name
+    subprocess.run(
+        [
+            "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key, "-out", cert, "-days", "3", "-subj",
+            "/CN=interim-origin"
+        ],
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL)
+    return cert, key
+
+
+def parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("address")
+    p.add_argument("port", type=int)
+    p.add_argument("--mode", default="single", choices=["single", "multi", "continue", "cont", "none", "endstream", "finalsplit"])
+    return p.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    cert, key = make_cert()
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ctx.load_cert_chain(cert, key)
+    ctx.set_alpn_protocols(["h2"])
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    srv.bind((args.address, args.port))
+    srv.listen(16)
+    print(f"interim h2 origin listening on {args.address}:{args.port} mode={args.mode}", flush=True)
+    while True:
+        conn, _ = srv.accept()
+        try:
+            tls = ctx.wrap_socket(conn, server_side=True)
+        except Exception as e:
+            sys.stderr.write(f"tls error: {e}\n")
+            conn.close()
+            continue
+        threading.Thread(target=handle, args=(tls, args.mode), daemon=True).start()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())