1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.compression;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufAllocator;
20 import io.netty.util.internal.UnstableApi;
21
22 import static io.netty.handler.codec.compression.Snappy.validateChecksum;
23
24
25
26
27
28
29 @UnstableApi
30 public final class SnappyFrameDecompressor extends InputBufferingDecompressor {
31
32 private enum ChunkType {
33 STREAM_IDENTIFIER,
34 COMPRESSED_DATA,
35 UNCOMPRESSED_DATA,
36 RESERVED_UNSKIPPABLE,
37 RESERVED_SKIPPABLE
38 }
39
40 private static final int SNAPPY_IDENTIFIER_LEN = 6;
41
42 private static final int MAX_UNCOMPRESSED_DATA_SIZE = 65536 + 4;
43
44 private static final int MIN_UNCOMPRESSED_DATA_SIZE = 4;
45
46 private static final int MAX_DECOMPRESSED_DATA_SIZE = 65536;
47
48 private static final int MAX_COMPRESSED_CHUNK_SIZE = 16777216 - 1;
49
50 private static final int MIN_COMPRESSED_CHUNK_SIZE = 5;
51
52 private final Snappy snappy = new Snappy();
53 private final boolean validateChecksums;
54
55 private boolean started;
56 private int numBytesToSkip;
57
58 private ByteBuf pendingOutput;
59 private boolean eof;
60
61 SnappyFrameDecompressor(Builder builder, ByteBufAllocator allocator) {
62 super(allocator);
63 this.validateChecksums = builder.validateChecksums;
64 }
65
66 @Override
67 public void close() {
68 super.close();
69 if (pendingOutput != null) {
70 pendingOutput.release();
71 pendingOutput = null;
72 }
73 }
74
75 @Override
76 public Status status() throws DecompressionException {
77 if (pendingOutput != null) {
78 return Status.NEED_OUTPUT;
79 }
80 if (eof) {
81 return Status.COMPLETE;
82 }
83 return Status.NEED_INPUT;
84 }
85
86 @Override
87 public void endOfInput() throws DecompressionException {
88 if (available() != 0 || numBytesToSkip != 0) {
89 throw new DecompressionException("Unexpected end of input");
90 }
91 eof = true;
92 }
93
94 @Override
95 void processInput(ByteBuf in) throws DecompressionException {
96 while (in.isReadable()) {
97 if (numBytesToSkip != 0) {
98
99 int skipBytes = Math.min(numBytesToSkip, in.readableBytes());
100 in.skipBytes(skipBytes);
101 numBytesToSkip -= skipBytes;
102
103
104 continue;
105 }
106
107 int idx = in.readerIndex();
108 final int inSize = in.readableBytes();
109 if (inSize < 4) {
110
111
112 return;
113 }
114
115 final int chunkTypeVal = in.getUnsignedByte(idx);
116 final ChunkType chunkType = mapChunkType((byte) chunkTypeVal);
117 final int chunkLength = in.getUnsignedMediumLE(idx + 1);
118
119 switch (chunkType) {
120 case STREAM_IDENTIFIER:
121 if (chunkLength != SNAPPY_IDENTIFIER_LEN) {
122 throw new DecompressionException("Unexpected length of stream identifier: " + chunkLength);
123 }
124
125 if (inSize < 4 + SNAPPY_IDENTIFIER_LEN) {
126 return;
127 }
128
129 in.skipBytes(4);
130 int offset = in.readerIndex();
131 in.skipBytes(SNAPPY_IDENTIFIER_LEN);
132
133 checkByte(in.getByte(offset++), (byte) 's');
134 checkByte(in.getByte(offset++), (byte) 'N');
135 checkByte(in.getByte(offset++), (byte) 'a');
136 checkByte(in.getByte(offset++), (byte) 'P');
137 checkByte(in.getByte(offset++), (byte) 'p');
138 checkByte(in.getByte(offset), (byte) 'Y');
139
140 started = true;
141 break;
142 case RESERVED_SKIPPABLE:
143 if (!started) {
144 throw new DecompressionException("Received RESERVED_SKIPPABLE tag before STREAM_IDENTIFIER");
145 }
146
147 in.skipBytes(4);
148
149 int skipBytes = Math.min(chunkLength, in.readableBytes());
150 in.skipBytes(skipBytes);
151 if (skipBytes != chunkLength) {
152
153
154 numBytesToSkip = chunkLength - skipBytes;
155 }
156 break;
157 case RESERVED_UNSKIPPABLE:
158
159
160
161 throw new DecompressionException(
162 "Found reserved unskippable chunk type: 0x" + Integer.toHexString(chunkTypeVal));
163 case UNCOMPRESSED_DATA:
164 if (!started) {
165 throw new DecompressionException("Received UNCOMPRESSED_DATA tag before STREAM_IDENTIFIER");
166 }
167 if (chunkLength > MAX_UNCOMPRESSED_DATA_SIZE) {
168 throw new DecompressionException("Received UNCOMPRESSED_DATA larger than " +
169 MAX_UNCOMPRESSED_DATA_SIZE + " bytes");
170 }
171 if (chunkLength < MIN_UNCOMPRESSED_DATA_SIZE) {
172 throw new DecompressionException("Received UNCOMPRESSED_DATA with invalid chunk length: " +
173 chunkLength);
174 }
175
176 if (inSize < 4 + chunkLength) {
177 return;
178 }
179
180 in.skipBytes(4);
181 if (validateChecksums) {
182 int checksum = in.readIntLE();
183 validateChecksum(checksum, in, in.readerIndex(), chunkLength - 4);
184 } else {
185 in.skipBytes(4);
186 }
187 pendingOutput = in.readRetainedSlice(chunkLength - 4);
188 return;
189 case COMPRESSED_DATA:
190 if (!started) {
191 throw new DecompressionException("Received COMPRESSED_DATA tag before STREAM_IDENTIFIER");
192 }
193
194 if (chunkLength > MAX_COMPRESSED_CHUNK_SIZE) {
195 throw new DecompressionException("Received COMPRESSED_DATA that contains" +
196 " chunk that exceeds " + MAX_COMPRESSED_CHUNK_SIZE + " bytes");
197 }
198 if (chunkLength < MIN_COMPRESSED_CHUNK_SIZE) {
199 throw new DecompressionException("Received COMPRESSED_DATA with invalid chunk length: " +
200 chunkLength);
201 }
202
203 if (inSize < 4 + chunkLength) {
204 return;
205 }
206
207 in.skipBytes(4);
208 int checksum = in.readIntLE();
209
210 int uncompressedSize = snappy.getPreamble(in);
211 if (uncompressedSize > MAX_DECOMPRESSED_DATA_SIZE) {
212 throw new DecompressionException("Received COMPRESSED_DATA that contains" +
213 " uncompressed data that exceeds " + MAX_DECOMPRESSED_DATA_SIZE + " bytes");
214 }
215
216 ByteBuf uncompressed = allocator.buffer(uncompressedSize, MAX_DECOMPRESSED_DATA_SIZE);
217 try {
218 if (validateChecksums) {
219 int oldWriterIndex = in.writerIndex();
220 try {
221 in.writerIndex(in.readerIndex() + chunkLength - 4);
222 snappy.decode(in, uncompressed);
223 } finally {
224 in.writerIndex(oldWriterIndex);
225 }
226 validateChecksum(checksum, uncompressed, 0, uncompressed.writerIndex());
227 } else {
228 snappy.decode(in.readSlice(chunkLength - 4), uncompressed);
229 }
230 pendingOutput = uncompressed;
231 uncompressed = null;
232 } finally {
233 if (uncompressed != null) {
234 uncompressed.release();
235 }
236 }
237 snappy.reset();
238 return;
239 }
240 }
241 }
242
243 @Override
244 ByteBuf processOutput(ByteBuf buf) throws DecompressionException {
245 ByteBuf p = pendingOutput;
246 pendingOutput = null;
247 return p;
248 }
249
250 private static void checkByte(byte actual, byte expect) {
251 if (actual != expect) {
252 throw new DecompressionException("Unexpected stream identifier contents. Mismatched snappy " +
253 "protocol version?");
254 }
255 }
256
257
258
259
260
261
262
263 private static ChunkType mapChunkType(byte type) {
264 if (type == 0) {
265 return ChunkType.COMPRESSED_DATA;
266 } else if (type == 1) {
267 return ChunkType.UNCOMPRESSED_DATA;
268 } else if (type == (byte) 0xff) {
269 return ChunkType.STREAM_IDENTIFIER;
270 } else if ((type & 0x80) == 0x80) {
271 return ChunkType.RESERVED_SKIPPABLE;
272 } else {
273 return ChunkType.RESERVED_UNSKIPPABLE;
274 }
275 }
276
277 @UnstableApi
278 public static Builder builder() {
279 return new Builder();
280 }
281
282 @UnstableApi
283 public static final class Builder extends AbstractDecompressorBuilder {
284 boolean validateChecksums;
285
286 Builder() {
287 }
288
289
290
291
292
293
294
295
296
297 public Builder validateChecksums(boolean validateChecksums) {
298 this.validateChecksums = validateChecksums;
299 return this;
300 }
301
302 @Override
303 public Decompressor build(ByteBufAllocator allocator) throws DecompressionException {
304 return new DefensiveDecompressor(new SnappyFrameDecompressor(this, allocator));
305 }
306 }
307 }