Handle HTTP/2 1xx interim responses from the origin

src/proxy/http2/Http2ConnectionState.cc

@@ -289,6 +289,41 @@ 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;
+  }
+  // An HTTP/2 :status is always exactly three ASCII digits; 1xx is informational.
+  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)
 {
@@ -510,6 +545,15 @@ 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)) {
+      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);
+    }
+
     // Set up the State Machine
     if (!stream->is_outbound_connection() && !stream->trailing_header_is_possible()) {
       SCOPED_MUTEX_LOCK(stream_lock, stream->mutex, this_ethread());
@@ -1057,6 +1101,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");
@@ -1125,6 +1178,15 @@ 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)) {
+      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);
+    }
+
     // Set up the State Machine
     SCOPED_MUTEX_LOCK(stream_lock, stream->mutex, this_ethread());
     stream->mark_milestone(Http2StreamMilestone::START_TXN);

tests/gold_tests/h2/h2_interim_origin.py

@@ -0,0 +1,155 @@
+#!/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 deepwiki/Vercel 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)
+"""
+#  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, flags, sid, payload):
+    return struct.pack(">I", len(payload))[1:] + bytes([ftype, flags]) + struct.pack(">I", sid) + payload
+
+
+def lit(name, value):
+    # 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():
+    return b"\x88" + lit("content-type", "text/plain")  # :status 200 (static idx 8)
+
+
+def interim_block(status):
+    return lit(":status", status) + lit("link", "</style.css>; rel=preload; as=style")
+
+
+def send_response(sock, mode, sid):
+    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
+    # 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, mode):
+    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:
+            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)
+
+
+def make_cert():
+    cert = tempfile.NamedTemporaryFile(suffix=".crt", delete=False).name
+    key = tempfile.NamedTemporaryFile(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():
+    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"])
+    return p.parse_args()
+
+
+def main():
+    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")
+            continue
+        threading.Thread(target=handle, args=(tls, args.mode), daemon=True).start()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/gold_tests/h2/http2_origin_interim_response.test.py

@@ -0,0 +1,91 @@
+'''
+Verify ATS handles HTTP/2 1xx interim responses from the 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 os
+import sys
+from ports import get_port
+
+Test.Summary = '''
+Verify ATS correctly handles 1xx interim responses (e.g. 103 Early Hints) received
+from an origin over HTTP/2, returning the final 200 to the client.
+'''
+Test.ContinueOnFail = True
+
+ORIGIN = os.path.join(Test.TestDirectory, 'h2_interim_origin.py')
+
+# Each mode is a distinct origin behavior, routed by request path.
+#   single   : 103 then 200            (the deepwiki/Vercel case)
+#   multi    : 103,103,100 then 200     (multiple sequential interims)
+#   continue : 100 then 200
+#   cont     : 103 split across HEADERS+CONTINUATION, then 200
+#   none     : 200 only                 (control)
+MODES = ['single', 'multi', 'continue', 'cont', 'none']
+
+ts = Test.MakeATSProcess("ts", enable_tls=True)
+ts.addDefaultSSLFiles()
+ts.Disk.ssl_multicert_yaml.AddLines(
+    """
+ssl_multicert:
+  - dest_ip: "*"
+    ssl_cert_name: server.pem
+    ssl_key_name: server.key
+""".split("\n"))
+
+# Create an origin process per mode and build the remap table from their ports.
+origins = {}
+remap_lines = []
+for mode in MODES:
+    origin = Test.Processes.Process(f"origin-{mode}")
+    port = get_port(origin, f"port_{mode}")
+    origin.Command = f"{sys.executable} {ORIGIN} 127.0.0.1 {port} --mode {mode}"
+    origin.Ready = When.PortOpenv4(port)
+    origins[mode] = origin
+    remap_lines.append(f"map http://ats.test/{mode} https://127.0.0.1:{port}/")
+
+ts.Disk.remap_config.AddLines(remap_lines)
+ts.Disk.records_config.update(
+    {
+        'proxy.config.ssl.server.cert.path': ts.Variables.SSLDir,
+        'proxy.config.ssl.server.private_key.path': ts.Variables.SSLDir,
+        'proxy.config.ssl.client.alpn_protocols': 'h2,http/1.1',
+        'proxy.config.ssl.client.verify.server.policy': 'PERMISSIVE',
+        'proxy.config.http.server_session_sharing.pool': 'thread',
+        'proxy.config.exec_thread.autoconfig.enabled': 0,
+        'proxy.config.exec_thread.limit': 4,
+        'proxy.config.diags.debug.enabled': 1,
+        'proxy.config.diags.debug.tags': 'http2',
+    })
+
+first = True
+for mode in MODES:
+    tr = Test.AddTestRun(f"h2 origin interim response: mode={mode}")
+    if first:
+        for m in MODES:
+            tr.Processes.Default.StartBefore(origins[m])
+        tr.Processes.Default.StartBefore(ts)
+        first = False
+    tr.MakeCurlCommand(f'-v -s -H "Host: ats.test" http://127.0.0.1:{ts.Variables.port}/{mode}', ts=ts)
+    tr.Processes.Default.ReturnCode = 0
+    tr.StillRunningAfter = ts
+    for m in MODES:
+        tr.StillRunningAfter = origins[m]
+    tr.Processes.Default.Streams.All += Testers.ContainsExpression(
+        'HTTP/.* 200', f'mode={mode}: client must receive the final 200, not a 502')
+    tr.Processes.Default.Streams.All += Testers.ContainsExpression(
+        'interim-origin-body', f'mode={mode}: client must receive the 200 response body')