Fix TransformVConnection resource leak when abort_tunnel() is called with active request transform

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

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

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

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 <cstring>
 #include <stdio.h>
 #include <unistd.h>
 #include <inttypes.h>
@@ -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,14 +327,23 @@ 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 =
     TSStatCreate("tunnel_transform.os.bytes_sent", TS_RECORDDATATYPE_INT, TS_STAT_NON_PERSISTENT, TS_STAT_SYNC_SUM);
   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;