8263031: HttpClient throws Exception if it receives a Push Promise that is too large

Reviewed-by: abakhtin, mbaesken
Backport-of: 4d2cd26ab5092ad0a169e4239164a869a4255bd3

Author: martinuy

src/java.net.http/share/classes/jdk/internal/net/http/Http2Connection.java

@@ -263,6 +263,8 @@ void markPrefaceSent() {
     private final Decoder hpackIn;
     final SettingsFrame clientSettings;
     private volatile SettingsFrame serverSettings;
+    private record PushContinuationState(HeaderDecoder pushContDecoder, PushPromiseFrame pushContFrame) {}
+    private volatile PushContinuationState pushContinuationState;
     private final String key; // for HttpClientImpl.connections map
     private final FramesDecoder framesDecoder;
     private final FramesEncoder framesEncoder = new FramesEncoder();
@@ -773,8 +775,8 @@ void processFrame(Http2Frame frame) throws IOException {
                 }
 
                 if (!(frame instanceof ResetFrame)) {
-                    if (frame instanceof DataFrame) {
-                        dropDataFrame((DataFrame)frame);
+                    if (frame instanceof DataFrame df) {
+                        dropDataFrame(df);
                     }
                     if (isServerInitiatedStream(streamid)) {
                         if (streamid < nextPushStream) {
@@ -791,26 +793,44 @@ void processFrame(Http2Frame frame) throws IOException {
                 }
                 return;
             }
-            if (frame instanceof PushPromiseFrame) {
-                PushPromiseFrame pp = (PushPromiseFrame)frame;
-                try {
-                    handlePushPromise(stream, pp);
-                } catch (UncheckedIOException e) {
-                    protocolError(ResetFrame.PROTOCOL_ERROR, e.getMessage());
-                    return;
-                }
-            } else if (frame instanceof HeaderFrame) {
-                // decode headers (or continuation)
-                try {
-                    decodeHeaders((HeaderFrame) frame, stream.rspHeadersConsumer());
-                } catch (UncheckedIOException e) {
-                    debug.log("Error decoding headers: " + e.getMessage(), e);
-                    protocolError(ResetFrame.PROTOCOL_ERROR, e.getMessage());
+
+            // While push frame is not null, the only acceptable frame on this
+            // stream is a Continuation frame
+            if (pushContinuationState != null) {
+                if (frame instanceof ContinuationFrame cf) {
+                    try {
+                        handlePushContinuation(stream, cf);
+                    } catch (UncheckedIOException e) {
+                        debug.log("Error handling Push Promise with Continuation: " + e.getMessage(), e);
+                        protocolError(ErrorFrame.PROTOCOL_ERROR, e.getMessage());
+                        return;
+                    }
+                } else {
+                    pushContinuationState = null;
+                    protocolError(ErrorFrame.PROTOCOL_ERROR, "Expected a Continuation frame but received " + frame);
                     return;
                 }
-                stream.incoming(frame);
             } else {
-                stream.incoming(frame);
+                if (frame instanceof PushPromiseFrame pp) {
+                    try {
+                        handlePushPromise(stream, pp);
+                    } catch (UncheckedIOException e) {
+                        protocolError(ErrorFrame.PROTOCOL_ERROR, e.getMessage());
+                        return;
+                    }
+                } else if (frame instanceof HeaderFrame hf) {
+                    // decode headers
+                    try {
+                        decodeHeaders(hf, stream.rspHeadersConsumer());
+                    } catch (UncheckedIOException e) {
+                        debug.log("Error decoding headers: " + e.getMessage(), e);
+                        protocolError(ErrorFrame.PROTOCOL_ERROR, e.getMessage());
+                        return;
+                    }
+                    stream.incoming(frame);
+                } else {
+                    stream.incoming(frame);
+                }
             }
         }
     }
@@ -841,11 +861,34 @@ private <T> void handlePushPromise(Stream<T> parent, PushPromiseFrame pp)
     {
         // always decode the headers as they may affect connection-level HPACK
         // decoding state
+        assert pushContinuationState == null;
         HeaderDecoder decoder = new HeaderDecoder();
         decodeHeaders(pp, decoder);
+        int promisedStreamid = pp.getPromisedStream();
+        if (pp.endHeaders()) {
+            completePushPromise(promisedStreamid, parent, decoder.headers());
+        } else {
+            pushContinuationState = new PushContinuationState(decoder, pp);
+        }
+    }
+
+    private <T> void handlePushContinuation(Stream<T> parent, ContinuationFrame cf)
+            throws IOException {
+        var pcs = pushContinuationState;
+        decodeHeaders(cf, pcs.pushContDecoder);
+        // if all continuations are sent, set pushWithContinuation to null
+        if (cf.endHeaders()) {
+            completePushPromise(pcs.pushContFrame.getPromisedStream(), parent,
+                    pcs.pushContDecoder.headers());
+            pushContinuationState = null;
+        }
+    }
 
+    private <T> void completePushPromise(int promisedStreamid, Stream<T> parent, HttpHeaders headers)
+            throws IOException {
+        // Perhaps the following checks could be moved to handlePushPromise()
+        // to reset the PushPromise stream earlier?
         HttpRequestImpl parentReq = parent.request;
-        int promisedStreamid = pp.getPromisedStream();
         if (promisedStreamid != nextPushStream) {
             resetStream(promisedStreamid, ResetFrame.PROTOCOL_ERROR);
             return;
@@ -856,7 +899,6 @@ private <T> void handlePushPromise(Stream<T> parent, PushPromiseFrame pp)
             nextPushStream += 2;
         }
 
-        HttpHeaders headers = decoder.headers();
         HttpRequestImpl pushReq = HttpRequestImpl.createPushRequest(parentReq, headers);
         Exchange<T> pushExch = new Exchange<>(pushReq, parent.exchange.multi);
         Stream.PushedStream<T> pushStream = createPushStream(parent, pushExch);