Improve Full Packet parsing for HTTP (#1029)

This PR does two primary things: 
- ensures that HasFullPayload returns false if the packet doesnt start with a valid request or response header. This was necessary because a chunked response would return true from HasFullPayload if it got contiguous packets including the last packet (with the trailer) before it got the first packet (with the header).
- When the request payload is chunked across multiple packets, only correct for 100-Continue responses once. In requests with > 2 packets, the Ack number is incremented for each packet in the message, sometimes resulting in packets not being correlated with each other. This ensures that the message is corrected once and only once.

proto/proto.go

@@ -19,6 +19,7 @@ package proto
 import (
 	"bufio"
 	"bytes"
+
 	_ "fmt"
 	"net/http"
 	"net/textproto"
@@ -477,6 +478,15 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
 			m.SetProtocolState(state)
 		}
 	}
+
+	// Http Packets can only start with a few things, check if this is one of them
+	if len(payloads) == 0 {
+		return false
+	}
+	if !HasRequestTitle(payloads[0]) && !HasResponseTitle(payloads[0]) {
+		return false
+	}
+
 	if state.HeaderStart < 1 {
 		for _, data := range payloads {
 			state.HeaderStart = MIMEHeadersStartPos(data)

proto/proto_test.go

@@ -485,6 +485,14 @@ func TestHasFullPayload(t *testing.T) {
 	if got != expected {
 		t.Errorf("expected %v to equal %v", got, expected)
 	}
+
+	// check with trailer and no header
+	m = "Content-Type: text/plain\r\nContent-Length: 23\r\n\r\nMozillaDeveloperNetwork"
+	got = HasFullPayload(nil, []byte(m))
+	expected = false
+	if got != expected {
+		t.Errorf("expected %v to equal %v", got, expected)
+	}
 }
 
 func BenchmarkHasFullPayload(b *testing.B) {

tcp/tcp_message.go

@@ -64,10 +64,11 @@ type Stats struct {
 
 // Message is the representation of a tcp message
 type Message struct {
-	packets  []*Packet
-	parser   *MessageParser
-	feedback interface{}
-	Idx      uint16
+	packets          []*Packet
+	parser           *MessageParser
+	feedback         interface{}
+	Idx              uint16
+	continueAdjusted bool
 	Stats
 }
 
@@ -394,7 +395,8 @@ func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool {
 }
 
 func (parser *MessageParser) Fix100Continue(m *Message) {
-	if state, ok := m.feedback.(*proto.HTTPState); ok && state.Continue100 {
+	// Only adjust a message once
+	if state, ok := m.feedback.(*proto.HTTPState); ok && state.Continue100 && !m.continueAdjusted {
 		delete(parser.m[m.Idx], m.packets[0].MessageID())
 
 		// Shift Ack by given offset
@@ -413,6 +415,7 @@ func (parser *MessageParser) Fix100Continue(m *Message) {
 
 		// Re-add (or override) again with new message and ID
 		parser.m[m.Idx][m.packets[0].MessageID()] = m
+		m.continueAdjusted = true
 	}
 }
 
@@ -442,7 +445,7 @@ func (parser *MessageParser) timer(now time.Time, index int) {
 	packetQueueLen.Set(int64(len(parser.packets)))
 	messageQueueLen.Set(int64(len(parser.m[index])))
 
-	for _, m := range parser.m[index] {
+	for id, m := range parser.m[index] {
 		if now.Sub(m.End) > parser.messageExpire {
 			m.TimedOut = true
 			stats.Add("message_timeout_count", 1)
@@ -451,7 +454,7 @@ func (parser *MessageParser) timer(now time.Time, index int) {
 				parser.Emit(m)
 			}
 
-			delete(parser.m[index], m.packets[0].MessageID())
+			delete(parser.m[index], id)
 		}
 	}