From 81a3b0dff944ff6344f81f0c7e4290478333f408 Mon Sep 17 00:00:00 2001 From: bneradt Date: Fri, 17 Jul 2026 16:55:37 -0500 Subject: [PATCH] Extend pipeline autest to verify HTTP/1.1 request framing Traffic Server already handles these request shapes correctly; this patch adds no product code and changes no behavior. It only adds test coverage so that the currently correct handling cannot regress unnoticed. A proxy that disagrees with its origin about where a request body ends can be desynchronized from that origin. Two common shapes provoke this: pipelining a body-less POST (Content-Length: 0) ahead of a second request, and sending a request with conflicting Content-Length header fields. Traffic Server must keep such request boundaries intact. This extends the pipeline autest with two runs that exercise those shapes. This confirms that Traffic Server delivers the pipelined requests to the origin as two independent requests, so the second request cannot be folded into the first, and that it rejects the ambiguously-framed request rather than forwarding it. Co-Authored-By: Claude Opus 4.8 --- tests/gold_tests/pipeline/pipeline.test.py | 116 ++++++++++- .../pipeline/request_framing_client.py | 119 ++++++++++++ .../pipeline/request_framing_server.py | 180 ++++++++++++++++++ 3 files changed, 414 insertions(+), 1 deletion(-) create mode 100644 tests/gold_tests/pipeline/request_framing_client.py create mode 100644 tests/gold_tests/pipeline/request_framing_server.py diff --git a/tests/gold_tests/pipeline/pipeline.test.py b/tests/gold_tests/pipeline/pipeline.test.py index dea95511fb4..68217fd4cf9 100644 --- a/tests/gold_tests/pipeline/pipeline.test.py +++ b/tests/gold_tests/pipeline/pipeline.test.py @@ -19,7 +19,7 @@ from ports import get_port import sys -Test.Summary = '''Test pipelined requests.''' +Test.Summary = '''Test pipelined requests and HTTP/1.1 request framing.''' IP_ALLOW_CONTENT = ''' ip_allow: @@ -133,5 +133,119 @@ def _configure_client(self, tr: 'TestRun') -> 'Process': client.StartBefore(self._ts) +class TestRequestFraming: + """Verify Traffic Server preserves HTTP/1.1 request framing. + + A body-less POST (Content-Length: 0) immediately followed by a second + pipelined request must be delivered to the origin as two independent + requests, so the second request cannot be folded into the first. A request + with conflicting Content-Length header fields must be rejected rather than + forwarded. + """ + + _client_script: str = 'request_framing_client.py' + _server_script: str = 'request_framing_server.py' + _counter: int = 0 + + def __init__(self, mode: str) -> None: + """Configure a test run for the given request mode. + + :param mode: 'pipeline' for a body-less POST followed by a pipelined + request, or 'conflicting_cl' for a request with conflicting + Content-Length header fields. + """ + self._mode = mode + self._name = f'framing_{mode}_{TestRequestFraming._counter}' + TestRequestFraming._counter += 1 + + description = { + 'pipeline': 'Test a body-less POST followed by a pipelined request.', + 'conflicting_cl': 'Test a request with conflicting Content-Length headers.', + }[mode] + tr = Test.AddTestRun(description) + tr.TimeOut = 20 + self._configure_server(tr) + self._configure_traffic_server(tr) + self._configure_client(tr) + + def _configure_server(self, tr: 'TestRun') -> 'Process': + """Configure the recording origin server.""" + server = tr.Processes.Process(f'origin_{self._name}') + tr.Setup.Copy(self._server_script) + http_port = get_port(server, "http_port") + server.Command = f'{sys.executable} {self._server_script} 127.0.0.1 {http_port} ' + server.ReturnCode = 0 + server.Ready = When.PortOpenv4(http_port) + + if self._mode == 'pipeline': + # The origin must see exactly the two requests the client sent, with + # their boundaries intact. If ATS folded them together, the origin + # would see a single request or the second request's bytes inside + # the POST body. + server.Streams.All += Testers.ContainsExpression( + r'REQUEST_LINE: POST / HTTP/1.1', 'Origin should receive the POST as its own request.') + server.Streams.All += Testers.ContainsExpression( + r'REQUEST_LINE: GET /second HTTP/1.1', 'Origin should receive the GET as its own request.') + server.Streams.All += Testers.ContainsExpression( + r'ORIGIN_REQUEST_COUNT: 2', 'Origin should receive the second request as a distinct request.') + server.Streams.All += Testers.ExcludesExpression( + r'ORIGIN_REQUEST_COUNT: 3', 'Origin should receive exactly two requests, no more.') + # The second request's bytes must never appear inside the first + # (POST) request's body. + server.Streams.All += Testers.ExcludesExpression( + r"BODY:.*GET /second", 'The GET request must not appear inside the POST body.') + server.Streams.All += Testers.ExcludesExpression( + r"BODY:.*X-Marker", 'The second request header must not appear in the POST body.') + else: + # An ambiguously-framed request must be rejected by ATS before it + # ever reaches the origin. + server.Streams.All += Testers.ExcludesExpression( + r'REQUEST_LINE:', 'Origin must not receive an ambiguously-framed request.') + self._server = server + return server + + def _configure_traffic_server(self, tr: 'TestRun') -> 'Process': + """Configure ATS as a reverse proxy in front of the origin.""" + ts = tr.MakeATSProcess(f'ts_{self._name}', enable_cache=False) + self._ts = ts + ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._server.Variables.http_port}/') + ts.Disk.records_config.update({ + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'http', + }) + return ts + + def _configure_client(self, tr: 'TestRun') -> 'Process': + """Configure the client that sends the framed request.""" + client = tr.Processes.Default + tr.Setup.Copy(self._client_script) + client.Command = ( + f'{sys.executable} {self._client_script} 127.0.0.1 {self._ts.Variables.port} ' + f'www.example.com {self._mode}') + client.ReturnCode = 0 + if self._mode == 'pipeline': + # Two independent responses must come back, one per request. + client.Streams.All += Testers.ContainsExpression( + r'STATUS_LINE_COUNT: 2', 'Client should receive two independent responses.') + client.Streams.All += Testers.ContainsExpression( + r'X-Origin-Response: first', 'Client should receive the response to the POST.') + client.Streams.All += Testers.ContainsExpression( + r'X-Origin-Response: second', 'Client should receive the response to the GET.') + else: + # The conflicting Content-Length request must be rejected with a + # 400, exactly one response must come back, and the second request + # must never be answered. + client.Streams.All += Testers.ContainsExpression( + r'HTTP/1.1 400', 'Client should receive a 400 for the ambiguous request.') + client.Streams.All += Testers.ContainsExpression(r'STATUS_LINE_COUNT: 1', 'Client should receive exactly one response.') + client.Streams.All += Testers.ExcludesExpression( + r'X-Origin-Response: second', 'The second request must not be answered.') + client.StartBefore(self._server) + client.StartBefore(self._ts) + + TestPipelining(buffer_requests=False) TestPipelining(buffer_requests=True) + +TestRequestFraming(mode='pipeline') +TestRequestFraming(mode='conflicting_cl') diff --git a/tests/gold_tests/pipeline/request_framing_client.py b/tests/gold_tests/pipeline/request_framing_client.py new file mode 100644 index 00000000000..adc959b27ef --- /dev/null +++ b/tests/gold_tests/pipeline/request_framing_client.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +"""Send a request over a raw socket and print the responses received. + +Two request shapes are supported: a body-less POST (Content-Length: 0) followed +by a second pipelined request on the same connection, and a single request that +carries conflicting Content-Length header fields. +""" + +# 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 sys + + +def parse_args() -> argparse.Namespace: + """Parse the command line arguments.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("proxy_address", help="Address of the proxy to connect to.") + parser.add_argument("proxy_port", type=int, help="The port of the proxy to connect to.") + parser.add_argument("host", help="The Host header field value to use.") + parser.add_argument( + "mode", + nargs='?', + default='pipeline', + choices=['pipeline', 'conflicting_cl'], + help="Which request to send: a body-less POST followed by a pipelined " + "request, or a request with conflicting Content-Length headers.") + return parser.parse_args() + + +def build_request(mode: str, host: str) -> bytes: + """Build the raw request bytes for the given mode. + + :param mode: 'pipeline' for a body-less POST followed by a pipelined GET, or + 'conflicting_cl' for a request carrying conflicting Content-Length + header fields. + :param host: The Host header field value. + :returns: The raw request bytes. + """ + if mode == 'conflicting_cl': + # Two different Content-Length values are an ambiguous framing that a + # careful proxy must reject (RFC 9112 section 6.3) rather than forward, + # since a downstream server might frame the body differently. + return ( + f'POST / HTTP/1.1\r\n' + f'Host: {host}\r\n' + f'Content-Length: 0\r\n' + f'Content-Length: 38\r\n' + f'Connection: keep-alive\r\n' + f'\r\n' + f'GET /second HTTP/1.1\r\n' + f'Host: {host}\r\n' + f'X-Marker: second-request\r\n' + f'\r\n').encode() + + return ( + f'POST / HTTP/1.1\r\n' + f'Host: {host}\r\n' + f'Content-Length: 0\r\n' + f'Connection: keep-alive\r\n' + f'\r\n' + f'GET /second HTTP/1.1\r\n' + f'Host: {host}\r\n' + f'X-Marker: second-request\r\n' + f'\r\n').encode() + + +def main() -> int: + """Send the request and print the received responses.""" + args = parse_args() + + request = build_request(args.mode, args.host) + + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.connect((args.proxy_address, args.proxy_port)) + print(f'Connected to {args.proxy_address}:{args.proxy_port}') + print(f'Sending request ({len(request)} bytes):') + print(request) + sock.sendall(request) + + sock.settimeout(5.0) + responses = b"" + try: + while True: + data = sock.recv(4096) + if not data: + break + responses += data + # We expect at most two responses; stop once we have both status + # lines and the connection goes quiet. + if responses.count(b'HTTP/1.1') >= 2 and responses.endswith(b'\n'): + break + except socket.timeout: + print('Read timed out.') + + print('==== RESPONSES RECEIVED ====') + print(responses.decode(errors='replace')) + print('==== END RESPONSES ====') + print(f'STATUS_LINE_COUNT: {responses.count(b"HTTP/1.1")}') + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/gold_tests/pipeline/request_framing_server.py b/tests/gold_tests/pipeline/request_framing_server.py new file mode 100644 index 00000000000..01f90985925 --- /dev/null +++ b/tests/gold_tests/pipeline/request_framing_server.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +"""An origin server that records HTTP/1.1 request boundaries. + +The server parses each request's framing itself (headers, then a +Content-Length-delimited body) and prints what it received, so a test can verify +that the proxy delivered each request to the origin with its boundaries intact. +""" + +# 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 signal +import socket +import sys + + +def parse_args() -> argparse.Namespace: + """Parse the command line arguments.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("address", help="Address to listen on.") + parser.add_argument("port", type=int, help="The port to listen on.") + return parser.parse_args() + + +def get_listening_socket(address: str, port: int) -> socket.socket: + """Create a listening socket.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((address, port)) + sock.listen(1) + return sock + + +def recv_until(sock: socket.socket, buffer: bytes, delimiter: bytes) -> bytes: + """Read from the socket until the buffer contains the delimiter. + + :param sock: The socket to read from. + :param buffer: Bytes already read from the socket. + :param delimiter: The delimiter to read until. + :returns: The buffer, guaranteed to contain the delimiter, or all bytes read + before the socket closed. + """ + while delimiter not in buffer: + data = sock.recv(4096) + if not data: + break + buffer += data + return buffer + + +def response_for(method: str, path: str) -> bytes: + """Build the origin response for a given request. + + :param method: The request method. + :param path: The request target. + :returns: The raw response bytes. + """ + if path == '/second': + body = b'second response body\n' + return ( + b'HTTP/1.1 200 OK\r\n' + b'X-Origin-Response: second\r\n' + b'Content-Type: text/plain\r\n' + b'Content-Length: ' + str(len(body)).encode() + b'\r\n\r\n' + body) + body = b'first response body\n' + return ( + b'HTTP/1.1 200 OK\r\n' + b'X-Origin-Response: first\r\n' + b'Content-Type: text/plain\r\n' + b'Content-Length: ' + str(len(body)).encode() + b'\r\n\r\n' + body) + + +def handle_connection(sock: socket.socket) -> None: + """Read and record every request received on a single connection. + + :param sock: The accepted client socket. + """ + sock.settimeout(5.0) + buffer = b"" + request_count = 0 + while True: + try: + buffer = recv_until(sock, buffer, b'\r\n\r\n') + except socket.timeout: + print("Timed out waiting for a request.") + break + if b'\r\n\r\n' not in buffer: + print("Connection closed by peer.") + break + + header_bytes, _, rest = buffer.partition(b'\r\n\r\n') + header_text = header_bytes.decode(errors='replace') + lines = header_text.split('\r\n') + request_line = lines[0] + method = request_line.split(' ')[0] if request_line else '' + path = request_line.split(' ')[1] if len(request_line.split(' ')) > 1 else '' + + content_length = 0 + for line in lines[1:]: + name, _, value = line.partition(':') + if name.strip().lower() == 'content-length': + try: + content_length = int(value.strip()) + except ValueError: + content_length = 0 + + # Read the body, if any, according to Content-Length. + body = rest + timed_out = False + try: + while len(body) < content_length: + data = sock.recv(4096) + if not data: + break + body += data + except socket.timeout: + print("Timed out waiting for the request body.") + timed_out = True + if timed_out: + break + remainder = body[content_length:] + body = body[:content_length] + + request_count += 1 + print(f'---- ORIGIN REQUEST {request_count} ----') + print(f'REQUEST_LINE: {request_line}') + for line in lines[1:]: + print(f'HEADER: {line}') + print(f'BODY_LEN: {len(body)}') + print(f'BODY: {body!r}') + print(f'---- END ORIGIN REQUEST {request_count} ----') + print(f'ORIGIN_REQUEST_COUNT: {request_count}') + sys.stdout.flush() + + sock.sendall(response_for(method, path)) + + # Any bytes past this request's body belong to the next pipelined + # request on the connection. + buffer = remainder + + print(f'TOTAL_ORIGIN_REQUESTS: {request_count}') + sys.stdout.flush() + + +def main() -> int: + """Run the recording origin server until terminated by the test harness.""" + # The test harness terminates this long-running server with SIGTERM once the + # client is done. Exit cleanly (0) so the process return code is expected. + signal.signal(signal.SIGTERM, lambda *_: sys.exit(0)) + + args = parse_args() + try: + with get_listening_socket(args.address, args.port) as listening_sock: + print(f"Listening on {args.address}:{args.port}") + sys.stdout.flush() + while True: + conn, _ = listening_sock.accept() + with conn: + handle_connection(conn) + except (KeyboardInterrupt, SystemExit, OSError): + pass + return 0 + + +if __name__ == "__main__": + sys.exit(main())