Fixes #7348 - Slow CONNECT request causes NPE (#7349)

* Fixes #7348 - Slow CONNECT request causes NPE

Added NPE guard in `HttpReceiverOverHTTP.onUpgradeFrom()`.
Expanded logic in `HttpReceiverOverHTTP.parse()` to return true in case of CONNECT + 200.

Fixed `ProxyConnection.toConnectionString()` to avoid NPEs.

Fixed `HttpClientTest.testCONNECTWithHTTP10()` logic
after changes to fix this issue.

Now a tunneled connection is not put back into the connection pool,
and if applications explicitly want to use it, they must re-enable
fill interest, similarly to what should be done after upgrade+101.

Signed-off-by: Simone Bordet <[email protected]>

jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpChannelOverHTTP.java

@@ -136,6 +136,7 @@ public void exchangeTerminated(HttpExchange exchange, Result result)
     {
         super.exchangeTerminated(exchange, result);
 
+        String method = exchange.getRequest().getMethod();
         Response response = result.getResponse();
         HttpFields responseHeaders = response.getHeaders();
 
@@ -154,7 +155,7 @@ else if (sender.isShutdown())
                 // HTTP 1.0 must close the connection unless it has
                 // an explicit keep alive or it's a CONNECT method.
                 boolean keepAlive = responseHeaders.contains(HttpHeader.CONNECTION, HttpHeaderValue.KEEP_ALIVE.asString());
-                boolean connect = HttpMethod.CONNECT.is(exchange.getRequest().getMethod());
+                boolean connect = HttpMethod.CONNECT.is(method);
                 if (!keepAlive && !connect)
                     closeReason = "http/1.0";
             }
@@ -174,7 +175,8 @@ else if (sender.isShutdown())
         }
         else
         {
-            if (response.getStatus() == HttpStatus.SWITCHING_PROTOCOLS_101)
+            int status = response.getStatus();
+            if (status == HttpStatus.SWITCHING_PROTOCOLS_101 || isTunnel(method, status))
                 connection.remove();
             else
                 release();
@@ -191,6 +193,11 @@ protected long getMessagesOut()
         return outMessages.longValue();
     }
 
+    boolean isTunnel(String method, int status)
+    {
+        return HttpMethod.CONNECT.is(method) && HttpStatus.isSuccess(status);
+    }
+
     @Override
     public String toString()
     {

jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java

@@ -47,6 +47,7 @@ public class HttpReceiverOverHTTP extends HttpReceiver implements HttpParser.Res
     private boolean shutdown;
     private boolean complete;
     private boolean unsolicited;
+    private String method;
     private int status;
 
     public HttpReceiverOverHTTP(HttpChannelOverHTTP channel)
@@ -119,6 +120,10 @@ private void releaseNetworkBuffer()
 
     protected ByteBuffer onUpgradeFrom()
     {
+        RetainableByteBuffer networkBuffer = this.networkBuffer;
+        if (networkBuffer == null)
+            return null;
+
         ByteBuffer upgradeBuffer = null;
         if (networkBuffer.hasRemaining())
         {
@@ -127,7 +132,6 @@ protected ByteBuffer onUpgradeFrom()
             BufferUtil.put(networkBuffer.getBuffer(), upgradeBuffer);
             BufferUtil.flipToFlush(upgradeBuffer, 0);
         }
-
         releaseNetworkBuffer();
         return upgradeBuffer;
     }
@@ -215,14 +219,20 @@ private boolean parse()
             boolean complete = this.complete;
             this.complete = false;
             if (LOG.isDebugEnabled())
-                LOG.debug("Parse complete={}, remaining {} {}", complete, networkBuffer.remaining(), parser);
+                LOG.debug("Parse complete={}, {} {}", complete, networkBuffer, parser);
 
             if (complete)
             {
                 int status = this.status;
                 this.status = 0;
+                // Connection upgrade due to 101, bail out.
                 if (status == HttpStatus.SWITCHING_PROTOCOLS_101)
                     return true;
+                // Connection upgrade due to CONNECT + 200, bail out.
+                String method = this.method;
+                this.method = null;
+                if (getHttpChannel().isTunnel(method, status))
+                    return true;
             }
 
             if (networkBuffer.isEmpty())
@@ -279,10 +289,9 @@ public boolean startResponse(HttpVersion version, int status, String reason)
         if (exchange == null)
             return false;
 
+        this.method = exchange.getRequest().getMethod();
         this.status = status;
-        String method = exchange.getRequest().getMethod();
-        parser.setHeadResponse(HttpMethod.HEAD.is(method) ||
-            (HttpMethod.CONNECT.is(method) && status == HttpStatus.OK_200));
+        parser.setHeadResponse(HttpMethod.HEAD.is(method) || getHttpChannel().isTunnel(method, status));
         exchange.getResponse().version(version).status(status).reason(reason);
 
         return !responseBegin(exchange);

jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java

@@ -1606,8 +1606,18 @@ public void testCONNECTWithHTTP10(Scenario scenario) throws Exception
 
                 ContentResponse response = listener.get(5, TimeUnit.SECONDS);
                 assertEquals(200, response.getStatus());
+                assertThat(connection, Matchers.instanceOf(HttpConnectionOverHTTP.class));
+                HttpConnectionOverHTTP httpConnection = (HttpConnectionOverHTTP)connection;
+                EndPoint endPoint = httpConnection.getEndPoint();
+                assertTrue(endPoint.isOpen());
+
+                // After a CONNECT+200, this connection is in "tunnel mode",
+                // and applications that want to deal with tunnel bytes must
+                // likely access the underlying EndPoint.
+                // For the purpose of this test, we just re-enable fill interest
+                // so that we can send another clear-text HTTP request.
+                httpConnection.fillInterested();
 
-                // Test that I can send another request on the same connection.
                 request = client.newRequest(host, port);
                 listener = new FutureResponseListener(request);
                 connection.send(request, listener);

jetty-proxy/src/main/java/org/eclipse/jetty/proxy/ProxyConnection.java

@@ -79,11 +79,12 @@ public void onFillable()
     @Override
     public String toConnectionString()
     {
-        return String.format("%s@%x[l:%d<=>r:%d]",
+        EndPoint endPoint = getEndPoint();
+        return String.format("%s@%x[l:%s<=>r:%s]",
             getClass().getSimpleName(),
             hashCode(),
-            getEndPoint().getLocalAddress().getPort(),
-            getEndPoint().getRemoteAddress().getPort());
+            endPoint.getLocalAddress(),
+            endPoint.getRemoteAddress());
     }
 
     private class ProxyIteratingCallback extends IteratingCallback

jetty-proxy/src/test/java/org/eclipse/jetty/proxy/ForwardProxyTLSServerTest.java

@@ -344,10 +344,10 @@ protected void handleConnect(Request baseRequest, HttpServletRequest request, Ht
                 try
                 {
                     // Make sure the proxy remains idle enough.
-                    Thread.sleep(2 * idleTimeout);
+                    sleep(2 * idleTimeout);
                     super.handleConnect(baseRequest, request, response, serverAddress);
                 }
-                catch (InterruptedException x)
+                catch (Throwable x)
                 {
                     onConnectFailure(request, response, null, x);
                 }
@@ -789,6 +789,47 @@ public SSLEngine newSSLEngine(String host, int port)
         }
     }
 
+    @ParameterizedTest
+    @MethodSource("proxyTLS")
+    public void testRequestCompletionDelayed(SslContextFactory.Server proxyTLS) throws Exception
+    {
+        startTLSServer(new ServerHandler());
+        startProxy(proxyTLS);
+
+        HttpClient httpClient = new HttpClient(newClientSslContextFactory());
+        httpClient.getProxyConfiguration().getProxies().add(newHttpProxy());
+        httpClient.start();
+
+        try
+        {
+            httpClient.getRequestListeners().add(new org.eclipse.jetty.client.api.Request.Listener()
+            {
+                @Override
+                public void onSuccess(org.eclipse.jetty.client.api.Request request)
+                {
+                    if (HttpMethod.CONNECT.is(request.getMethod()))
+                        sleep(250);
+                }
+            });
+
+            String body = "BODY";
+            ContentResponse response = httpClient.newRequest("localhost", serverConnector.getLocalPort())
+                .scheme(HttpScheme.HTTPS.asString())
+                .method(HttpMethod.GET)
+                .path("/echo?body=" + URLEncoder.encode(body, "UTF-8"))
+                .timeout(5, TimeUnit.SECONDS)
+                .send();
+
+            assertEquals(HttpStatus.OK_200, response.getStatus());
+            String content = response.getContentAsString();
+            assertEquals(body, content);
+        }
+        finally
+        {
+            httpClient.stop();
+        }
+    }
+
     @Test
     @Tag("external")
     @Disabled
@@ -824,6 +865,18 @@ public void testExternalProxy() throws Exception
         }
     }
 
+    private static void sleep(long ms)
+    {
+        try
+        {
+            Thread.sleep(ms);
+        }
+        catch (InterruptedException x)
+        {
+            throw new RuntimeException(x);
+        }
+    }
+
     private static class ServerHandler extends AbstractHandler
     {
         @Override