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

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

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 <simone.bordet@gmail.com>
(cherry picked from commit 5eb7b70df7)
Signed-off-by: Simone Bordet <simone.bordet@gmail.com>

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

@@ -25,6 +25,7 @@ import org.eclipse.jetty.http.HttpHeaderValue;
 import org.eclipse.jetty.http.HttpMethod;
 import org.eclipse.jetty.http.HttpStatus;
 import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.http.MetaData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -95,6 +96,7 @@ public class HttpChannelOverHTTP extends HttpChannel
     {
         super.exchangeTerminated(exchange, result);
 
+        String method = exchange.getRequest().getMethod();
         Response response = result.getResponse();
         HttpFields responseHeaders = response.getHeaders();
 
@@ -113,7 +115,7 @@ public class HttpChannelOverHTTP extends HttpChannel
                 // 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";
             }
@@ -133,7 +135,8 @@ public class HttpChannelOverHTTP extends HttpChannel
         }
         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();
@@ -150,6 +153,11 @@ public class HttpChannelOverHTTP extends HttpChannel
         return outMessages.longValue();
     }
 
+    boolean isTunnel(String method, int status)
+    {
+        return MetaData.isTunnel(method, status);
+    }
+
     @Override
     public String toString()
     {

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

@@ -48,6 +48,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)
@@ -130,6 +131,10 @@ public class HttpReceiverOverHTTP extends HttpReceiver implements HttpParser.Res
 
     protected ByteBuffer onUpgradeFrom()
     {
+        RetainableByteBuffer networkBuffer = this.networkBuffer;
+        if (networkBuffer == null)
+            return null;
+
         ByteBuffer upgradeBuffer = null;
         if (networkBuffer.hasRemaining())
         {
@@ -226,14 +231,20 @@ public class HttpReceiverOverHTTP extends HttpReceiver implements HttpParser.Res
             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())
@@ -283,10 +294,9 @@ public class HttpReceiverOverHTTP extends HttpReceiver implements HttpParser.Res
         if (exchange == null)
             return;
 
+        this.method = exchange.getRequest().getMethod();
         this.status = status;
-        String method = exchange.getRequest().getMethod();
+        parser.setHeadResponse(HttpMethod.HEAD.is(method) || getHttpChannel().isTunnel(method, status));
-        parser.setHeadResponse(HttpMethod.HEAD.is(method) ||
-            (HttpMethod.CONNECT.is(method) && status == HttpStatus.OK_200));
         exchange.getResponse().version(version).status(status).reason(reason);
 
         responseBegin(exchange);

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

@@ -1535,8 +1535,18 @@ public class HttpClientTest extends AbstractHttpClientServerTest
 
                 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-http/src/main/java/org/eclipse/jetty/http/MetaData.java

@@ -19,6 +19,19 @@ import java.util.function.Supplier;
 
 public class MetaData implements Iterable<HttpField>
 {
+    /**
+     * <p>Returns whether the given HTTP request method and HTTP response status code
+     * identify a successful HTTP CONNECT tunnel.</p>
+     *
+     * @param method the HTTP request method
+     * @param status the HTTP response status code
+     * @return whether method and status identify a successful HTTP CONNECT tunnel
+     */
+    public static boolean isTunnel(String method, int status)
+    {
+        return HttpMethod.CONNECT.is(method) && HttpStatus.isSuccess(status);
+    }
+
     private final HttpVersion _httpVersion;
     private final HttpFields _fields;
     private final long _contentLength;

jetty-http2/http2-http-client-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpReceiverOverHTTP2.java

@@ -31,7 +31,6 @@ import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.client.api.Response;
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.http.HttpFields;
-import org.eclipse.jetty.http.HttpMethod;
 import org.eclipse.jetty.http.HttpStatus;
 import org.eclipse.jetty.http.MetaData;
 import org.eclipse.jetty.http2.ErrorCode;
@@ -102,7 +101,7 @@ public class HttpReceiverOverHTTP2 extends HttpReceiver implements HTTP2Channel.
                 }
 
                 HttpRequest httpRequest = exchange.getRequest();
-                if (HttpMethod.CONNECT.is(httpRequest.getMethod()) && httpResponse.getStatus() == HttpStatus.OK_200)
+                if (MetaData.isTunnel(httpRequest.getMethod(), httpResponse.getStatus()))
                 {
                     ClientHTTP2StreamEndPoint endPoint = new ClientHTTP2StreamEndPoint((IStream)stream);
                     long idleTimeout = httpRequest.getIdleTimeout();

jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HTTP2ServerConnection.java

@@ -28,7 +28,6 @@ import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.http.HttpMethod;
-import org.eclipse.jetty.http.HttpStatus;
 import org.eclipse.jetty.http.MetaData;
 import org.eclipse.jetty.http.MetaData.Request;
 import org.eclipse.jetty.http2.ErrorCode;
@@ -355,7 +354,7 @@ public class HTTP2ServerConnection extends HTTP2Connection
 
         private boolean isTunnel()
         {
-            return HttpMethod.CONNECT.is(getRequest().getMethod()) && getResponse().getStatus() == HttpStatus.OK_200;
+            return MetaData.isTunnel(getRequest().getMethod(), getResponse().getStatus());
         }
 
         @Override

jetty-http2/http2-server/src/main/java/org/eclipse/jetty/http2/server/HttpTransportOverHTTP2.java

@@ -261,7 +261,7 @@ public class HttpTransportOverHTTP2 implements HttpTransport
 
     private boolean isTunnel(MetaData.Request request, MetaData.Response response)
     {
-        return HttpMethod.CONNECT.is(request.getMethod()) && response.getStatus() == HttpStatus.OK_200;
+        return MetaData.isTunnel(request.getMethod(), response.getStatus());
     }
 
     @Override

jetty-http3/http3-server/src/main/java/org/eclipse/jetty/http3/server/internal/HttpTransportOverHTTP3.java

@@ -239,7 +239,7 @@ public class HttpTransportOverHTTP3 implements HttpTransport
 
     private boolean isTunnel(MetaData.Request request, MetaData.Response response)
     {
-        return HttpMethod.CONNECT.is(request.getMethod()) && response.getStatus() == HttpStatus.OK_200;
+        return MetaData.isTunnel(request.getMethod(), response.getStatus());
     }
 
     @Override

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

@@ -79,11 +79,12 @@ public abstract class ProxyConnection extends AbstractConnection
     @Override
     public String toConnectionString()
     {
+        EndPoint endPoint = getEndPoint();
         return String.format("%s@%x[l:%s<=>r:%s]",
             getClass().getSimpleName(),
             hashCode(),
-            getEndPoint().getLocalSocketAddress(),
+            endPoint.getLocalSocketAddress(),
-            getEndPoint().getRemoteSocketAddress());
+            endPoint.getRemoteSocketAddress());
     }
 
     private class ProxyIteratingCallback extends IteratingCallback

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

@@ -351,10 +351,10 @@ public class ForwardProxyTLSServerTest
                 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);
                 }
@@ -788,6 +788,47 @@ public class ForwardProxyTLSServerTest
         }
     }
 
+    @ParameterizedTest
+    @MethodSource("proxyTLS")
+    public void testRequestCompletionDelayed(SslContextFactory.Server proxyTLS) throws Exception
+    {
+        startTLSServer(new ServerHandler());
+        startProxy(proxyTLS);
+
+        HttpClient httpClient = newHttpClient();
+        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, StandardCharsets.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
@@ -823,6 +864,18 @@ public class ForwardProxyTLSServerTest
         }
     }
 
+    private static void sleep(long ms)
+    {
+        try
+        {
+            Thread.sleep(ms);
+        }
+        catch (InterruptedException x)
+        {
+            throw new RuntimeException(x);
+        }
+    }
+
     private static class ServerHandler extends AbstractHandler
     {
         @Override