Wait for CONNECT origin handshakes

src/proxy/http/HttpSM.cc

@@ -1140,11 +1140,37 @@ HttpSM::state_raw_http_server_open(int event, void *data)
   pending_action = nullptr;
   switch (event) {
   case NET_EVENT_OPEN: {
+    netvc = static_cast<NetVConnection *>(data);
+    if (plugin_tunnel_type == HttpPluginTunnel_t::NONE) {
+      _netvc             = netvc;
+      _netvc_read_buffer = new_MIOBuffer(HTTP_SERVER_RESP_HDR_BUFFER_INDEX);
+      _netvc_reader      = _netvc_read_buffer->alloc_reader();
+
+      // Wait for write readiness to verify that the nonblocking TCP connection
+      // completed before reporting a successful tunnel to the client.
+      _netvc->do_io_write(this, 1, _netvc_reader);
+      _netvc->set_inactivity_timeout(get_server_connect_timeout());
+      return 0;
+    }
+    [[fallthrough]];
+  }
+  case VC_EVENT_READ_COMPLETE:
+  case VC_EVENT_WRITE_READY:
+  case VC_EVENT_WRITE_COMPLETE: {
+    if (netvc == nullptr) {
+      netvc = _netvc;
+      netvc->do_io_write(nullptr, 0, nullptr);
+      free_MIOBuffer(_netvc_read_buffer);
+      _netvc             = nullptr;
+      _netvc_read_buffer = nullptr;
+      _netvc_reader      = nullptr;
+    }
+
     // Record the VC in our table
-    server_entry     = vc_table.new_entry();
-    server_entry->vc = netvc = static_cast<NetVConnection *>(data);
-    server_entry->vc_type    = HttpVC_t::RAW_SERVER_VC;
-    t_state.current.state    = HttpTransact::CONNECTION_ALIVE;
+    server_entry          = vc_table.new_entry();
+    server_entry->vc      = netvc;
+    server_entry->vc_type = HttpVC_t::RAW_SERVER_VC;
+    t_state.current.state = HttpTransact::CONNECTION_ALIVE;
     ats_ip_copy(&t_state.server_info.src_addr, netvc->get_local_addr());
 
     netvc->set_inactivity_timeout(get_server_inactivity_timeout());
@@ -1157,9 +1183,24 @@ HttpSM::state_raw_http_server_open(int event, void *data)
 
     break;
   }
+  case VC_EVENT_INACTIVITY_TIMEOUT:
+  case VC_EVENT_ACTIVE_TIMEOUT:
+    t_state.set_connect_fail(ETIMEDOUT);
+    [[fallthrough]];
   case VC_EVENT_ERROR:
   case VC_EVENT_EOS:
-  case NET_EVENT_OPEN_FAILED:
+  case NET_EVENT_OPEN_FAILED: {
+    if (_netvc != nullptr) {
+      if (event == VC_EVENT_ERROR || event == NET_EVENT_OPEN_FAILED) {
+        t_state.set_connect_fail(_netvc->lerrno);
+      }
+      _netvc->do_io_write(nullptr, 0, nullptr);
+      _netvc->do_io_close();
+      _netvc = nullptr;
+      free_MIOBuffer(_netvc_read_buffer);
+      _netvc_read_buffer = nullptr;
+      _netvc_reader      = nullptr;
+    }
     if (t_state.cause_of_death_errno == -UNKNOWN_INTERNAL_ERROR) {
       if (event == VC_EVENT_EOS) {
         t_state.set_connect_fail(EPIPE);
@@ -1171,6 +1212,7 @@ HttpSM::state_raw_http_server_open(int event, void *data)
     // use this value just to get around other values
     t_state.hdr_info.response_error = HttpTransact::ResponseError_t::STATUS_CODE_SERVER_ERROR;
     break;
+  }
   case EVENT_INTERVAL:
     // If we get EVENT_INTERNAL it means that we moved the transaction
     // to a different thread in do_http_server_open.  Since we didn't

tests/gold_tests/connect/connect_handshake.test.py

@@ -0,0 +1,62 @@
+'''Verify CONNECT waits for the origin TCP handshake.'''
+#  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.
+
+Test.Summary = 'Verify CONNECT waits for the origin TCP handshake.'
+
+
+class ConnectHandshakeTest:
+    '''Verify CONNECT failure is reported before establishing a tunnel.'''
+
+    replay_file: str = 'replays/connect_handshake.replay.yaml'
+
+    def __init__(self) -> None:
+        '''Configure the test run.'''
+        tr = Test.AddTestRun('CONNECT to a refused origin port')
+        self._configure_unavailable_origin()
+        self._configure_traffic_server(tr)
+        self._configure_client(tr)
+
+    def _configure_unavailable_origin(self) -> 'Process':
+        '''Reserve an origin port without starting a listening server.'''
+        origin = Test.MakeOriginServer('unavailable-origin')
+        self._origin = origin
+        return origin
+
+    def _configure_traffic_server(self, tr: 'TestRun') -> 'Process':
+        '''Configure Traffic Server as an explicit proxy.'''
+        ts = tr.MakeATSProcess('ts', enable_cache=False)
+        self._ts = ts
+
+        origin_port = self._origin.Variables.Port
+        ts.Disk.records_config.update(
+            {
+                'proxy.config.diags.debug.enabled': 1,
+                'proxy.config.diags.debug.tags': 'http|iocore_net',
+                'proxy.config.http.connect_ports': f'{origin_port}',
+            })
+        ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{origin_port}')
+        ts.addPrivateConnectAllowYaml()
+        return ts
+
+    def _configure_client(self, tr: 'TestRun') -> 'Process':
+        '''Configure a Proxy Verifier client that expects the refusal.'''
+        client = tr.AddVerifierClientProcess('client', self.replay_file, http_ports=[self._ts.Variables.port])
+        client.StartBefore(self._ts)
+        return client
+
+
+ConnectHandshakeTest()

tests/gold_tests/connect/replays/connect_handshake.replay.yaml

@@ -0,0 +1,34 @@
+#  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.
+
+meta:
+  version: '1.0'
+
+sessions:
+  - transactions:
+      - client-request:
+          method: CONNECT
+          version: '1.1'
+          url: www.example.com:443
+          headers:
+            fields:
+              - [Host, www.example.com:443]
+              - [uuid, connect-refused]
+
+        # ATS must report the refused origin connection rather than claim the
+        # tunnel was established.
+        proxy-response:
+          status: 502