Extend pipeline autest to verify HTTP/1.1 request framing

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

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