View Javadoc
1   /*
2    * Copyright 2014 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
5    * "License"); you may not use this file except in compliance with the License. You may obtain a
6    * copy of the License at:
7    *
8    * https://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software distributed under the License
11   * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
12   * or implied. See the License for the specific language governing permissions and limitations under
13   * the License.
14   */
15  package io.netty.handler.codec.http2;
16  
17  import io.netty.buffer.ByteBuf;
18  import io.netty.channel.ChannelHandlerContext;
19  import io.netty.handler.codec.http.HttpHeaderNames;
20  import io.netty.handler.codec.http.HttpStatusClass;
21  import io.netty.handler.codec.http.HttpUtil;
22  import io.netty.handler.codec.http2.Http2Connection.Endpoint;
23  import io.netty.util.internal.logging.InternalLogger;
24  import io.netty.util.internal.logging.InternalLoggerFactory;
25  
26  import java.util.Iterator;
27  import java.util.List;
28  import java.util.Map.Entry;
29  
30  import static io.netty.handler.codec.http.HttpStatusClass.INFORMATIONAL;
31  import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
32  import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
33  import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
34  import static io.netty.handler.codec.http2.Http2Error.STREAM_CLOSED;
35  import static io.netty.handler.codec.http2.Http2Exception.connectionError;
36  import static io.netty.handler.codec.http2.Http2Exception.streamError;
37  import static io.netty.handler.codec.http2.Http2PromisedRequestVerifier.ALWAYS_VERIFY;
38  import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
39  import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
40  import static io.netty.util.internal.ObjectUtil.checkNotNull;
41  import static java.lang.Integer.MAX_VALUE;
42  import static java.lang.Math.min;
43  
44  /**
45   * Provides the default implementation for processing inbound frame events and delegates to a
46   * {@link Http2FrameListener}
47   * <p>
48   * This class will read HTTP/2 frames and delegate the events to a {@link Http2FrameListener}
49   * <p>
50   * This interface enforces inbound flow control functionality through
51   * {@link Http2LocalFlowController}
52   */
53  public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
54      private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2ConnectionDecoder.class);
55      private Http2FrameListener internalFrameListener = new PrefaceFrameListener();
56      private final Http2Connection connection;
57      private Http2LifecycleManager lifecycleManager;
58      private final Http2ConnectionEncoder encoder;
59      private final Http2FrameReader frameReader;
60      private Http2FrameListener listener;
61      private final Http2PromisedRequestVerifier requestVerifier;
62      private final Http2SettingsReceivedConsumer settingsReceivedConsumer;
63      private final boolean autoAckPing;
64      private final Http2Connection.PropertyKey contentLengthKey;
65      private final boolean validateHeaders;
66  
67      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
68                                           Http2ConnectionEncoder encoder,
69                                           Http2FrameReader frameReader) {
70          this(connection, encoder, frameReader, ALWAYS_VERIFY);
71      }
72  
73      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
74                                           Http2ConnectionEncoder encoder,
75                                           Http2FrameReader frameReader,
76                                           Http2PromisedRequestVerifier requestVerifier) {
77          this(connection, encoder, frameReader, requestVerifier, true);
78      }
79  
80      /**
81       * Create a new instance.
82       * @param connection The {@link Http2Connection} associated with this decoder.
83       * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
84       * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
85       *                    h2 semantics on top of the frames.
86       * @param requestVerifier Determines if push promised streams are valid.
87       * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
88       *  The {@code Http2ConnectionEncoder} is expected to be an instance of {@link Http2SettingsReceivedConsumer} and
89       *  will apply the earliest received but not yet ACKed SETTINGS when writing the SETTINGS ACKs.
90       * {@code true} to enable automatically applying and sending settings acknowledge frame.
91       */
92      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
93                                           Http2ConnectionEncoder encoder,
94                                           Http2FrameReader frameReader,
95                                           Http2PromisedRequestVerifier requestVerifier,
96                                           boolean autoAckSettings) {
97          this(connection, encoder, frameReader, requestVerifier, autoAckSettings, true);
98      }
99  
100     @Deprecated
101     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
102                                          Http2ConnectionEncoder encoder,
103                                          Http2FrameReader frameReader,
104                                          Http2PromisedRequestVerifier requestVerifier,
105                                          boolean autoAckSettings,
106                                          boolean autoAckPing) {
107         this(connection, encoder, frameReader, requestVerifier, autoAckSettings, autoAckPing, true);
108     }
109 
110     /**
111      * Create a new instance.
112      * @param connection The {@link Http2Connection} associated with this decoder.
113      * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
114      * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
115      *                    h2 semantics on top of the frames.
116      * @param requestVerifier Determines if push promised streams are valid.
117      * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
118      *                        The {@code Http2ConnectionEncoder} is expected to be an instance of
119      *                        {@link Http2SettingsReceivedConsumer} and will apply the earliest received but not yet
120      *                        ACKed SETTINGS when writing the SETTINGS ACKs. {@code true} to enable automatically
121      *                        applying and sending settings acknowledge frame.
122      * @param autoAckPing {@code false} to disable automatically sending ping acknowledge frame. {@code true} to enable
123      *                    automatically sending ping ack frame.
124      */
125     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
126                                          Http2ConnectionEncoder encoder,
127                                          Http2FrameReader frameReader,
128                                          Http2PromisedRequestVerifier requestVerifier,
129                                          boolean autoAckSettings,
130                                          boolean autoAckPing,
131                                          boolean validateHeaders) {
132         this.validateHeaders = validateHeaders;
133         this.autoAckPing = autoAckPing;
134         if (autoAckSettings) {
135             settingsReceivedConsumer = null;
136         } else {
137             if (!(encoder instanceof Http2SettingsReceivedConsumer)) {
138                 throw new IllegalArgumentException("disabling autoAckSettings requires the encoder to be a " +
139                         Http2SettingsReceivedConsumer.class);
140             }
141             settingsReceivedConsumer = (Http2SettingsReceivedConsumer) encoder;
142         }
143         this.connection = checkNotNull(connection, "connection");
144         contentLengthKey = this.connection.newKey();
145         this.frameReader = checkNotNull(frameReader, "frameReader");
146         this.encoder = checkNotNull(encoder, "encoder");
147         this.requestVerifier = checkNotNull(requestVerifier, "requestVerifier");
148         if (connection.local().flowController() == null) {
149             connection.local().flowController(new DefaultHttp2LocalFlowController(connection));
150         }
151         connection.local().flowController().frameWriter(encoder.frameWriter());
152     }
153 
154     @Override
155     public void lifecycleManager(Http2LifecycleManager lifecycleManager) {
156         this.lifecycleManager = checkNotNull(lifecycleManager, "lifecycleManager");
157     }
158 
159     @Override
160     public Http2Connection connection() {
161         return connection;
162     }
163 
164     @Override
165     public final Http2LocalFlowController flowController() {
166         return connection.local().flowController();
167     }
168 
169     @Override
170     public void frameListener(Http2FrameListener listener) {
171         this.listener = checkNotNull(listener, "listener");
172     }
173 
174     @Override
175     public Http2FrameListener frameListener() {
176         return listener;
177     }
178 
179     @Override
180     public boolean prefaceReceived() {
181         return FrameReadListener.class == internalFrameListener.getClass();
182     }
183 
184     @Override
185     public void decodeFrame(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Http2Exception {
186         frameReader.readFrame(ctx, in, internalFrameListener);
187     }
188 
189     @Override
190     public Http2Settings localSettings() {
191         Http2Settings settings = new Http2Settings();
192         Http2FrameReader.Configuration config = frameReader.configuration();
193         Http2HeadersDecoder.Configuration headersConfig = config.headersConfiguration();
194         Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
195         settings.initialWindowSize(flowController().initialWindowSize());
196         settings.maxConcurrentStreams(connection.remote().maxActiveStreams());
197         settings.headerTableSize(headersConfig.maxHeaderTableSize());
198         settings.maxFrameSize(frameSizePolicy.maxFrameSize());
199         settings.maxHeaderListSize(headersConfig.maxHeaderListSize());
200         if (!connection.isServer()) {
201             // Only set the pushEnabled flag if this is a client endpoint.
202             settings.pushEnabled(connection.local().allowPushTo());
203         }
204         return settings;
205     }
206 
207     @Override
208     public void close() {
209         frameReader.close();
210     }
211 
212     /**
213      * Calculate the threshold in bytes which should trigger a {@code GO_AWAY} if a set of headers exceeds this amount.
214      * @param maxHeaderListSize
215      *      <a href="https://tools.ietf.org/html/rfc7540#section-6.5.2">SETTINGS_MAX_HEADER_LIST_SIZE</a> for the local
216      *      endpoint.
217      * @return the threshold in bytes which should trigger a {@code GO_AWAY} if a set of headers exceeds this amount.
218      */
219     protected long calculateMaxHeaderListSizeGoAway(long maxHeaderListSize) {
220         return Http2CodecUtil.calculateMaxHeaderListSizeGoAway(maxHeaderListSize);
221     }
222 
223     private int unconsumedBytes(Http2Stream stream) {
224         return flowController().unconsumedBytes(stream);
225     }
226 
227     void onGoAwayRead0(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
228             throws Http2Exception {
229         listener.onGoAwayRead(ctx, lastStreamId, errorCode, debugData);
230         connection.goAwayReceived(lastStreamId, errorCode, debugData);
231     }
232 
233     // See https://tools.ietf.org/html/rfc7540#section-8.1.2.6
234     private void verifyContentLength(Http2Stream stream, int data, boolean isEnd) throws Http2Exception {
235         ContentLength contentLength = stream.getProperty(contentLengthKey);
236         if (contentLength != null) {
237             try {
238                 contentLength.increaseReceivedBytes(connection.isServer(), stream.id(), data, isEnd);
239             } finally {
240                 if (isEnd) {
241                     stream.removeProperty(contentLengthKey);
242                 }
243             }
244         }
245     }
246 
247     /**
248      * Handles all inbound frames from the network.
249      */
250     private final class FrameReadListener implements Http2FrameListener {
251         @Override
252         public int onDataRead(final ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
253                               boolean endOfStream) throws Http2Exception {
254             Http2Stream stream = connection.stream(streamId);
255             Http2LocalFlowController flowController = flowController();
256             int readable = data.readableBytes();
257             int bytesToReturn = readable + padding;
258 
259             final boolean shouldIgnore;
260             try {
261                 shouldIgnore = shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "DATA");
262             } catch (Http2Exception e) {
263                 // Ignoring this frame. We still need to count the frame towards the connection flow control
264                 // window, but we immediately mark all bytes as consumed.
265                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
266                 flowController.consumeBytes(stream, bytesToReturn);
267                 throw e;
268             } catch (Throwable t) {
269                 throw connectionError(INTERNAL_ERROR, t, "Unhandled error on data stream id %d", streamId);
270             }
271 
272             if (shouldIgnore) {
273                 // Ignoring this frame. We still need to count the frame towards the connection flow control
274                 // window, but we immediately mark all bytes as consumed.
275                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
276                 flowController.consumeBytes(stream, bytesToReturn);
277 
278                 // Verify that the stream may have existed after we apply flow control.
279                 verifyStreamMayHaveExisted(streamId, endOfStream, "DATA");
280 
281                 // All bytes have been consumed.
282                 return bytesToReturn;
283             }
284             Http2Exception error = null;
285             switch (stream.state()) {
286                 case OPEN:
287                 case HALF_CLOSED_LOCAL:
288                     break;
289                 case HALF_CLOSED_REMOTE:
290                 case CLOSED:
291                     error = streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
292                         stream.id(), stream.state());
293                     break;
294                 default:
295                     error = streamError(stream.id(), PROTOCOL_ERROR,
296                         "Stream %d in unexpected state: %s", stream.id(), stream.state());
297                     break;
298             }
299 
300             int unconsumedBytes = unconsumedBytes(stream);
301             try {
302                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
303                 // Update the unconsumed bytes after flow control is applied.
304                 unconsumedBytes = unconsumedBytes(stream);
305 
306                 // If the stream is in an invalid state to receive the frame, throw the error.
307                 if (error != null) {
308                     throw error;
309                 }
310 
311                 verifyContentLength(stream, readable, endOfStream);
312 
313                 // Call back the application and retrieve the number of bytes that have been
314                 // immediately processed.
315                 bytesToReturn = listener.onDataRead(ctx, streamId, data, padding, endOfStream);
316 
317                 if (endOfStream) {
318                     lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
319                 }
320 
321                 return bytesToReturn;
322             } catch (Http2Exception e) {
323                 // If an exception happened during delivery, the listener may have returned part
324                 // of the bytes before the error occurred. If that's the case, subtract that from
325                 // the total processed bytes so that we don't return too many bytes.
326                 int delta = unconsumedBytes - unconsumedBytes(stream);
327                 bytesToReturn -= delta;
328                 throw e;
329             } catch (RuntimeException e) {
330                 // If an exception happened during delivery, the listener may have returned part
331                 // of the bytes before the error occurred. If that's the case, subtract that from
332                 // the total processed bytes so that we don't return too many bytes.
333                 int delta = unconsumedBytes - unconsumedBytes(stream);
334                 bytesToReturn -= delta;
335                 throw e;
336             } finally {
337                 // If appropriate, return the processed bytes to the flow controller.
338                 flowController.consumeBytes(stream, bytesToReturn);
339             }
340         }
341 
342         @Override
343         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
344                 boolean endOfStream) throws Http2Exception {
345             onHeadersRead(ctx, streamId, headers, 0, DEFAULT_PRIORITY_WEIGHT, false, padding, endOfStream);
346         }
347 
348         @Override
349         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
350                 short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
351             Http2Stream stream = connection.stream(streamId);
352             boolean allowHalfClosedRemote = false;
353             boolean isTrailers = false;
354             if (stream == null && !connection.streamMayHaveExisted(streamId)) {
355                 stream = connection.remote().createStream(streamId, endOfStream);
356                 // Allow the state to be HALF_CLOSE_REMOTE if we're creating it in that state.
357                 allowHalfClosedRemote = stream.state() == HALF_CLOSED_REMOTE;
358             } else if (stream != null) {
359                 isTrailers = stream.isHeadersReceived();
360             }
361 
362             if (shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "HEADERS")) {
363                 return;
364             }
365 
366             boolean isInformational = !connection.isServer() &&
367                     HttpStatusClass.valueOf(headers.status()) == INFORMATIONAL;
368             if ((isInformational || !endOfStream) && stream.isHeadersReceived() || stream.isTrailersReceived()) {
369                 throw streamError(streamId, PROTOCOL_ERROR,
370                                   "Stream %d received too many headers EOS: %s state: %s",
371                                   streamId, endOfStream, stream.state());
372             }
373 
374             switch (stream.state()) {
375                 case RESERVED_REMOTE:
376                     stream.open(endOfStream);
377                     break;
378                 case OPEN:
379                 case HALF_CLOSED_LOCAL:
380                     // Allowed to receive headers in these states.
381                     break;
382                 case HALF_CLOSED_REMOTE:
383                     if (!allowHalfClosedRemote) {
384                         throw streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
385                                 stream.id(), stream.state());
386                     }
387                     break;
388                 case CLOSED:
389                     throw streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
390                             stream.id(), stream.state());
391                 default:
392                     // Connection error.
393                     throw connectionError(PROTOCOL_ERROR, "Stream %d in unexpected state: %s", stream.id(),
394                             stream.state());
395             }
396 
397             if (!isTrailers) {
398                 // extract the content-length header
399                 List<? extends CharSequence> contentLength = headers.getAll(HttpHeaderNames.CONTENT_LENGTH);
400                 if (contentLength != null && !contentLength.isEmpty()) {
401                     try {
402                         long cLength = HttpUtil.normalizeAndGetContentLength(contentLength, false, true);
403                         if (cLength != -1) {
404                             headers.setLong(HttpHeaderNames.CONTENT_LENGTH, cLength);
405                             stream.setProperty(contentLengthKey, new ContentLength(cLength));
406                         }
407                     } catch (IllegalArgumentException e) {
408                         throw streamError(stream.id(), PROTOCOL_ERROR, e,
409                                 "Multiple content-length headers received");
410                     }
411                 }
412                 // Use size() instead of isEmpty() for backward compatibility with grpc-java prior to 1.59.1,
413                 // see https://github.com/grpc/grpc-java/issues/10665
414             } else if (validateHeaders && headers.size() > 0) {
415                 // Need to check trailers don't contain pseudo headers. According to RFC 9113
416                 // Trailers MUST NOT include pseudo-header fields (Section 8.3).
417                 for (Iterator<Entry<CharSequence, CharSequence>> iterator =
418                     headers.iterator(); iterator.hasNext();) {
419                     CharSequence name = iterator.next().getKey();
420                     if (Http2Headers.PseudoHeaderName.hasPseudoHeaderFormat(name)) {
421                         throw streamError(stream.id(), PROTOCOL_ERROR,
422                                 "Found invalid Pseudo-Header in trailers: %s", name);
423                     }
424                 }
425             }
426 
427             stream.headersReceived(isInformational);
428             verifyContentLength(stream, 0, endOfStream);
429             encoder.flowController().updateDependencyTree(streamId, streamDependency, weight, exclusive);
430             listener.onHeadersRead(ctx, streamId, headers, streamDependency,
431                     weight, exclusive, padding, endOfStream);
432             // If the headers completes this stream, close it.
433             if (endOfStream) {
434                 lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
435             }
436         }
437 
438         @Override
439         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
440                 boolean exclusive) throws Http2Exception {
441             encoder.flowController().updateDependencyTree(streamId, streamDependency, weight, exclusive);
442 
443             listener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
444         }
445 
446         @Override
447         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
448             Http2Stream stream = connection.stream(streamId);
449             if (stream == null) {
450                 verifyStreamMayHaveExisted(streamId, false, "RST_STREAM");
451                 return;
452             }
453 
454             switch(stream.state()) {
455             case IDLE:
456                 throw connectionError(PROTOCOL_ERROR, "RST_STREAM received for IDLE stream %d", streamId);
457             case CLOSED:
458                 return; // RST_STREAM frames must be ignored for closed streams.
459             default:
460                 break;
461             }
462 
463             listener.onRstStreamRead(ctx, streamId, errorCode);
464 
465             lifecycleManager.closeStream(stream, ctx.newSucceededFuture());
466         }
467 
468         @Override
469         public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
470             // Apply oldest outstanding local settings here. This is a synchronization point between endpoints.
471             Http2Settings settings = encoder.pollSentSettings();
472 
473             if (settings != null) {
474                 applyLocalSettings(settings);
475             }
476 
477             listener.onSettingsAckRead(ctx);
478         }
479 
480         /**
481          * Applies settings sent from the local endpoint.
482          * <p>
483          * This method is only called after the local settings have been acknowledged from the remote endpoint.
484          */
485         private void applyLocalSettings(Http2Settings settings) throws Http2Exception {
486             Boolean pushEnabled = settings.pushEnabled();
487             final Http2FrameReader.Configuration config = frameReader.configuration();
488             final Http2HeadersDecoder.Configuration headerConfig = config.headersConfiguration();
489             final Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
490             if (pushEnabled != null) {
491                 if (connection.isServer()) {
492                     throw connectionError(PROTOCOL_ERROR, "Server sending SETTINGS frame with ENABLE_PUSH specified");
493                 }
494                 connection.local().allowPushTo(pushEnabled);
495             }
496 
497             Long maxConcurrentStreams = settings.maxConcurrentStreams();
498             if (maxConcurrentStreams != null) {
499                 connection.remote().maxActiveStreams((int) min(maxConcurrentStreams, MAX_VALUE));
500             }
501 
502             Long headerTableSize = settings.headerTableSize();
503             if (headerTableSize != null) {
504                 headerConfig.maxHeaderTableSize(headerTableSize);
505             }
506 
507             Long maxHeaderListSize = settings.maxHeaderListSize();
508             if (maxHeaderListSize != null) {
509                 headerConfig.maxHeaderListSize(maxHeaderListSize, calculateMaxHeaderListSizeGoAway(maxHeaderListSize));
510             }
511 
512             Integer maxFrameSize = settings.maxFrameSize();
513             if (maxFrameSize != null) {
514                 frameSizePolicy.maxFrameSize(maxFrameSize);
515             }
516 
517             Integer initialWindowSize = settings.initialWindowSize();
518             if (initialWindowSize != null) {
519                 flowController().initialWindowSize(initialWindowSize);
520             }
521         }
522 
523         @Override
524         public void onSettingsRead(final ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
525             if (settingsReceivedConsumer == null) {
526                 // Acknowledge receipt of the settings. We should do this before we process the settings to ensure our
527                 // remote peer applies these settings before any subsequent frames that we may send which depend upon
528                 // these new settings. See https://github.com/netty/netty/issues/6520.
529                 encoder.writeSettingsAck(ctx, ctx.newPromise());
530 
531                 encoder.remoteSettings(settings);
532             } else {
533                 settingsReceivedConsumer.consumeReceivedSettings(settings);
534             }
535 
536             listener.onSettingsRead(ctx, settings);
537         }
538 
539         @Override
540         public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
541             if (autoAckPing) {
542                 // Send an ack back to the remote client.
543                 encoder.writePing(ctx, true, data, ctx.newPromise());
544             }
545             listener.onPingRead(ctx, data);
546         }
547 
548         @Override
549         public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
550             listener.onPingAckRead(ctx, data);
551         }
552 
553         @Override
554         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
555                 Http2Headers headers, int padding) throws Http2Exception {
556             // A client cannot push.
557             if (connection().isServer()) {
558                 throw connectionError(PROTOCOL_ERROR, "A client cannot push.");
559             }
560 
561             Http2Stream parentStream = connection.stream(streamId);
562 
563             if (shouldIgnoreHeadersOrDataFrame(ctx, streamId, parentStream, false, "PUSH_PROMISE")) {
564                 return;
565             }
566 
567             switch (parentStream.state()) {
568               case OPEN:
569               case HALF_CLOSED_LOCAL:
570                   // Allowed to receive push promise in these states.
571                   break;
572               default:
573                   // Connection error.
574                   throw connectionError(PROTOCOL_ERROR,
575                       "Stream %d in unexpected state for receiving push promise: %s",
576                       parentStream.id(), parentStream.state());
577             }
578 
579             if (!requestVerifier.isAuthoritative(ctx, headers)) {
580                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
581                         "Promised request on stream %d for promised stream %d is not authoritative",
582                         streamId, promisedStreamId);
583             }
584             if (!requestVerifier.isCacheable(headers)) {
585                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
586                         "Promised request on stream %d for promised stream %d is not known to be cacheable",
587                         streamId, promisedStreamId);
588             }
589             if (!requestVerifier.isSafe(headers)) {
590                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
591                         "Promised request on stream %d for promised stream %d is not known to be safe",
592                         streamId, promisedStreamId);
593             }
594 
595             // Reserve the push stream based with a priority based on the current stream's priority.
596             connection.remote().reservePushStream(promisedStreamId, parentStream);
597 
598             listener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
599         }
600 
601         @Override
602         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
603                 throws Http2Exception {
604             onGoAwayRead0(ctx, lastStreamId, errorCode, debugData);
605         }
606 
607         @Override
608         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
609                 throws Http2Exception {
610             Http2Stream stream = connection.stream(streamId);
611             if (stream == null || stream.state() == CLOSED || streamCreatedAfterGoAwaySent(streamId)) {
612                 // Ignore this frame.
613                 verifyStreamMayHaveExisted(streamId, false, "WINDOW_UPDATE");
614                 return;
615             }
616 
617             // Update the outbound flow control window.
618             encoder.flowController().incrementWindowSize(stream, windowSizeIncrement);
619 
620             listener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
621         }
622 
623         @Override
624         public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
625                 ByteBuf payload) throws Http2Exception {
626             Http2Stream stream = connection.stream(streamId);
627             if (stream == null) {
628                 return;
629             }
630 
631             listener.onUnknownFrame(ctx, frameType, streamId, flags, payload);
632         }
633 
634         /**
635          * Helper method to determine if a frame that has the semantics of headers or data should be ignored for the
636          * {@code stream} (which may be {@code null}) associated with {@code streamId}.
637          */
638         private boolean shouldIgnoreHeadersOrDataFrame(ChannelHandlerContext ctx, int streamId, Http2Stream stream,
639                 boolean endOfStream, String frameName) throws Http2Exception {
640             if (stream == null) {
641                 if (streamCreatedAfterGoAwaySent(streamId)) {
642                     logger.info("{} ignoring {} frame for stream {}. Stream sent after GOAWAY sent",
643                             ctx.channel(), frameName, streamId);
644                     return true;
645                 }
646 
647                 // Make sure it's not an out-of-order frame, like a rogue DATA frame, for a stream that could
648                 // never have existed.
649                 verifyStreamMayHaveExisted(streamId, endOfStream, frameName);
650 
651                 // Its possible that this frame would result in stream ID out of order creation (PROTOCOL ERROR) and its
652                 // also possible that this frame is received on a CLOSED stream (STREAM_CLOSED after a RST_STREAM is
653                 // sent). We don't have enough information to know for sure, so we choose the lesser of the two errors.
654                 throw streamError(streamId, STREAM_CLOSED, "Received %s frame for an unknown stream %d",
655                                   frameName, streamId);
656             }
657             if (stream.isResetSent() || streamCreatedAfterGoAwaySent(streamId)) {
658                 // If we have sent a reset stream it is assumed the stream will be closed after the write completes.
659                 // If we have not sent a reset, but the stream was created after a GoAway this is not supported by
660                 // DefaultHttp2Connection and if a custom Http2Connection is used it is assumed the lifetime is managed
661                 // elsewhere so we don't close the stream or otherwise modify the stream's state.
662 
663                 if (logger.isInfoEnabled()) {
664                     logger.info("{} ignoring {} frame for stream {}", ctx.channel(), frameName,
665                             stream.isResetSent() ? "RST_STREAM sent." :
666                                     "Stream created after GOAWAY sent. Last known stream by peer " +
667                                      connection.remote().lastStreamKnownByPeer());
668                 }
669 
670                 return true;
671             }
672             return false;
673         }
674 
675         /**
676          * Helper method for determining whether or not to ignore inbound frames. A stream is considered to be created
677          * after a {@code GOAWAY} is sent if the following conditions hold:
678          * <p/>
679          * <ul>
680          *     <li>A {@code GOAWAY} must have been sent by the local endpoint</li>
681          *     <li>The {@code streamId} must identify a legitimate stream id for the remote endpoint to be creating</li>
682          *     <li>{@code streamId} is greater than the Last Known Stream ID which was sent by the local endpoint
683          *     in the last {@code GOAWAY} frame</li>
684          * </ul>
685          * <p/>
686          */
687         private boolean streamCreatedAfterGoAwaySent(int streamId) {
688             Endpoint<?> remote = connection.remote();
689             return connection.goAwaySent() && remote.isValidStreamId(streamId) &&
690                     streamId > remote.lastStreamKnownByPeer();
691         }
692 
693         private void verifyStreamMayHaveExisted(int streamId, boolean endOfStream, String frameName)
694                 throws Http2Exception {
695             if (!connection.streamMayHaveExisted(streamId)) {
696                 throw connectionError(PROTOCOL_ERROR,
697                         "Stream %d does not exist for inbound frame %s, endOfStream = %b",
698                         streamId, frameName, endOfStream);
699             }
700         }
701     }
702 
703     private final class PrefaceFrameListener implements Http2FrameListener {
704         /**
705          * Verifies that the HTTP/2 connection preface has been received from the remote endpoint.
706          * It is possible that the current call to
707          * {@link Http2FrameReader#readFrame(ChannelHandlerContext, ByteBuf, Http2FrameListener)} will have multiple
708          * frames to dispatch. So it may be OK for this class to get legitimate frames for the first readFrame.
709          */
710         private void verifyPrefaceReceived() throws Http2Exception {
711             if (!prefaceReceived()) {
712                 throw connectionError(PROTOCOL_ERROR, "Received non-SETTINGS as first frame.");
713             }
714         }
715 
716         @Override
717         public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
718                 throws Http2Exception {
719             verifyPrefaceReceived();
720             return internalFrameListener.onDataRead(ctx, streamId, data, padding, endOfStream);
721         }
722 
723         @Override
724         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
725                 boolean endOfStream) throws Http2Exception {
726             verifyPrefaceReceived();
727             internalFrameListener.onHeadersRead(ctx, streamId, headers, padding, endOfStream);
728         }
729 
730         @Override
731         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
732                 short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
733             verifyPrefaceReceived();
734             internalFrameListener.onHeadersRead(ctx, streamId, headers, streamDependency, weight,
735                     exclusive, padding, endOfStream);
736         }
737 
738         @Override
739         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
740                 boolean exclusive) throws Http2Exception {
741             verifyPrefaceReceived();
742             internalFrameListener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
743         }
744 
745         @Override
746         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
747             verifyPrefaceReceived();
748             internalFrameListener.onRstStreamRead(ctx, streamId, errorCode);
749         }
750 
751         @Override
752         public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
753             verifyPrefaceReceived();
754             internalFrameListener.onSettingsAckRead(ctx);
755         }
756 
757         @Override
758         public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
759             // The first settings should change the internalFrameListener to the "real" listener
760             // that expects the preface to be verified.
761             if (!prefaceReceived()) {
762                 internalFrameListener = new FrameReadListener();
763             }
764             internalFrameListener.onSettingsRead(ctx, settings);
765         }
766 
767         @Override
768         public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
769             verifyPrefaceReceived();
770             internalFrameListener.onPingRead(ctx, data);
771         }
772 
773         @Override
774         public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
775             verifyPrefaceReceived();
776             internalFrameListener.onPingAckRead(ctx, data);
777         }
778 
779         @Override
780         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
781                 Http2Headers headers, int padding) throws Http2Exception {
782             verifyPrefaceReceived();
783             internalFrameListener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
784         }
785 
786         @Override
787         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
788                 throws Http2Exception {
789             onGoAwayRead0(ctx, lastStreamId, errorCode, debugData);
790         }
791 
792         @Override
793         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
794                 throws Http2Exception {
795             verifyPrefaceReceived();
796             internalFrameListener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
797         }
798 
799         @Override
800         public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
801                 ByteBuf payload) throws Http2Exception {
802             verifyPrefaceReceived();
803             internalFrameListener.onUnknownFrame(ctx, frameType, streamId, flags, payload);
804         }
805     }
806 
807     private static final class ContentLength {
808         private final long expected;
809         private long seen;
810 
811         ContentLength(long expected) {
812             this.expected = expected;
813         }
814 
815         void increaseReceivedBytes(boolean server, int streamId, int bytes, boolean isEnd) throws Http2Exception {
816             seen += bytes;
817             // Check for overflow
818             if (seen < 0) {
819                 throw streamError(streamId, PROTOCOL_ERROR,
820                         "Received amount of data did overflow and so not match content-length header %d", expected);
821             }
822             // Check if we received more data then what was advertised via the content-length header.
823             if (seen > expected) {
824                 throw streamError(streamId, PROTOCOL_ERROR,
825                         "Received amount of data %d does not match content-length header %d", seen, expected);
826             }
827 
828             if (isEnd) {
829                 if (seen == 0 && !server) {
830                     // This may be a response to a HEAD request, let's just allow it.
831                     return;
832                 }
833 
834                 // Check that we really saw what was told via the content-length header.
835                 if (expected > seen) {
836                     throw streamError(streamId, PROTOCOL_ERROR,
837                             "Received amount of data %d does not match content-length header %d", seen, expected);
838                 }
839             }
840         }
841     }
842 }