1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.stomp;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.Unpooled;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.handler.codec.DecoderException;
22 import io.netty.handler.codec.DecoderResult;
23 import io.netty.handler.codec.ReplayingDecoder;
24 import io.netty.handler.codec.TooLongFrameException;
25 import io.netty.handler.codec.stomp.StompSubframeDecoder.State;
26 import io.netty.util.ByteProcessor;
27 import io.netty.util.internal.AppendableCharSequence;
28 import io.netty.util.internal.StringUtil;
29
30 import java.util.List;
31
32 import static io.netty.buffer.ByteBufUtil.*;
33 import static io.netty.util.internal.ObjectUtil.*;
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56 public class StompSubframeDecoder extends ReplayingDecoder<State> {
57
58 private static final int DEFAULT_CHUNK_SIZE = 8132;
59 private static final int DEFAULT_MAX_LINE_LENGTH = 1024;
60 private static final int DEFAULT_MAX_NUMBER_HEADERS = 128;
61
62
63
64
65 @Deprecated
66 public enum State {
67 SKIP_CONTROL_CHARACTERS,
68 READ_HEADERS,
69 READ_CONTENT,
70 FINALIZE_FRAME_READ,
71 BAD_FRAME,
72 INVALID_CHUNK
73 }
74
75 private final Utf8LineParser commandParser;
76 private final HeaderParser headerParser;
77 private final int maxChunkSize;
78 private int alreadyReadChunkSize;
79 private LastStompContentSubframe lastContent;
80 private long contentLength = -1;
81
82 public StompSubframeDecoder() {
83 this(DEFAULT_MAX_LINE_LENGTH, DEFAULT_CHUNK_SIZE);
84 }
85
86 public StompSubframeDecoder(boolean validateHeaders) {
87 this(DEFAULT_MAX_LINE_LENGTH, DEFAULT_CHUNK_SIZE, DEFAULT_MAX_NUMBER_HEADERS, validateHeaders);
88 }
89
90 public StompSubframeDecoder(int maxLineLength, int maxChunkSize) {
91 this(maxLineLength, maxChunkSize, false);
92 }
93
94 public StompSubframeDecoder(int maxLineLength, int maxChunkSize, boolean validateHeaders) {
95 this(maxLineLength, maxChunkSize, DEFAULT_MAX_NUMBER_HEADERS, validateHeaders);
96 }
97
98 public StompSubframeDecoder(int maxLineLength, int maxChunkSize, int maxNumHeaders, boolean validateHeaders) {
99 super(State.SKIP_CONTROL_CHARACTERS);
100 checkPositive(maxLineLength, "maxLineLength");
101 checkPositive(maxChunkSize, "maxChunkSize");
102 checkPositive(maxNumHeaders, "maxNumHeaders");
103
104 this.maxChunkSize = maxChunkSize;
105 commandParser = new Utf8LineParser(new AppendableCharSequence(16), maxLineLength);
106 headerParser = new HeaderParser(new AppendableCharSequence(128), maxLineLength, maxNumHeaders, validateHeaders);
107 }
108
109 @Override
110 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
111 switch (state()) {
112 case SKIP_CONTROL_CHARACTERS:
113 skipControlCharacters(in);
114 checkpoint(State.READ_HEADERS);
115
116 case READ_HEADERS:
117 StompCommand command = StompCommand.UNKNOWN;
118 StompHeadersSubframe frame = null;
119 try {
120 command = readCommand(in);
121 frame = new DefaultStompHeadersSubframe(command);
122 checkpoint(readHeaders(in, frame));
123 out.add(frame);
124 } catch (Exception e) {
125 if (frame == null) {
126 frame = new DefaultStompHeadersSubframe(command);
127 }
128 frame.setDecoderResult(DecoderResult.failure(e));
129 out.add(frame);
130 checkpoint(State.BAD_FRAME);
131 return;
132 }
133 break;
134 case BAD_FRAME:
135 in.skipBytes(actualReadableBytes());
136 return;
137 }
138 try {
139 switch (state()) {
140 case READ_CONTENT:
141 int toRead = in.readableBytes();
142 if (toRead == 0) {
143 return;
144 }
145 if (toRead > maxChunkSize) {
146 toRead = maxChunkSize;
147 }
148 if (contentLength >= 0) {
149 int remainingLength = (int) (contentLength - alreadyReadChunkSize);
150 if (toRead > remainingLength) {
151 toRead = remainingLength;
152 }
153 ByteBuf chunkBuffer = readBytes(ctx.alloc(), in, toRead);
154 if ((alreadyReadChunkSize += toRead) >= contentLength) {
155 lastContent = new DefaultLastStompContentSubframe(chunkBuffer);
156 checkpoint(State.FINALIZE_FRAME_READ);
157 } else {
158 out.add(new DefaultStompContentSubframe(chunkBuffer));
159 return;
160 }
161 } else {
162 int nulIndex = indexOf(in, in.readerIndex(), in.writerIndex(), StompConstants.NUL);
163 if (nulIndex == in.readerIndex()) {
164 checkpoint(State.FINALIZE_FRAME_READ);
165 } else {
166 if (nulIndex > 0) {
167 toRead = nulIndex - in.readerIndex();
168 } else {
169 toRead = in.writerIndex() - in.readerIndex();
170 }
171 ByteBuf chunkBuffer = readBytes(ctx.alloc(), in, toRead);
172 alreadyReadChunkSize += toRead;
173 if (nulIndex > 0) {
174 lastContent = new DefaultLastStompContentSubframe(chunkBuffer);
175 checkpoint(State.FINALIZE_FRAME_READ);
176 } else {
177 out.add(new DefaultStompContentSubframe(chunkBuffer));
178 return;
179 }
180 }
181 }
182
183 case FINALIZE_FRAME_READ:
184 skipNullCharacter(in);
185 if (lastContent == null) {
186 lastContent = LastStompContentSubframe.EMPTY_LAST_CONTENT;
187 }
188 out.add(lastContent);
189 resetDecoder();
190 }
191 } catch (Exception e) {
192 if (lastContent != null) {
193 lastContent.release();
194 lastContent = null;
195 }
196
197 StompContentSubframe errorContent = new DefaultLastStompContentSubframe(Unpooled.EMPTY_BUFFER);
198 errorContent.setDecoderResult(DecoderResult.failure(e));
199 out.add(errorContent);
200 checkpoint(State.BAD_FRAME);
201 }
202 }
203
204 private StompCommand readCommand(ByteBuf in) {
205 CharSequence commandSequence = commandParser.parse(in);
206 if (commandSequence == null) {
207 throw new DecoderException("Failed to read command from channel");
208 }
209 String commandStr = commandSequence.toString();
210 try {
211 return StompCommand.valueOf(commandStr);
212 } catch (IllegalArgumentException iae) {
213 throw new DecoderException("Cannot to parse command " + commandStr);
214 }
215 }
216
217 private State readHeaders(ByteBuf buffer, StompHeadersSubframe headersSubframe) {
218 StompHeaders headers = headersSubframe.headers();
219 for (;;) {
220 boolean headerRead = headerParser.parseHeader(headersSubframe, buffer);
221 if (!headerRead) {
222 if (headers.contains(StompHeaders.CONTENT_LENGTH)) {
223 contentLength = getContentLength(headers);
224 if (contentLength == 0) {
225 return State.FINALIZE_FRAME_READ;
226 }
227 }
228 return State.READ_CONTENT;
229 }
230 }
231 }
232
233 private static long getContentLength(StompHeaders headers) {
234 long contentLength = headers.getLong(StompHeaders.CONTENT_LENGTH, 0L);
235 if (contentLength < 0) {
236 throw new DecoderException(StompHeaders.CONTENT_LENGTH + " must be non-negative");
237 }
238 return contentLength;
239 }
240
241 private static void skipNullCharacter(ByteBuf buffer) {
242 byte b = buffer.readByte();
243 if (b != StompConstants.NUL) {
244 throw new IllegalStateException("unexpected byte in buffer " + b + " while expecting NULL byte");
245 }
246 }
247
248 private static void skipControlCharacters(ByteBuf buffer) {
249 byte b;
250 for (;;) {
251 if (!buffer.isReadable()) {
252 return;
253 }
254 b = buffer.readByte();
255 if (b != StompConstants.CR && b != StompConstants.LF) {
256 buffer.readerIndex(buffer.readerIndex() - 1);
257 break;
258 }
259 }
260 }
261
262 private void resetDecoder() {
263 checkpoint(State.SKIP_CONTROL_CHARACTERS);
264 contentLength = -1;
265 alreadyReadChunkSize = 0;
266 lastContent = null;
267 }
268
269 private static class Utf8LineParser implements ByteProcessor {
270
271 private final AppendableCharSequence charSeq;
272 private final int maxLineLength;
273
274 private int lineLength;
275 private char interim;
276 private boolean nextRead;
277
278 Utf8LineParser(AppendableCharSequence charSeq, int maxLineLength) {
279 this.charSeq = checkNotNull(charSeq, "charSeq");
280 this.maxLineLength = maxLineLength;
281 }
282
283 AppendableCharSequence parse(ByteBuf byteBuf) {
284 reset();
285 int offset = byteBuf.forEachByte(this);
286 if (offset == -1) {
287 return null;
288 }
289
290 byteBuf.readerIndex(offset + 1);
291 return charSeq;
292 }
293
294 AppendableCharSequence charSequence() {
295 return charSeq;
296 }
297
298 @Override
299 public boolean process(byte nextByte) throws Exception {
300 if (nextByte == StompConstants.CR) {
301 interim = 0;
302 nextRead = false;
303 ++lineLength;
304 return true;
305 }
306
307 if (nextByte == StompConstants.LF) {
308 return false;
309 }
310
311 if (++lineLength > maxLineLength) {
312 throw new TooLongFrameException("An STOMP line is larger than " + maxLineLength + " bytes.");
313 }
314
315
316
317
318 if (nextRead) {
319 interim |= (nextByte & 0x3F) << 6;
320 nextRead = false;
321 } else if (interim != 0) {
322 appendTo(charSeq, (char) (interim | (nextByte & 0x3F)));
323 interim = 0;
324 } else if (nextByte >= 0) {
325
326 appendTo(charSeq, (char) nextByte);
327 } else if ((nextByte & 0xE0) == 0xC0) {
328
329
330 interim = (char) ((nextByte & 0x1F) << 6);
331 } else {
332
333 interim = (char) ((nextByte & 0x0F) << 12);
334 nextRead = true;
335 }
336
337 return true;
338 }
339
340 protected void appendTo(AppendableCharSequence charSeq, char chr) {
341 charSeq.append(chr);
342 }
343
344 protected void reset() {
345 charSeq.reset();
346 lineLength = 0;
347 interim = 0;
348 nextRead = false;
349 }
350 }
351
352 private static final class HeaderParser extends Utf8LineParser {
353
354 private final boolean validateHeaders;
355 private final int maxNumHeaders;
356 private int numHeaders;
357 private String name;
358 private boolean valid;
359
360 private boolean shouldUnescape;
361 private boolean unescapeInProgress;
362
363 HeaderParser(AppendableCharSequence charSeq, int maxLineLength, int maxNumHeaders, boolean validateHeaders) {
364 super(charSeq, maxLineLength);
365 this.validateHeaders = validateHeaders;
366 this.maxNumHeaders = maxNumHeaders;
367 }
368
369 boolean parseHeader(StompHeadersSubframe headersSubframe, ByteBuf buf) {
370 shouldUnescape = shouldUnescape(headersSubframe.command());
371 AppendableCharSequence value = super.parse(buf);
372 if (value == null || (name == null && value.length() == 0)) {
373 numHeaders = 0;
374 return false;
375 }
376
377 numHeaders++;
378 if (maxNumHeaders < numHeaders) {
379 throw new TooLongFrameException("maximum number of headers exceeded: " + maxNumHeaders);
380 }
381 if (valid) {
382 headersSubframe.headers().add(name, value.toString());
383 } else if (validateHeaders) {
384 if (StringUtil.isNullOrEmpty(name)) {
385 throw new IllegalArgumentException("received an invalid header line '" + value + '\'');
386 }
387 String line = name + ':' + value;
388 throw new IllegalArgumentException("a header value or name contains a prohibited character ':'"
389 + ", " + line);
390 }
391 return true;
392 }
393
394 @Override
395 public boolean process(byte nextByte) throws Exception {
396 if (nextByte == StompConstants.COLON) {
397 if (name == null) {
398 AppendableCharSequence charSeq = charSequence();
399 if (charSeq.length() != 0) {
400 name = charSeq.substring(0, charSeq.length());
401 charSeq.reset();
402 valid = true;
403 return true;
404 } else {
405 name = StringUtil.EMPTY_STRING;
406 }
407 } else {
408 valid = false;
409 }
410 }
411
412 return super.process(nextByte);
413 }
414
415 @Override
416 protected void appendTo(AppendableCharSequence charSeq, char chr) {
417 if (!shouldUnescape) {
418 super.appendTo(charSeq, chr);
419 return;
420 }
421
422 if (chr == '\\') {
423 if (unescapeInProgress) {
424 super.appendTo(charSeq, chr);
425 unescapeInProgress = false;
426 } else {
427 unescapeInProgress = true;
428 }
429 return;
430 }
431
432 if (unescapeInProgress) {
433 if (chr == 'c') {
434 charSeq.append(':');
435 } else if (chr == 'r') {
436 charSeq.append('\r');
437 } else if (chr == 'n') {
438 charSeq.append('\n');
439 } else {
440 charSeq.append('\\').append(chr);
441 throw new IllegalArgumentException("received an invalid escape header sequence '" + charSeq + '\'');
442 }
443
444 unescapeInProgress = false;
445 return;
446 }
447
448 super.appendTo(charSeq, chr);
449 }
450
451 @Override
452 protected void reset() {
453 name = null;
454 valid = false;
455 unescapeInProgress = false;
456 super.reset();
457 }
458
459 private static boolean shouldUnescape(StompCommand command) {
460 return command != StompCommand.CONNECT && command != StompCommand.CONNECTED;
461 }
462 }
463 }