Fix exception in an obscure chunked-body case.

The exception was happening when the final CRLF sequence was separated
into two different packets. Added a new state to correctly handle this
case. Closes elastic#84.

http.go

@@ -17,6 +17,7 @@ const (
 	BODY
 	BODY_CHUNKED_START
 	BODY_CHUNKED
+	BODY_CHUNKED_WAIT_FINAL_CRLF
 )
 
 // Http Message
@@ -402,13 +403,30 @@ func (http *Http) messageParser(s *HttpStream) (bool, bool) {
 			if !cont {
 				return ok, complete
 			}
+
+		case BODY_CHUNKED_WAIT_FINAL_CRLF:
+			return state_body_chunked_wait_final_crlf(s, m)
 		}
 
 	}
 
 	return true, false
 }
 
+func state_body_chunked_wait_final_crlf(s *HttpStream, m *HttpMessage) (ok bool, complete bool) {
+	if len(s.data[s.parseOffset:]) < 2 {
+		return true, false
+	} else {
+		if s.data[s.parseOffset] != '\r' || s.data[s.parseOffset+1] != '\n' {
+			WARN("Expected CRLF sequence at end of message")
+			return false, false
+		}
+		s.parseOffset += 2 // skip final CRLF
+		m.end = s.parseOffset
+		return true, true
+	}
+}
+
 func state_body_chunked_start(s *HttpStream, m *HttpMessage) (cont bool, ok bool, complete bool) {
 	// read hexa length
 	i := bytes.Index(s.data[s.parseOffset:], []byte("\r\n"))
@@ -424,24 +442,34 @@ func state_body_chunked_start(s *HttpStream, m *HttpMessage) (cont bool, ok bool
 
 	s.parseOffset += i + 2 //+ \r\n
 	if m.chunked_length == 0 {
-		s.parseOffset += 2 // final \r\n
+		if len(s.data[s.parseOffset:]) < 2 {
+			s.parseState = BODY_CHUNKED_WAIT_FINAL_CRLF
+			return false, true, false
+		}
+		if s.data[s.parseOffset] != '\r' || s.data[s.parseOffset+1] != '\n' {
+			WARN("Expected CRLF sequence at end of message")
+			return false, false, false
+		}
+		s.parseOffset += 2 // skip final CRLF
+
 		m.end = s.parseOffset
 		return false, true, true
 	}
 	s.bodyReceived = 0
 	s.parseState = BODY_CHUNKED
 
-	return true, false, false
+	return true, true, false
 }
 
 func state_body_chunked(s *HttpStream, m *HttpMessage) (cont bool, ok bool, complete bool) {
+
 	if len(s.data[s.parseOffset:]) >= m.chunked_length-s.bodyReceived+2 /*\r\n*/ {
 		// Received more data than expected
 		m.chunked_body = append(m.chunked_body, s.data[s.parseOffset:s.parseOffset+m.chunked_length-s.bodyReceived]...)
 		s.parseOffset += (m.chunked_length - s.bodyReceived + 2 /*\r\n*/)
 		m.ContentLength += m.chunked_length
 		s.parseState = BODY_CHUNKED_START
-		return true, false, false
+		return true, true, false
 	} else {
 		if len(s.data[s.parseOffset:]) >= m.chunked_length-s.bodyReceived {
 			// we need need to wait for the +2, else we can crash on next call

http_test.go

@@ -619,10 +619,15 @@ func TestHttpParser_301_response(t *testing.T) {
 }
 
 func TestEatBodyChunked(t *testing.T) {
+
+	if testing.Verbose() {
+		LogInit(LOG_DEBUG, "", false, []string{"http", "httpdetailed"})
+	}
+
 	msgs := [][]byte{
 		[]byte("03\r"),
 		[]byte("\n123\r\n03\r\n123\r"),
-		[]byte("\n0\r\n"),
+		[]byte("\n0\r\n\r\n"),
 	}
 	stream := &HttpStream{
 		data:         msgs[0],
@@ -712,6 +717,69 @@ func TestEatBodyChunked(t *testing.T) {
 
 	cont, ok, complete = state_body_chunked_start(stream, message)
 	if cont != false || ok != true || complete != true {
-		t.Errorf("Wrong return values")
+		t.Error("Wrong return values", cont, ok, complete)
+	}
+}
+
+func TestEatBodyChunkedWaitCRLF(t *testing.T) {
+
+	if testing.Verbose() {
+		LogInit(LOG_DEBUG, "", false, []string{"http", "httpdetailed"})
+	}
+
+	msgs := [][]byte{
+		[]byte("03\r\n123\r\n0\r\n\r"),
+		[]byte("\n"),
+	}
+	stream := &HttpStream{
+		data:         msgs[0],
+		parseOffset:  0,
+		bodyReceived: 0,
+		parseState:   BODY_CHUNKED_START,
+	}
+	message := &HttpMessage{
+		chunked_length: 5,
+		ContentLength:  0,
+	}
+
+	cont, ok, complete := state_body_chunked_start(stream, message)
+	if cont != true || ok != true || complete != false {
+		t.Error("Wrong return values", cont, ok, complete)
+	}
+	if stream.parseState != BODY_CHUNKED {
+		t.Error("Unexpected state", stream.parseState)
+	}
+
+	cont, ok, complete = state_body_chunked(stream, message)
+	if cont != true || ok != true || complete != false {
+		t.Error("Wrong return values", cont, ok, complete)
+	}
+	if stream.parseState != BODY_CHUNKED_START {
+		t.Error("Unexpected state", stream.parseState)
+	}
+
+	cont, ok, complete = state_body_chunked_start(stream, message)
+	if cont != false || ok != true || complete != false {
+		t.Error("Wrong return values", cont, ok, complete)
+	}
+	if stream.parseState != BODY_CHUNKED_WAIT_FINAL_CRLF {
+		t.Error("Unexpected state", stream.parseState)
+	}
+
+	DEBUG("http", "parseOffset", stream.parseOffset)
+
+	ok, complete = state_body_chunked_wait_final_crlf(stream, message)
+	if ok != true || complete != false {
+		t.Error("Wrong return values", ok, complete)
+	}
+
+	stream.data = append(stream.data, msgs[1]...)
+
+	ok, complete = state_body_chunked_wait_final_crlf(stream, message)
+	if ok != true || complete != true {
+		t.Error("Wrong return values", ok, complete)
+	}
+	if message.end != 14 {
+		t.Error("Wrong message end", message.end)
 	}
 }