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.buffer.Unpooled;
19  import io.netty.channel.ChannelHandlerContext;
20  import io.netty.channel.embedded.EmbeddedChannel;
21  import io.netty.handler.codec.ByteToMessageDecoder;
22  import io.netty.handler.codec.compression.Brotli;
23  import io.netty.handler.codec.compression.BrotliDecoder;
24  import io.netty.handler.codec.compression.Zstd;
25  import io.netty.handler.codec.compression.ZstdDecoder;
26  import io.netty.handler.codec.compression.ZlibCodecFactory;
27  import io.netty.handler.codec.compression.ZlibWrapper;
28  import io.netty.handler.codec.compression.SnappyFrameDecoder;
29  
30  import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_ENCODING;
31  import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
32  import static io.netty.handler.codec.http.HttpHeaderValues.BR;
33  import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
34  import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
35  import static io.netty.handler.codec.http.HttpHeaderValues.IDENTITY;
36  import static io.netty.handler.codec.http.HttpHeaderValues.X_DEFLATE;
37  import static io.netty.handler.codec.http.HttpHeaderValues.X_GZIP;
38  import static io.netty.handler.codec.http.HttpHeaderValues.SNAPPY;
39  import static io.netty.handler.codec.http.HttpHeaderValues.ZSTD;
40  import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
41  import static io.netty.handler.codec.http2.Http2Exception.streamError;
42  import static io.netty.util.internal.ObjectUtil.checkNotNull;
43  import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
44  
45  /**
46   * An HTTP2 frame listener that will decompress data frames according to the {@code content-encoding} header for each
47   * stream. The decompression provided by this class will be applied to the data for the entire stream.
48   */
49  public class DelegatingDecompressorFrameListener extends Http2FrameListenerDecorator {
50  
51      private final Http2Connection connection;
52      private final boolean strict;
53      private boolean flowControllerInitialized;
54      private final Http2Connection.PropertyKey propertyKey;
55      private final int maxAllocation;
56  
57      /**
58       * Create a new instance.
59       *
60       * @param connection the connection to read data which should be decompressed
61       * @param listener the delegate listener used by {@link Http2FrameListenerDecorator}
62       * @deprecated
63       *      Use {@link DelegatingDecompressorFrameListener#DelegatingDecompressorFrameListener(Http2Connection,
64       *      Http2FrameListener, int)}
65       */
66      @Deprecated
67      public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener) {
68          this(connection, listener, 0);
69      }
70  
71      /**
72       * Create a new instance.
73       *
74       * @param connection the connection to read data which should be decompressed
75       * @param listener the delegate listener used by {@link Http2FrameListenerDecorator}
76       * @param maxAllocation maximum size of the decompression buffer. Must be >= 0.
77       *                      If zero, maximum size is not limited by decoder.
78       */
79      public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener,
80                                                 int maxAllocation) {
81          this(connection, listener, true, maxAllocation);
82      }
83  
84      /**
85       * Create a new instance.
86       *
87       * @param connection the connection to read data which should be decompressed
88       * @param listener the delegate listener used by {@link Http2FrameListenerDecorator}
89       * @param strict if `true`, {@link ZlibWrapper#ZLIB} will be used for the decoder,
90       *               otherwise the decoder can fallback to {@link ZlibWrapper#NONE}
91       * @deprecated
92       *      Use {@link DelegatingDecompressorFrameListener#DelegatingDecompressorFrameListener(Http2Connection,
93       *      Http2FrameListener, boolean, int)}
94       */
95      @Deprecated
96      public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener,
97                                                 boolean strict) {
98          this(connection, listener, strict, 0);
99      }
100 
101     /**
102      * Create a new instance.
103      *
104      * @param connection the connection to read data which should be decompressed
105      * @param listener the delegate listener used by {@link Http2FrameListenerDecorator}
106      * @param strict if `true`, {@link ZlibWrapper#ZLIB} will be used for the decoder,
107      *               otherwise the decoder can fallback to {@link ZlibWrapper#NONE}
108      * @param maxAllocation maximum size of the decompression buffer. Must be >= 0.
109      *                      If zero, maximum size is not limited by decoder.
110      */
111     public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener,
112                     boolean strict, int maxAllocation) {
113         super(listener);
114         this.connection = connection;
115         this.strict = strict;
116         this.maxAllocation = checkPositiveOrZero(maxAllocation, "maxAllocation");
117 
118         propertyKey = connection.newKey();
119         connection.addListener(new Http2ConnectionAdapter() {
120             @Override
121             public void onStreamRemoved(Http2Stream stream) {
122                 final Http2Decompressor decompressor = decompressor(stream);
123                 if (decompressor != null) {
124                     cleanup(decompressor);
125                 }
126             }
127         });
128     }
129 
130     @Override
131     public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
132             throws Http2Exception {
133         final Http2Stream stream = connection.stream(streamId);
134         final Http2Decompressor decompressor = decompressor(stream);
135         if (decompressor == null) {
136             // The decompressor may be null if no compatible encoding type was found in this stream's headers
137             return listener.onDataRead(ctx, streamId, data, padding, endOfStream);
138         }
139 
140         final EmbeddedChannel channel = decompressor.decompressor();
141         final int compressedBytes = data.readableBytes() + padding;
142         decompressor.incrementCompressedBytes(compressedBytes);
143         try {
144             // call retain here as it will call release after its written to the channel
145             channel.writeInbound(data.retain());
146             ByteBuf buf = nextReadableBuf(channel);
147             if (buf == null && endOfStream && channel.finish()) {
148                 buf = nextReadableBuf(channel);
149             }
150             if (buf == null) {
151                 if (endOfStream) {
152                     listener.onDataRead(ctx, streamId, Unpooled.EMPTY_BUFFER, padding, true);
153                 }
154                 // No new decompressed data was extracted from the compressed data. This means the application could
155                 // not be provided with data and thus could not return how many bytes were processed. We will assume
156                 // there is more data coming which will complete the decompression block. To allow for more data we
157                 // return all bytes to the flow control window (so the peer can send more data).
158                 decompressor.incrementDecompressedBytes(compressedBytes);
159                 return compressedBytes;
160             }
161             try {
162                 Http2LocalFlowController flowController = connection.local().flowController();
163                 decompressor.incrementDecompressedBytes(padding);
164                 for (;;) {
165                     ByteBuf nextBuf = nextReadableBuf(channel);
166                     boolean decompressedEndOfStream = nextBuf == null && endOfStream;
167                     if (decompressedEndOfStream && channel.finish()) {
168                         nextBuf = nextReadableBuf(channel);
169                         decompressedEndOfStream = nextBuf == null;
170                     }
171 
172                     decompressor.incrementDecompressedBytes(buf.readableBytes());
173                     // Immediately return the bytes back to the flow controller. ConsumedBytesConverter will convert
174                     // from the decompressed amount which the user knows about to the compressed amount which flow
175                     // control knows about.
176                     flowController.consumeBytes(stream,
177                             listener.onDataRead(ctx, streamId, buf, padding, decompressedEndOfStream));
178                     if (nextBuf == null) {
179                         break;
180                     }
181 
182                     padding = 0; // Padding is only communicated once on the first iteration.
183                     buf.release();
184                     buf = nextBuf;
185                 }
186                 // We consume bytes each time we call the listener to ensure if multiple frames are decompressed
187                 // that the bytes are accounted for immediately. Otherwise the user may see an inconsistent state of
188                 // flow control.
189                 return 0;
190             } finally {
191                 buf.release();
192             }
193         } catch (Http2Exception e) {
194             throw e;
195         } catch (Throwable t) {
196             throw streamError(stream.id(), INTERNAL_ERROR, t,
197                     "Decompressor error detected while delegating data read on streamId %d", stream.id());
198         }
199     }
200 
201     @Override
202     public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
203                     boolean endStream) throws Http2Exception {
204         initDecompressor(ctx, streamId, headers, endStream);
205         listener.onHeadersRead(ctx, streamId, headers, padding, endStream);
206     }
207 
208     @Override
209     public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
210                     short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception {
211         initDecompressor(ctx, streamId, headers, endStream);
212         listener.onHeadersRead(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endStream);
213     }
214 
215     /**
216      * Returns a new {@link EmbeddedChannel} that decodes the HTTP2 message content encoded in the specified
217      * {@code contentEncoding}.
218      *
219      * @param contentEncoding the value of the {@code content-encoding} header
220      * @return a new {@link ByteToMessageDecoder} if the specified encoding is supported. {@code null} otherwise
221      *         (alternatively, you can throw a {@link Http2Exception} to block unknown encoding).
222      * @throws Http2Exception If the specified encoding is not supported and warrants an exception
223      */
224     protected EmbeddedChannel newContentDecompressor(final ChannelHandlerContext ctx, CharSequence contentEncoding)
225             throws Http2Exception {
226         if (GZIP.contentEqualsIgnoreCase(contentEncoding) || X_GZIP.contentEqualsIgnoreCase(contentEncoding)) {
227             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
228                     ctx.channel().config(), ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP, maxAllocation));
229         }
230         if (DEFLATE.contentEqualsIgnoreCase(contentEncoding) || X_DEFLATE.contentEqualsIgnoreCase(contentEncoding)) {
231             final ZlibWrapper wrapper = strict ? ZlibWrapper.ZLIB : ZlibWrapper.ZLIB_OR_NONE;
232             // To be strict, 'deflate' means ZLIB, but some servers were not implemented correctly.
233             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
234                     ctx.channel().config(), ZlibCodecFactory.newZlibDecoder(wrapper, maxAllocation));
235         }
236         if (Brotli.isAvailable() && BR.contentEqualsIgnoreCase(contentEncoding)) {
237             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
238               ctx.channel().config(), new BrotliDecoder());
239         }
240         if (SNAPPY.contentEqualsIgnoreCase(contentEncoding)) {
241             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
242                     ctx.channel().config(), new SnappyFrameDecoder());
243         }
244         if (Zstd.isAvailable() && ZSTD.contentEqualsIgnoreCase(contentEncoding)) {
245             return new EmbeddedChannel(ctx.channel().id(), ctx.channel().metadata().hasDisconnect(),
246                     ctx.channel().config(), new ZstdDecoder());
247         }
248         // 'identity' or unsupported
249         return null;
250     }
251 
252     /**
253      * Returns the expected content encoding of the decoded content. This getMethod returns {@code "identity"} by
254      * default, which is the case for most decompressors.
255      *
256      * @param contentEncoding the value of the {@code content-encoding} header
257      * @return the expected content encoding of the new content.
258      * @throws Http2Exception if the {@code contentEncoding} is not supported and warrants an exception
259      */
260     protected CharSequence getTargetContentEncoding(@SuppressWarnings("UnusedParameters") CharSequence contentEncoding)
261                     throws Http2Exception {
262         return IDENTITY;
263     }
264 
265     /**
266      * Checks if a new decompressor object is needed for the stream identified by {@code streamId}.
267      * This method will modify the {@code content-encoding} header contained in {@code headers}.
268      *
269      * @param ctx The context
270      * @param streamId The identifier for the headers inside {@code headers}
271      * @param headers Object representing headers which have been read
272      * @param endOfStream Indicates if the stream has ended
273      * @throws Http2Exception If the {@code content-encoding} is not supported
274      */
275     private void initDecompressor(ChannelHandlerContext ctx, int streamId, Http2Headers headers, boolean endOfStream)
276             throws Http2Exception {
277         final Http2Stream stream = connection.stream(streamId);
278         if (stream == null) {
279             return;
280         }
281 
282         Http2Decompressor decompressor = decompressor(stream);
283         if (decompressor == null && !endOfStream) {
284             // Determine the content encoding.
285             CharSequence contentEncoding = headers.get(CONTENT_ENCODING);
286             if (contentEncoding == null) {
287                 contentEncoding = IDENTITY;
288             }
289             final EmbeddedChannel channel = newContentDecompressor(ctx, contentEncoding);
290             if (channel != null) {
291                 decompressor = new Http2Decompressor(channel);
292                 stream.setProperty(propertyKey, decompressor);
293                 // Decode the content and remove or replace the existing headers
294                 // so that the message looks like a decoded message.
295                 CharSequence targetContentEncoding = getTargetContentEncoding(contentEncoding);
296                 if (IDENTITY.contentEqualsIgnoreCase(targetContentEncoding)) {
297                     headers.remove(CONTENT_ENCODING);
298                 } else {
299                     headers.set(CONTENT_ENCODING, targetContentEncoding);
300                 }
301             }
302         }
303 
304         if (decompressor != null) {
305             // The content length will be for the compressed data. Since we will decompress the data
306             // this content-length will not be correct. Instead of queuing messages or delaying sending
307             // header frames just remove the content-length header.
308             headers.remove(CONTENT_LENGTH);
309 
310             // The first time that we initialize a decompressor, decorate the local flow controller to
311             // properly convert consumed bytes.
312             if (!flowControllerInitialized) {
313                 flowControllerInitialized = true;
314                 connection.local().flowController(new ConsumedBytesConverter(connection.local().flowController()));
315             }
316         }
317     }
318 
319     Http2Decompressor decompressor(Http2Stream stream) {
320         return stream == null ? null : (Http2Decompressor) stream.getProperty(propertyKey);
321     }
322 
323     /**
324      * Release remaining content from the {@link EmbeddedChannel}.
325      *
326      * @param decompressor The decompressor for {@code stream}
327      */
328     private static void cleanup(Http2Decompressor decompressor) {
329         decompressor.decompressor().finishAndReleaseAll();
330     }
331 
332     /**
333      * Read the next decompressed {@link ByteBuf} from the {@link EmbeddedChannel}
334      * or {@code null} if one does not exist.
335      *
336      * @param decompressor The channel to read from
337      * @return The next decoded {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist
338      */
339     private static ByteBuf nextReadableBuf(EmbeddedChannel decompressor) {
340         for (;;) {
341             final ByteBuf buf = decompressor.readInbound();
342             if (buf == null) {
343                 return null;
344             }
345             if (!buf.isReadable()) {
346                 buf.release();
347                 continue;
348             }
349             return buf;
350         }
351     }
352 
353     /**
354      * A decorator around the local flow controller that converts consumed bytes from uncompressed to compressed.
355      */
356     private final class ConsumedBytesConverter implements Http2LocalFlowController {
357         private final Http2LocalFlowController flowController;
358 
359         ConsumedBytesConverter(Http2LocalFlowController flowController) {
360             this.flowController = checkNotNull(flowController, "flowController");
361         }
362 
363         @Override
364         public Http2LocalFlowController frameWriter(Http2FrameWriter frameWriter) {
365             return flowController.frameWriter(frameWriter);
366         }
367 
368         @Override
369         public void channelHandlerContext(ChannelHandlerContext ctx) throws Http2Exception {
370             flowController.channelHandlerContext(ctx);
371         }
372 
373         @Override
374         public void initialWindowSize(int newWindowSize) throws Http2Exception {
375             flowController.initialWindowSize(newWindowSize);
376         }
377 
378         @Override
379         public int initialWindowSize() {
380             return flowController.initialWindowSize();
381         }
382 
383         @Override
384         public int windowSize(Http2Stream stream) {
385             return flowController.windowSize(stream);
386         }
387 
388         @Override
389         public void incrementWindowSize(Http2Stream stream, int delta) throws Http2Exception {
390             flowController.incrementWindowSize(stream, delta);
391         }
392 
393         @Override
394         public void receiveFlowControlledFrame(Http2Stream stream, ByteBuf data, int padding,
395                 boolean endOfStream) throws Http2Exception {
396             flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
397         }
398 
399         @Override
400         public boolean consumeBytes(Http2Stream stream, int numBytes) throws Http2Exception {
401             Http2Decompressor decompressor = decompressor(stream);
402             if (decompressor != null) {
403                 // Convert the decompressed bytes to compressed (on the wire) bytes.
404                 numBytes = decompressor.consumeBytes(stream.id(), numBytes);
405             }
406             try {
407                 return flowController.consumeBytes(stream, numBytes);
408             } catch (Http2Exception e) {
409                 throw e;
410             } catch (Throwable t) {
411                 // The stream should be closed at this point. We have already changed our state tracking the compressed
412                 // bytes, and there is no guarantee we can recover if the underlying flow controller throws.
413                 throw streamError(stream.id(), INTERNAL_ERROR, t, "Error while returning bytes to flow control window");
414             }
415         }
416 
417         @Override
418         public int unconsumedBytes(Http2Stream stream) {
419             return flowController.unconsumedBytes(stream);
420         }
421 
422         @Override
423         public int initialWindowSize(Http2Stream stream) {
424             return flowController.initialWindowSize(stream);
425         }
426     }
427 
428     /**
429      * Provides the state for stream {@code DATA} frame decompression.
430      */
431     private static final class Http2Decompressor {
432         private final EmbeddedChannel decompressor;
433         private int compressed;
434         private int decompressed;
435 
436         Http2Decompressor(EmbeddedChannel decompressor) {
437             this.decompressor = decompressor;
438         }
439 
440         /**
441          * Responsible for taking compressed bytes in and producing decompressed bytes.
442          */
443         EmbeddedChannel decompressor() {
444             return decompressor;
445         }
446 
447         /**
448          * Increment the number of bytes received prior to doing any decompression.
449          */
450         void incrementCompressedBytes(int delta) {
451             assert delta >= 0;
452             compressed += delta;
453         }
454 
455         /**
456          * Increment the number of bytes after the decompression process.
457          */
458         void incrementDecompressedBytes(int delta) {
459             assert delta >= 0;
460             decompressed += delta;
461         }
462 
463         /**
464          * Determines the ratio between {@code numBytes} and {@link Http2Decompressor#decompressed}.
465          * This ratio is used to decrement {@link Http2Decompressor#decompressed} and
466          * {@link Http2Decompressor#compressed}.
467          * @param streamId the stream ID
468          * @param decompressedBytes The number of post-decompressed bytes to return to flow control
469          * @return The number of pre-decompressed bytes that have been consumed.
470          */
471         int consumeBytes(int streamId, int decompressedBytes) throws Http2Exception {
472             checkPositiveOrZero(decompressedBytes, "decompressedBytes");
473             if (decompressed - decompressedBytes < 0) {
474                 throw streamError(streamId, INTERNAL_ERROR,
475                         "Attempting to return too many bytes for stream %d. decompressed: %d " +
476                                 "decompressedBytes: %d", streamId, decompressed, decompressedBytes);
477             }
478             double consumedRatio = decompressedBytes / (double) decompressed;
479             int consumedCompressed = Math.min(compressed, (int) Math.ceil(compressed * consumedRatio));
480             if (compressed - consumedCompressed < 0) {
481                 throw streamError(streamId, INTERNAL_ERROR,
482                         "overflow when converting decompressed bytes to compressed bytes for stream %d." +
483                                 "decompressedBytes: %d decompressed: %d compressed: %d consumedCompressed: %d",
484                         streamId, decompressedBytes, decompressed, compressed, consumedCompressed);
485             }
486             decompressed -= decompressedBytes;
487             compressed -= consumedCompressed;
488 
489             return consumedCompressed;
490         }
491     }
492 }