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