diff --git a/src/proxy/http/HttpSM.cc b/src/proxy/http/HttpSM.cc index 0d54ad376d7..7fa59ac9f2c 100644 --- a/src/proxy/http/HttpSM.cc +++ b/src/proxy/http/HttpSM.cc @@ -2145,6 +2145,16 @@ HttpSM::state_read_server_response_header(int event, void *data) // If there is a post body in transit, give up on it if (tunnel.is_tunnel_alive()) { tunnel.abort_tunnel(); + // abort_tunnel() cancels I/O but does not close VCs or clean up + // vc_table entries. When a request transform is active the + // post_transform_info entry still references the TransformVConnection + // with in_tunnel=true, which causes cleanup_entry() to skip + // do_io_close() — leaking the VC. Close it explicitly here. + if (post_transform_info.entry != nullptr) { + post_transform_info.vc->do_io_close(); + vc_table.cleanup_entry(post_transform_info.entry); + post_transform_info.entry = nullptr; + } // Make sure client connection is closed when we are done in case there is cruft left over t_state.client_info.keep_alive = HTTPKeepAlive::NO_KEEPALIVE; // Similarly the server connection should also be closed diff --git a/tests/gold_tests/slow_post/partial_post_client.py b/tests/gold_tests/slow_post/partial_post_client.py new file mode 100644 index 00000000000..d6ad9240a5b --- /dev/null +++ b/tests/gold_tests/slow_post/partial_post_client.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +"""Send a partial POST to trigger the abort_tunnel code path. + +Sends POST headers claiming a large Content-Length but only sends a small +chunk of body data. When a request transform plugin is active, this causes +ATS to call abort_tunnel() while the transform entry is still in the vc_table. +""" + +# 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 socket +import sys + + +def main() -> int: + """Run the client.""" + host = sys.argv[1] if len(sys.argv) > 1 else '127.0.0.1' + port = int(sys.argv[2]) if len(sys.argv) > 2 else 8080 + + request = ( + 'POST / HTTP/1.1\r\n' + 'Host: quick.server.com\r\n' + 'Content-Type: application/octet-stream\r\n' + 'Content-Length: 100000\r\n' + '\r\n').encode() + + partial_body = b'x' * 4096 + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(5) + try: + sock.connect((host, port)) + sock.sendall(request + partial_body) + print(f'Sent POST headers (Content-Length: 100000) + {len(partial_body)} bytes') + + try: + response = sock.recv(4096) + except ConnectionError: + # ATS may reset the connection after responding since the POST + # body is incomplete. This is acceptable — the important thing + # is that ATS did not crash. + print('HTTP/1.1 connection reset (expected for partial POST)') + return 0 + except socket.timeout: + print('ERROR: timeout waiting for response', file=sys.stderr) + return 1 + + if response: + first_line = response.split(b'\r\n')[0].decode(errors='replace') + print(first_line) + if first_line.startswith('HTTP/1.1'): + return 0 + print('ERROR: unexpected response', file=sys.stderr) + return 1 + else: + print('ERROR: connection closed with no response', file=sys.stderr) + return 1 + finally: + sock.close() + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tests/gold_tests/slow_post/quick_server.test.py b/tests/gold_tests/slow_post/quick_server.test.py index fe250407987..ee64aa55902 100644 --- a/tests/gold_tests/slow_post/quick_server.test.py +++ b/tests/gold_tests/slow_post/quick_server.test.py @@ -1,4 +1,8 @@ -"""Verify ATS handles a server that replies before receiving a full request.""" +"""Verify ATS handles a server that replies before receiving a full request. + +Also verifies ATS does not leak the TransformVConnection when abort_tunnel() +is called with a request transform plugin active (use_request_transform=True). +""" # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file @@ -27,24 +31,27 @@ class QuickServerTest: """Verify that ATS doesn't delay responses behind slow posts.""" _init_file = '__init__.py' - _http_utils = 'http_utils.py' _slow_post_client = 'slow_post_client.py' + _partial_post_client = 'partial_post_client.py' _quick_server = 'quick_server.py' _dns_counter = 0 _server_counter = 0 _ts_counter = 0 - def __init__(self, abort_request: bool, drain_request: bool, abort_response_headers: bool): + def __init__(self, abort_request: bool, drain_request: bool, abort_response_headers: bool, use_request_transform: bool = False): """Initialize the test. :param drain_request: Whether the server should drain the request body. :param abort_request: Whether the client should abort the request body. - before disconnecting. + :param abort_response_headers: Whether the server should abort response headers. + :param use_request_transform: Whether to install a request transform plugin + hooked at TS_HTTP_READ_REQUEST_HDR_HOOK and use a partial POST client. """ self._should_drain_request = drain_request self._should_abort_request = abort_request self._should_abort_response_headers = abort_response_headers + self._use_request_transform = use_request_transform def _configure_dns(self, tr: 'TestRun') -> None: """Configure the DNS. @@ -89,13 +96,17 @@ def _configure_traffic_server(self, tr: 'TestRun'): 'proxy.config.dns.nameservers': f'127.0.0.1:{self._dns.Variables.Port}', 'proxy.config.dns.resolv_conf': 'NULL', }) + if self._use_request_transform: + Test.PrepareTestPlugin( + os.path.join(Test.Variables.AtsTestPluginsDir, 'tunnel_transform.so'), self._ts, plugin_args='request_hdr') def run(self): """Run the test.""" tr = Test.AddTestRun( f'Aborting request: {self._should_abort_request}, ' f'Draining request: {self._should_drain_request}, ' - f'Aborting response headers: {self._should_abort_response_headers}') + f'Aborting response headers: {self._should_abort_response_headers}, ' + f'Request transform: {self._use_request_transform}') self._configure_dns(tr) self._configure_server(tr) @@ -105,22 +116,30 @@ def run(self): http_utils = os.path.join(tools_dir, 'http_utils.py') tr.Setup.CopyAs(self._init_file, Test.RunDirectory) tr.Setup.CopyAs(http_utils, Test.RunDirectory) - tr.Setup.CopyAs(self._slow_post_client, Test.RunDirectory) tr.Setup.CopyAs(self._quick_server, Test.RunDirectory) - client_command = (f'{sys.executable} {self._slow_post_client} ' - '127.0.0.1 ' - f'{self._ts.Variables.port} ') - if not self._should_abort_request: - client_command += '--finish-request ' - p = tr.Processes.Default - p.Command = client_command - if self._should_abort_request or self._should_abort_response_headers: - p.Streams.All += Testers.ExcludesExpression('HTTP/1.1 200 OK', 'Verify response was received') + if self._use_request_transform: + tr.Setup.CopyAs(self._partial_post_client, Test.RunDirectory) + p = tr.Processes.Default + p.Command = (f'{sys.executable} {self._partial_post_client} ' + f'127.0.0.1 {self._ts.Variables.port}') + p.ReturnCode = 0 + p.Streams.All += Testers.ContainsExpression('HTTP/1.1', 'Verify client received an HTTP response') else: - p.Streams.All += Testers.ContainsExpression('HTTP/1.1 200 OK', 'Verify response was received') + tr.Setup.CopyAs(self._slow_post_client, Test.RunDirectory) + client_command = (f'{sys.executable} {self._slow_post_client} ' + '127.0.0.1 ' + f'{self._ts.Variables.port} ') + if not self._should_abort_request: + client_command += '--finish-request ' + p = tr.Processes.Default + p.Command = client_command + if self._should_abort_request or self._should_abort_response_headers: + p.Streams.All += Testers.ExcludesExpression('HTTP/1.1 200 OK', 'Verify response was received') + else: + p.Streams.All += Testers.ContainsExpression('HTTP/1.1 200 OK', 'Verify response was received') + p.ReturnCode = 0 - p.ReturnCode = 0 self._ts.StartBefore(self._dns) self._ts.StartBefore(self._server) p.StartBefore(self._ts) @@ -132,3 +151,7 @@ def run(self): for abort_response_headers in [True, False]: test = QuickServerTest(abort_request, drain_request, abort_response_headers) test.run() + +# Partial POST with a request transform plugin: exercises the abort_tunnel() +# cleanup path for TransformVConnection entries in the vc_table. +QuickServerTest(abort_request=True, drain_request=False, abort_response_headers=False, use_request_transform=True).run() diff --git a/tests/tools/plugins/tunnel_transform.cc b/tests/tools/plugins/tunnel_transform.cc index 8b08acd70d6..0731af34772 100644 --- a/tests/tools/plugins/tunnel_transform.cc +++ b/tests/tools/plugins/tunnel_transform.cc @@ -1,6 +1,12 @@ /** @file - An example program that does a null transform of response body content. + Null transform plugin for request and/or response body content. + + Modes (selected by plugin argument): + (default) — hook at TS_HTTP_TUNNEL_START_HOOK, add both request + and response transforms (original behavior) + request_hdr — hook at TS_HTTP_READ_REQUEST_HDR_HOOK, add only the + request transform @section license License @@ -21,6 +27,7 @@ limitations under the License. */ +#include #include #include #include @@ -31,10 +38,11 @@ static const char PLUGIN_TAG[] = PLUGIN_NAME; static DbgCtl plugin_ctl{PLUGIN_TAG}; -static int stat_ua_bytes_sent = 0; // number of bytes seen by the transform from UA to OS -static int stat_os_bytes_sent = 0; // number of bytes seen by the transform from OS to UA -static int stat_error = 0; -static int stat_test_done = 0; +static int stat_ua_bytes_sent = 0; // number of bytes seen by the transform from UA to OS +static int stat_os_bytes_sent = 0; // number of bytes seen by the transform from OS to UA +static int stat_error = 0; +static int stat_test_done = 0; +static bool request_hdr_mode = false; // when true, hook at READ_REQUEST_HDR and request transform only typedef struct { TSVIO output_vio; @@ -99,7 +107,8 @@ handle_transform(TSCont contp, bool forward) data->output_buffer = TSIOBufferCreate(); data->output_reader = TSIOBufferReaderAlloc(data->output_buffer); Dbg(plugin_ctl, "\tWriting %" PRId64 " bytes on VConn", TSVIONBytesGet(input_vio)); - data->output_vio = TSVConnWrite(output_conn, contp, data->output_reader, INT64_MAX); + int64_t nbytes = (request_hdr_mode && TSVIONBytesGet(input_vio) > 0) ? TSVIONBytesGet(input_vio) : INT64_MAX; + data->output_vio = TSVConnWrite(output_conn, contp, data->output_reader, nbytes); TSContDataSet(contp, data); } @@ -267,10 +276,12 @@ static void transform_add(TSHttpTxn txnp) { Dbg(plugin_ctl, "Entering transform_add()"); - TSVConn connp = TSTransformCreate(forward_null_transform, txnp); - TSVConn rev_connp = TSTransformCreate(reverse_null_transform, txnp); + TSVConn connp = TSTransformCreate(forward_null_transform, txnp); TSHttpTxnHookAdd(txnp, TS_HTTP_REQUEST_TRANSFORM_HOOK, connp); - TSHttpTxnHookAdd(txnp, TS_HTTP_RESPONSE_TRANSFORM_HOOK, rev_connp); + if (!request_hdr_mode) { + TSVConn rev_connp = TSTransformCreate(reverse_null_transform, txnp); + TSHttpTxnHookAdd(txnp, TS_HTTP_RESPONSE_TRANSFORM_HOOK, rev_connp); + } } static int @@ -280,8 +291,9 @@ transform_plugin(TSCont /* contp ATS_UNUSED */, TSEvent event, void *edata) Dbg(plugin_ctl, "Entering transform_plugin()"); switch (event) { + case TS_EVENT_HTTP_READ_REQUEST_HDR: case TS_EVENT_HTTP_TUNNEL_START: - Dbg(plugin_ctl, "\tEvent is TS_EVENT_HTTP_TUNNEL_START"); + Dbg(plugin_ctl, "\tEvent is %d", event); transform_add(txnp); TSHttpTxnReenable(txnp, TS_EVENT_HTTP_CONTINUE); return 0; @@ -301,7 +313,7 @@ handleMsg(TSCont /* cont ATS_UNUSED */, TSEvent event, void * /* edata ATS_UNUSE } void -TSPluginInit(int /* argc ATS_UNUSED */, const char ** /* argv ATS_UNUSED */) +TSPluginInit(int argc, const char **argv) { TSPluginRegistrationInfo info; @@ -315,6 +327,11 @@ TSPluginInit(int /* argc ATS_UNUSED */, const char ** /* argv ATS_UNUSED */) goto Lerror; } + if (argc > 1 && strcmp(argv[1], "request_hdr") == 0) { + request_hdr_mode = true; + } + Dbg(plugin_ctl, "mode: %s", request_hdr_mode ? "request_hdr" : "tunnel_start"); + stat_ua_bytes_sent = TSStatCreate("tunnel_transform.ua.bytes_sent", TS_RECORDDATATYPE_INT, TS_STAT_NON_PERSISTENT, TS_STAT_SYNC_SUM); stat_os_bytes_sent = @@ -322,7 +339,11 @@ TSPluginInit(int /* argc ATS_UNUSED */, const char ** /* argv ATS_UNUSED */) stat_error = TSStatCreate("tunnel_transform.error", TS_RECORDDATATYPE_INT, TS_STAT_NON_PERSISTENT, TS_STAT_SYNC_SUM); stat_test_done = TSStatCreate("tunnel_transform.test.done", TS_RECORDDATATYPE_INT, TS_STAT_NON_PERSISTENT, TS_STAT_SYNC_SUM); - TSHttpHookAdd(TS_HTTP_TUNNEL_START_HOOK, TSContCreate(transform_plugin, nullptr)); + if (request_hdr_mode) { + TSHttpHookAdd(TS_HTTP_READ_REQUEST_HDR_HOOK, TSContCreate(transform_plugin, nullptr)); + } else { + TSHttpHookAdd(TS_HTTP_TUNNEL_START_HOOK, TSContCreate(transform_plugin, nullptr)); + } TSLifecycleHookAdd(TS_LIFECYCLE_MSG_HOOK, TSContCreate(handleMsg, TSMutexCreate())); return;