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 releaseLastContentIfNeeded();
193
194 StompContentSubframe errorContent = new DefaultLastStompContentSubframe(Unpooled.EMPTY_BUFFER);
195 errorContent.setDecoderResult(DecoderResult.failure(e));
196 out.add(errorContent);
197 checkpoint(State.BAD_FRAME);
198 }
199 }
200
201 private void releaseLastContentIfNeeded() {
202 if (lastContent != null) {
203 lastContent.release();
204 lastContent = null;
205 }
206 }
207
208 @Override
209 protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
210 releaseLastContentIfNeeded();
211 }
212
213 private StompCommand readCommand(ByteBuf in) {
214 CharSequence commandSequence = commandParser.parse(in);
215 if (commandSequence == null) {
216 throw new DecoderException("Failed to read command from channel");
217 }
218 String commandStr = commandSequence.toString();
219 try {
220 return StompCommand.valueOf(commandStr);
221 } catch (IllegalArgumentException iae) {
222 throw new DecoderException("Cannot to parse command " + commandStr);
223 }
224 }
225
226 private State readHeaders(ByteBuf buffer, StompHeadersSubframe headersSubframe) {
227 StompHeaders headers = headersSubframe.headers();
228
229
230 headerParser.resetNumHeaders();
231 for (;;) {
232 boolean headerRead = headerParser.parseHeader(headersSubframe, buffer);
233 if (!headerRead) {
234 if (headers.contains(StompHeaders.CONTENT_LENGTH)) {
235 contentLength = getContentLength(headers);
236 if (contentLength == 0) {
237 return State.FINALIZE_FRAME_READ;
238 }
239 }
240 return State.READ_CONTENT;
241 }
242 }
243 }
244
245 private static long getContentLength(StompHeaders headers) {
246 long contentLength = headers.getLong(StompHeaders.CONTENT_LENGTH, 0L);
247 if (contentLength < 0) {
248 throw new DecoderException(StompHeaders.CONTENT_LENGTH + " must be non-negative");
249 }
250 if (contentLength > Integer.MAX_VALUE) {
251 throw new TooLongFrameException(StompHeaders.CONTENT_LENGTH + " exceeds the maximum allowed value: "
252 + contentLength);
253 }
254 return contentLength;
255 }
256
257 private static void skipNullCharacter(ByteBuf buffer) {
258 byte b = buffer.readByte();
259 if (b != StompConstants.NUL) {
260 throw new IllegalStateException("unexpected byte in buffer " + b + " while expecting NULL byte");
261 }
262 }
263
264 private static void skipControlCharacters(ByteBuf buffer) {
265 byte b;
266 for (;;) {
267 if (!buffer.isReadable()) {
268 return;
269 }
270 b = buffer.readByte();
271 if (b != StompConstants.CR && b != StompConstants.LF) {
272 buffer.readerIndex(buffer.readerIndex() - 1);
273 break;
274 }
275 }
276 }
277
278 private void resetDecoder() {
279 checkpoint(State.SKIP_CONTROL_CHARACTERS);
280 contentLength = -1;
281 alreadyReadChunkSize = 0;
282 lastContent = null;
283 }
284
285 private static class Utf8LineParser implements ByteProcessor {
286
287 private final AppendableCharSequence charSeq;
288 private final int maxLineLength;
289
290 private int lineLength;
291 private char interim;
292 private boolean nextRead;
293
294 Utf8LineParser(AppendableCharSequence charSeq, int maxLineLength) {
295 this.charSeq = checkNotNull(charSeq, "charSeq");
296 this.maxLineLength = maxLineLength;
297 }
298
299 AppendableCharSequence parse(ByteBuf byteBuf) {
300 reset();
301 int offset = byteBuf.forEachByte(this);
302 if (offset == -1) {
303 return null;
304 }
305
306 byteBuf.readerIndex(offset + 1);
307 return charSeq;
308 }
309
310 AppendableCharSequence charSequence() {
311 return charSeq;
312 }
313
314 @Override
315 public boolean process(byte nextByte) throws Exception {
316 if (nextByte == StompConstants.CR) {
317 interim = 0;
318 nextRead = false;
319 ++lineLength;
320 return true;
321 }
322
323 if (nextByte == StompConstants.LF) {
324 return false;
325 }
326
327 if (++lineLength > maxLineLength) {
328 throw new TooLongFrameException("An STOMP line is larger than " + maxLineLength + " bytes.");
329 }
330
331
332
333
334 if (nextRead) {
335 interim |= (nextByte & 0x3F) << 6;
336 nextRead = false;
337 } else if (interim != 0) {
338 appendTo(charSeq, (char) (interim | (nextByte & 0x3F)));
339 interim = 0;
340 } else if (nextByte >= 0) {
341
342 appendTo(charSeq, (char) nextByte);
343 } else if ((nextByte & 0xE0) == 0xC0) {
344
345
346 interim = (char) ((nextByte & 0x1F) << 6);
347 } else {
348
349 interim = (char) ((nextByte & 0x0F) << 12);
350 nextRead = true;
351 }
352
353 return true;
354 }
355
356 protected void appendTo(AppendableCharSequence charSeq, char chr) {
357 charSeq.append(chr);
358 }
359
360 protected void reset() {
361 charSeq.reset();
362 lineLength = 0;
363 interim = 0;
364 nextRead = false;
365 }
366 }
367
368 private static final class HeaderParser extends Utf8LineParser {
369
370 private final boolean validateHeaders;
371 private final int maxNumHeaders;
372 private int numHeaders;
373 private String name;
374 private boolean valid;
375
376 private boolean shouldUnescape;
377 private boolean unescapeInProgress;
378
379 HeaderParser(AppendableCharSequence charSeq, int maxLineLength, int maxNumHeaders, boolean validateHeaders) {
380 super(charSeq, maxLineLength);
381 this.validateHeaders = validateHeaders;
382 this.maxNumHeaders = maxNumHeaders;
383 }
384
385 void resetNumHeaders() {
386 numHeaders = 0;
387 }
388
389 boolean parseHeader(StompHeadersSubframe headersSubframe, ByteBuf buf) {
390 shouldUnescape = shouldUnescape(headersSubframe.command());
391 AppendableCharSequence value = super.parse(buf);
392 if (value == null || (name == null && value.length() == 0)) {
393 numHeaders = 0;
394 return false;
395 }
396
397 numHeaders++;
398 if (maxNumHeaders < numHeaders) {
399 throw new TooLongFrameException("maximum number of headers exceeded: " + maxNumHeaders);
400 }
401 if (valid) {
402 headersSubframe.headers().add(name, value.toString());
403 } else if (validateHeaders) {
404 if (StringUtil.isNullOrEmpty(name)) {
405 throw new IllegalArgumentException("received an invalid header line '" + value + '\'');
406 }
407 String line = name + ':' + value;
408 throw new IllegalArgumentException("a header value or name contains a prohibited character ':'"
409 + ", " + line);
410 }
411 return true;
412 }
413
414 @Override
415 public boolean process(byte nextByte) throws Exception {
416 if (nextByte == StompConstants.COLON) {
417 if (name == null) {
418 AppendableCharSequence charSeq = charSequence();
419 if (charSeq.length() != 0) {
420 name = charSeq.substring(0, charSeq.length());
421 charSeq.reset();
422 valid = true;
423 return true;
424 } else {
425 name = StringUtil.EMPTY_STRING;
426 }
427 } else {
428 valid = false;
429 }
430 }
431
432 return super.process(nextByte);
433 }
434
435 @Override
436 protected void appendTo(AppendableCharSequence charSeq, char chr) {
437 if (!shouldUnescape) {
438 super.appendTo(charSeq, chr);
439 return;
440 }
441
442 if (chr == '\\') {
443 if (unescapeInProgress) {
444 super.appendTo(charSeq, chr);
445 unescapeInProgress = false;
446 } else {
447 unescapeInProgress = true;
448 }
449 return;
450 }
451
452 if (unescapeInProgress) {
453 if (chr == 'c') {
454 charSeq.append(':');
455 } else if (chr == 'r') {
456 charSeq.append('\r');
457 } else if (chr == 'n') {
458 charSeq.append('\n');
459 } else {
460 charSeq.append('\\').append(chr);
461 throw new IllegalArgumentException("received an invalid escape header sequence '" + charSeq + '\'');
462 }
463
464 unescapeInProgress = false;
465 return;
466 }
467
468 super.appendTo(charSeq, chr);
469 }
470
471 @Override
472 protected void reset() {
473 name = null;
474 valid = false;
475 unescapeInProgress = false;
476 super.reset();
477 }
478
479 private static boolean shouldUnescape(StompCommand command) {
480 return command != StompCommand.CONNECT && command != StompCommand.CONNECTED;
481 }
482 }
483 }