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())