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.channel.ChannelHandlerContext;
20 import io.netty.handler.codec.ByteToMessageDecoder;
21
22 import java.util.List;
23
24 import static io.netty.handler.codec.compression.Snappy.validateChecksum;
25
26
27
28
29
30
31
32
33
34
35
36
37 public class SnappyFrameDecoder extends ByteToMessageDecoder {
38
39 private enum ChunkType {
40 STREAM_IDENTIFIER,
41 COMPRESSED_DATA,
42 UNCOMPRESSED_DATA,
43 RESERVED_UNSKIPPABLE,
44 RESERVED_SKIPPABLE
45 }
46
47 private static final int SNAPPY_IDENTIFIER_LEN = 6;
48
49 private static final int MAX_UNCOMPRESSED_DATA_SIZE = 65536 + 4;
50
51 private static final int MIN_UNCOMPRESSED_DATA_SIZE = 4;
52
53 private static final int MAX_DECOMPRESSED_DATA_SIZE = 65536;
54
55 private static final int MAX_COMPRESSED_CHUNK_SIZE = 16777216 - 1;
56
57 private static final int MIN_COMPRESSED_CHUNK_SIZE = 5;
58
59 private final Snappy snappy = new Snappy();
60 private final boolean validateChecksums;
61
62 private boolean started;
63 private boolean corrupted;
64 private int numBytesToSkip;
65
66
67
68
69
70
71 public SnappyFrameDecoder() {
72 this(false);
73 }
74
75
76
77
78
79
80
81
82
83
84 public SnappyFrameDecoder(boolean validateChecksums) {
85 this.validateChecksums = validateChecksums;
86 }
87
88 @Override
89 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
90 if (corrupted) {
91 in.skipBytes(in.readableBytes());
92 return;
93 }
94
95 if (numBytesToSkip != 0) {
96
97 int skipBytes = Math.min(numBytesToSkip, in.readableBytes());
98 in.skipBytes(skipBytes);
99 numBytesToSkip -= skipBytes;
100
101
102 return;
103 }
104
105 try {
106 int idx = in.readerIndex();
107 final int inSize = in.readableBytes();
108 if (inSize < 4) {
109
110
111 return;
112 }
113
114 final int chunkTypeVal = in.getUnsignedByte(idx);
115 final ChunkType chunkType = mapChunkType((byte) chunkTypeVal);
116 final int chunkLength = in.getUnsignedMediumLE(idx + 1);
117
118 switch (chunkType) {
119 case STREAM_IDENTIFIER:
120 if (chunkLength != SNAPPY_IDENTIFIER_LEN) {
121 throw new DecompressionException("Unexpected length of stream identifier: " + chunkLength);
122 }
123
124 if (inSize < 4 + SNAPPY_IDENTIFIER_LEN) {
125 break;
126 }
127
128 in.skipBytes(4);
129 int offset = in.readerIndex();
130 in.skipBytes(SNAPPY_IDENTIFIER_LEN);
131
132 checkByte(in.getByte(offset++), (byte) 's');
133 checkByte(in.getByte(offset++), (byte) 'N');
134 checkByte(in.getByte(offset++), (byte) 'a');
135 checkByte(in.getByte(offset++), (byte) 'P');
136 checkByte(in.getByte(offset++), (byte) 'p');
137 checkByte(in.getByte(offset), (byte) 'Y');
138
139 started = true;
140 break;
141 case RESERVED_SKIPPABLE:
142 if (!started) {
143 throw new DecompressionException("Received RESERVED_SKIPPABLE tag before STREAM_IDENTIFIER");
144 }
145
146 in.skipBytes(4);
147
148 int skipBytes = Math.min(chunkLength, in.readableBytes());
149 in.skipBytes(skipBytes);
150 if (skipBytes != chunkLength) {
151
152
153 numBytesToSkip = chunkLength - skipBytes;
154 }
155 break;
156 case RESERVED_UNSKIPPABLE:
157
158
159
160 throw new DecompressionException(
161 "Found reserved unskippable chunk type: 0x" + Integer.toHexString(chunkTypeVal));
162 case UNCOMPRESSED_DATA:
163 if (!started) {
164 throw new DecompressionException("Received UNCOMPRESSED_DATA tag before STREAM_IDENTIFIER");
165 }
166 if (chunkLength > MAX_UNCOMPRESSED_DATA_SIZE) {
167 throw new DecompressionException("Received UNCOMPRESSED_DATA larger than " +
168 MAX_UNCOMPRESSED_DATA_SIZE + " bytes");
169 }
170 if (chunkLength < MIN_UNCOMPRESSED_DATA_SIZE) {
171 throw new DecompressionException("Received UNCOMPRESSED_DATA with invalid chunk length: " +
172 chunkLength);
173 }
174
175 if (inSize < 4 + chunkLength) {
176 return;
177 }
178
179 in.skipBytes(4);
180 if (validateChecksums) {
181 int checksum = in.readIntLE();
182 validateChecksum(checksum, in, in.readerIndex(), chunkLength - 4);
183 } else {
184 in.skipBytes(4);
185 }
186 out.add(in.readRetainedSlice(chunkLength - 4));
187 break;
188 case COMPRESSED_DATA:
189 if (!started) {
190 throw new DecompressionException("Received COMPRESSED_DATA tag before STREAM_IDENTIFIER");
191 }
192
193 if (chunkLength > MAX_COMPRESSED_CHUNK_SIZE) {
194 throw new DecompressionException("Received COMPRESSED_DATA that contains" +
195 " chunk that exceeds " + MAX_COMPRESSED_CHUNK_SIZE + " bytes");
196 }
197 if (chunkLength < MIN_COMPRESSED_CHUNK_SIZE) {
198 throw new DecompressionException("Received COMPRESSED_DATA with invalid chunk length: " +
199 chunkLength);
200 }
201
202 if (inSize < 4 + chunkLength) {
203 return;
204 }
205
206 in.skipBytes(4);
207 int checksum = in.readIntLE();
208
209 int uncompressedSize = snappy.getPreamble(in);
210 if (uncompressedSize > MAX_DECOMPRESSED_DATA_SIZE) {
211 throw new DecompressionException("Received COMPRESSED_DATA that contains" +
212 " uncompressed data that exceeds " + MAX_DECOMPRESSED_DATA_SIZE + " bytes");
213 }
214
215 ByteBuf uncompressed = ctx.alloc().buffer(uncompressedSize, MAX_DECOMPRESSED_DATA_SIZE);
216 try {
217 if (validateChecksums) {
218 int oldWriterIndex = in.writerIndex();
219 try {
220 in.writerIndex(in.readerIndex() + chunkLength - 4);
221 snappy.decode(in, uncompressed);
222 } finally {
223 in.writerIndex(oldWriterIndex);
224 }
225 validateChecksum(checksum, uncompressed, 0, uncompressed.writerIndex());
226 } else {
227 snappy.decode(in.readSlice(chunkLength - 4), uncompressed);
228 }
229 out.add(uncompressed);
230 uncompressed = null;
231 } finally {
232 if (uncompressed != null) {
233 uncompressed.release();
234 }
235 }
236 snappy.reset();
237 break;
238 }
239 } catch (Exception e) {
240 corrupted = true;
241 throw e;
242 }
243 }
244
245 private static void checkByte(byte actual, byte expect) {
246 if (actual != expect) {
247 throw new DecompressionException("Unexpected stream identifier contents. Mismatched snappy " +
248 "protocol version?");
249 }
250 }
251
252
253
254
255
256
257
258 private static ChunkType mapChunkType(byte type) {
259 if (type == 0) {
260 return ChunkType.COMPRESSED_DATA;
261 } else if (type == 1) {
262 return ChunkType.UNCOMPRESSED_DATA;
263 } else if (type == (byte) 0xff) {
264 return ChunkType.STREAM_IDENTIFIER;
265 } else if ((type & 0x80) == 0x80) {
266 return ChunkType.RESERVED_SKIPPABLE;
267 } else {
268 return ChunkType.RESERVED_UNSKIPPABLE;
269 }
270 }
271 }