Libpcap support and vairous bugfixes

Merge #260

.gitignore

@@ -13,3 +13,5 @@
 .idea
 *.iml
 gor
+
+*.mprof

.travis.yml

@@ -1,3 +1,6 @@
 language: go
 go: 1.6
-script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.6)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test ./... -v -timeout 60s -race"
+script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.6)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test ./... -v -timeout 120s -race"
+
+before_install:
+  - sudo apt-get install libpcap-dev -y

Dockerfile

@@ -13,10 +13,12 @@ RUN apt-get install oracle-java8-installer -y
 RUN wget http://apache-mirror.rbc.ru/pub/apache//commons/io/binaries/commons-io-2.4-bin.tar.gz -P /tmp
 RUN tar xzf /tmp/commons-io-2.4-bin.tar.gz -C /tmp
 
+RUN apt-get install libpcap-dev -y
+RUN go get github.com/google/gopacket
+RUN go get -u github.com/golang/lint/golint
+
 WORKDIR /go/src/github.com/buger/gor/
 ADD . /go/src/github.com/buger/gor/
 
 RUN javac -cp /tmp/commons-io-2.4/commons-io-2.4.jar ./examples/middleware/echo.java
-
-RUN go get -u github.com/golang/lint/golint
 RUN go get

LICENSE.txt

@@ -1,9 +1,7 @@
-Copyright (c) Contributed Systems LLC
-
-Sidekiq is an Open Source project licensed under the terms of
+Gor is an Open Source project licensed under the terms of
 the LGPLv3 license.  Please see <http://www.gnu.org/licenses/lgpl-3.0.html>
 for license text.
 
 Gor Pro has a commercial-friendly license allowing private forks
-and modifications of Gor.  Please see http://gortool.com/pro/ for
+and modifications of Gor.  Please see http://gortool.com/#pro for
 more detail.  You can find the commercial license terms in COMM-LICENSE.

Makefile

@@ -1,6 +1,7 @@
 SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go input_http.go output_http.go output_tcp.go plugins.go settings.go test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.go middleware.go protocol.go
 SOURCE_PATH = /go/src/github.com/buger/gor/
 RUN = docker run -v `pwd`:$(SOURCE_PATH) -p 0.0.0.0:8000:8000 -t -i gor
+BENCHMARK = BenchmarkRAWInput
 
 release: release-x86 release-x64
 
@@ -24,7 +25,10 @@ race:
 	$(RUN) go test ./... $(ARGS) -v -race -timeout 15s
 
 test:
-	$(RUN) go test ./... -timeout 10s $(ARGS) -v
+	$(RUN) go test ./. -timeout 30s $(ARGS) -v
+
+test_all:
+	$(RUN) go test ./... -timeout 30s $(ARGS) -v
 
 testone:
 	$(RUN) go test ./... -timeout 4s -run $(TEST) $(ARGS) -v
@@ -40,7 +44,12 @@ vet:
 	$(RUN) go vet
 
 bench:
-	$(RUN) go test -v -run NOT_EXISTING -bench HTTP
+	$(RUN) go test -v -run NOT_EXISTING -bench $(BENCHMARK) -benchtime 5s
+
+profile_test:
+	$(RUN) go test $(LDFLAGS) -run NOT_EXISTING -test.benchmem -bench $(BENCHMARK) ./. $(ARGS) -benchtime 5s -memprofile mem.mprof -v
+	$(RUN) go test $(LDFLAGS) -run NOT_EXISTING -test.benchmem -bench $(BENCHMARK) ./. $(ARGS) -benchtime 5s -cpuprofile cpu.out -v
+	$(RUN) go test $(LDFLAGS) -run NOT_EXISTING -test.benchmem -bench $(BENCHMARK) ./. $(ARGS) -c
 
 # Used mainly for debugging, because docker container do not have access to parent machine ports
 run:

README.md

@@ -41,15 +41,6 @@ It's recommended to use separate server for replaying traffic, but if you have e
 sudo gor --input-raw :80 --output-http "http://staging.com"
 ```
 
-### Guarantee of replay and HTTP input
-Due to how traffic interception works, there is chance of missing requests. If you want guarantee that requests will be replayed you can use http input, but it will require changes in your app as well. 
-
-```
-sudo gor --input-http :28019 --output-http "http://staging.com"
-```
-
-Then in your application you should send copy (e.g. like reverse proxy) all incoming requests to Gor http input. 
-
 ## Configuration
 
 ### Forward to multiple addresses
@@ -282,6 +273,13 @@ gor --input-raw :80 --output-http "http://user:pass@staging .com"
 
 Note: This will overwrite any Authorization headers in the original request.
 
+### Traffic interception engine
+By default Gor use `libpcap` for intercepting traffic. If you have any troubles with it, you may try alternative engine: `raw_socket`.
+
+```
+sudo gor --input-raw :80 --input-raw-engine "libpcap" --output-http "http://staging.com"
+```
+
 
 ## Stats 
 

circle.yml

@@ -0,0 +1,7 @@
+dependencies:
+  pre:
+    - sudo apt-get install libpcap-dev -y
+
+test:
+  override:
+    - sudo bash -l -c "export GOPATH='/home/ubuntu/.go_workspace:/usr/local/go_workspace:/home/ubuntu/.go_project' && GORACE='halt_on_error=1' /usr/local/go/bin/go test ./... -v -timeout 120s -race"

examples/middleware/token_modifier.go

@@ -45,7 +45,7 @@ func main() {
 		buf := make([]byte, len(encoded)/2)
 		hex.Decode(buf, encoded)
 
-		go process(buf)
+		process(buf)
 	}
 }
 

http_client.go

@@ -8,9 +8,9 @@ import (
 	"net"
 	"net/url"
 	"runtime/debug"
+	"strconv"
 	"strings"
 	"time"
-	"strconv"
 )
 
 var defaultPorts = map[string]string{

input_raw.go

@@ -14,15 +14,23 @@ type RAWInput struct {
 	address  string
 	expire   time.Duration
 	quit     chan bool
+	engine   int
 	listener *raw.Listener
 }
 
+// Available engines for intercepting traffic
+const (
+	EngineRawSocket = 1 << iota
+	EnginePcap
+)
+
 // NewRAWInput constructor for RAWInput. Accepts address with port as argument.
-func NewRAWInput(address string, expire time.Duration) (i *RAWInput) {
+func NewRAWInput(address string, engine int, expire time.Duration) (i *RAWInput) {
 	i = new(RAWInput)
 	i.data = make(chan *raw.TCPMessage)
 	i.address = address
 	i.expire = expire
+	i.engine = engine
 	i.quit = make(chan bool)
 
 	go i.listen(address)
@@ -59,7 +67,7 @@ func (i *RAWInput) listen(address string) {
 		log.Fatal("input-raw: error while parsing address", err)
 	}
 
-	i.listener = raw.NewListener(host, port, i.expire)
+	i.listener = raw.NewListener(host, port, i.engine, i.expire)
 
 	for {
 		select {
@@ -76,7 +84,7 @@ func (i *RAWInput) listen(address string) {
 }
 
 func (i *RAWInput) String() string {
-	return "RAW Socket input: " + i.address
+	return "Intercepting traffic from: " + i.address
 }
 
 func (i *RAWInput) Close() {

input_raw_test.go

@@ -5,17 +5,18 @@ import (
 	"io"
 	"io/ioutil"
 	"log"
+	"math/rand"
 	"net/http"
 	"net/http/httptest"
 	"net/http/httputil"
 	"os"
 	"os/exec"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
-	"math/rand"
 )
 
 const testRawExpire = time.Millisecond * 200
@@ -30,7 +31,7 @@ func TestRAWInput(t *testing.T) {
 
 	var respCounter, reqCounter int64
 
-	input := NewRAWInput(originAddr, testRawExpire)
+	input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
 	defer input.Close()
 
 	output := NewTestOutput(func(data []byte) {
@@ -52,14 +53,14 @@ func TestRAWInput(t *testing.T) {
 
 	client := NewHTTPClient(origin.URL, &HTTPClientConfig{})
 
-	time.Sleep(time.Millisecond)
-
 	go Start(quit)
+	time.Sleep(100 * time.Millisecond)
 
 	for i := 0; i < 100; i++ {
 		// request + response
 		wg.Add(2)
 		client.Get("/")
+		time.Sleep(2 * time.Millisecond)
 	}
 
 	wg.Wait()
@@ -70,7 +71,7 @@ func TestInputRAW100Expect(t *testing.T) {
 	wg := new(sync.WaitGroup)
 	quit := make(chan int)
 
-	fileContent, _ := ioutil.ReadFile("LICENSE.txt")
+	fileContent, _ := ioutil.ReadFile("COMM-LICENSE")
 
 	// Origing and Replay server initialization
 	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -82,7 +83,7 @@ func TestInputRAW100Expect(t *testing.T) {
 
 	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr, time.Second)
+	input := NewRAWInput(originAddr, EnginePcap, time.Second)
 	defer input.Close()
 
 	// We will use it to get content of raw HTTP request
@@ -117,10 +118,11 @@ func TestInputRAW100Expect(t *testing.T) {
 	Plugins.Outputs = []io.Writer{testOutput, httpOutput}
 
 	go Start(quit)
+	time.Sleep(100 * time.Millisecond)
 
 	// Origin + Response/Request Test Output + Request Http Output
 	wg.Add(4)
-	curl := exec.Command("curl", "http://"+originAddr, "--data-binary", "@LICENSE.txt")
+	curl := exec.Command("curl", "http://"+originAddr, "--data-binary", "@COMM-LICENSE")
 	err := curl.Run()
 	if err != nil {
 		log.Fatal(err)
@@ -145,7 +147,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 	}))
 
 	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
-	input := NewRAWInput(originAddr, time.Second)
+	input := NewRAWInput(originAddr, EnginePcap, time.Second)
 	defer input.Close()
 
 	replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -167,10 +169,11 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 	Plugins.Outputs = []io.Writer{httpOutput}
 
 	go Start(quit)
+	time.Sleep(100 * time.Millisecond)
 
 	wg.Add(2)
 
-	curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--data-binary", "@README.md")
+	curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--header", "Expect:", "--data-binary", "@README.md")
 	err := curl.Run()
 	if err != nil {
 		log.Fatal(err)
@@ -188,9 +191,10 @@ func TestInputRAWLargePayload(t *testing.T) {
 	}
 	wg := new(sync.WaitGroup)
 	quit := make(chan int)
+	sizeKb := 100
 
 	// Generate 100kb file
-	dd := exec.Command("dd", "if=/dev/urandom", "of=/tmp/large", "bs=1KB", "count=100")
+	dd := exec.Command("dd", "if=/dev/urandom", "of=/tmp/large", "bs=1KB", "count="+strconv.Itoa(sizeKb))
 	err := dd.Run()
 	if err != nil {
 		log.Fatal("dd error:", err)
@@ -200,7 +204,7 @@ func TestInputRAWLargePayload(t *testing.T) {
 		defer req.Body.Close()
 		body, _ := ioutil.ReadAll(req.Body)
 
-		if len(body) != 100*1000 {
+		if len(body) != sizeKb*1000 {
 			t.Error("File size should be 1mb:", len(body))
 		}
 
@@ -208,17 +212,18 @@ func TestInputRAWLargePayload(t *testing.T) {
 	}))
 	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr, time.Second)
+	input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
 	defer input.Close()
 
 	replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024)
-		buf := make([]byte, 1*1024*1024)
-		n, _ := req.Body.Read(buf)
-		body := buf[0:n]
+		body, _ := ioutil.ReadAll(req.Body)
+		// // req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024)
+		// // buf := make([]byte, 1*1024*1024)
+		// n, _ := req.Body.Read(buf)
+		// body := buf[0:n]
 
-		if len(body) != 100*1000 {
-			t.Error("File size should be 100000 bytes:", len(body))
+		if len(body) != sizeKb*1000 {
+			t.Errorf("File size should be %d bytes: %d", sizeKb*1000, len(body))
 		}
 
 		wg.Done()
@@ -232,8 +237,10 @@ func TestInputRAWLargePayload(t *testing.T) {
 
 	go Start(quit)
 
+	time.Sleep(100 * time.Millisecond)
+
 	wg.Add(2)
-	curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--data-binary", "@/tmp/large")
+	curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--header", "Expect:", "--data-binary", "@/tmp/large")
 	err = curl.Run()
 	if err != nil {
 		log.Fatal("curl error:", err)
@@ -252,7 +259,7 @@ func BenchmarkRAWInput(b *testing.B) {
 
 	var respCounter, reqCounter int64
 
-	input := NewRAWInput(originAddr, testRawExpire)
+	input := NewRAWInput(originAddr, EnginePcap, testRawExpire)
 	defer input.Close()
 
 	output := NewTestOutput(func(data []byte) {
@@ -278,9 +285,9 @@ func BenchmarkRAWInput(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		wg := new(sync.WaitGroup)
 		wg.Add(10 * 100)
-		emitted += 10*100
+		emitted += 10 * 100
 		for w := 0; w < 100; w++ {
-			go func(){
+			go func() {
 				client := NewHTTPClient(origin.URL, &HTTPClientConfig{})
 				for i := 0; i < 10; i++ {
 					if rand.Int63n(2) == 0 {
@@ -296,7 +303,7 @@ func BenchmarkRAWInput(b *testing.B) {
 		wg.Wait()
 	}
 
-	time.Sleep(201 * time.Millisecond)
+	time.Sleep(400 * time.Millisecond)
 	log.Println("Emitted ", emitted, ", Captured ", reqCounter, "requests and ", respCounter, " responses")
 
 	close(quit)

middleware_test.go

@@ -116,7 +116,8 @@ func TestEchoMiddleware(t *testing.T) {
 	Settings.middleware = "./examples/middleware/echo.sh"
 
 	// Catch traffic from one service
-	input := NewRAWInput(from.Listener.Addr().String(), testRawExpire)
+	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
+	input := NewRAWInput(fromAddr, EnginePcap, testRawExpire)
 	defer input.Close()
 
 	// And redirect to another
@@ -129,7 +130,7 @@ func TestEchoMiddleware(t *testing.T) {
 	go Start(quit)
 
 	// Wait till middleware initialization
-	time.Sleep(10*time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
 
 	// Should receive 2 requests from original + 2 from replayed
 	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false})
@@ -138,7 +139,9 @@ func TestEchoMiddleware(t *testing.T) {
 		wg.Add(4)
 		// Request should be echoed
 		client.Get("/a")
+		time.Sleep(5 * time.Millisecond)
 		client.Get("/b")
+		time.Sleep(5 * time.Millisecond)
 	}
 
 	wg.Wait()
@@ -176,7 +179,7 @@ func TestTokenMiddleware(t *testing.T) {
 
 	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 	// Catch traffic from one service
-	input := NewRAWInput(fromAddr, testRawExpire)
+	input := NewRAWInput(fromAddr, EnginePcap, testRawExpire)
 	defer input.Close()
 
 	// And redirect to another

plugins.go

@@ -79,8 +79,13 @@ func InitPlugins() {
 		registerPlugin(NewDummyOutput, options)
 	}
 
+	engine := EnginePcap
+	if Settings.inputRAWEngine == "raw_socket" {
+		engine = EngineRawSocket
+	}
+
 	for _, options := range Settings.inputRAW {
-		registerPlugin(NewRAWInput, options, time.Duration(0))
+		registerPlugin(NewRAWInput, engine, options, time.Duration(0))
 	}
 
 	for _, options := range Settings.inputTCP {

raw_socket_listener/listener.go

@@ -15,16 +15,25 @@ package rawSocket
 import (
 	"bytes"
 	"encoding/binary"
+	"fmt"
+	"github.com/google/gopacket"
+	_ "github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcap"
+	"io"
 	"log"
 	"net"
 	"runtime/debug"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
+var _ = fmt.Println
+
 // Listener handle traffic capture
 type Listener struct {
+	mu sync.Mutex
 	// buffer of TCPMessages waiting to be send
 	// ID -> TCPMessage
 	messages map[string]*TCPMessage
@@ -57,12 +66,19 @@ type Listener struct {
 }
 
 type request struct {
+	id string
 	start time.Time
 	ack   uint32
 }
 
+// Available engines for intercepting traffic
+const (
+	EngineRawSocket = 1 << iota
+	EnginePcap
+)
+
 // NewListener creates and initializes new Listener object
-func NewListener(addr string, port string, expire time.Duration) (l *Listener) {
+func NewListener(addr string, port string, engine int, expire time.Duration) (l *Listener) {
 	l = &Listener{}
 
 	l.packetsChan = make(chan *TCPPacket, 10000)
@@ -86,15 +102,34 @@ func NewListener(addr string, port string, expire time.Duration) (l *Listener) {
 	l.messageExpire = expire
 
 	go l.listen()
+	go l.processPackets()
 
 	// Special case for testing
 	if l.port != 0 {
-		go l.readRAWSocket()
+		switch engine {
+		case EngineRawSocket:
+			go l.readRAWSocket()
+		case EnginePcap:
+			go l.readPcap()
+		default:
+			log.Fatal("Unknown traffic interception engine:", engine)
+		}
 	}
 
 	return
 }
 
+func (t *Listener) processPackets() {
+	for {
+		// We need to use channels to process each packet to avoid data races
+		packet := <-t.packetsChan
+		// log.Println(packet)
+		t.mu.Lock()
+		t.processTCPPacket(packet)
+		t.mu.Unlock()
+	}
+}
+
 func (t *Listener) listen() {
 	gcTicker := time.Tick(t.messageExpire / 2)
 
@@ -105,36 +140,41 @@ func (t *Listener) listen() {
 				t.conn.Close()
 			}
 			return
-		// We need to use channels to process each packet to avoid data races
-		case packet := <-t.packetsChan:
-			maxLen := len(packet.Data)
-			if maxLen > 500 {
-				maxLen = 500
-			}
-
-			t.processTCPPacket(packet)
-
 		case <-gcTicker:
 			now := time.Now()
+			// log.Println("GC")
 
+			t.mu.Lock()
+			// Dispatch requests before responses
 			for _, message := range t.messages {
-				if now.Sub(message.Start) >= t.messageExpire {
+				if now.Sub(message.End) >= t.messageExpire {
 					t.dispatchMessage(message)
 				}
 			}
+
+			t.mu.Unlock()
 		}
 	}
 }
 
 func (t *Listener) dispatchMessage(message *TCPMessage) {
+	// If already dispatched
+	if _, ok := t.messages[message.ID]; !ok {
+		return
+	}
+
 	delete(t.ackAliases, message.Ack)
 	delete(t.messages, message.ID)
 
+	// log.Println("Dispatching, message", message.Seq, message.Ack, string(message.Bytes()))
+
 	if message.IsIncoming {
 		// If there were response before request
+		// log.Println("Looking for Response: ", t.respWithoutReq, message.ResponseAck)
 		if respID, ok := t.respWithoutReq[message.ResponseAck]; ok {
 			if resp, rok := t.messages[respID]; rok {
 				if resp.RequestAck == 0 {
+					// log.Println("FOUND RESPONSE")
 					resp.RequestAck = message.Ack
 					resp.RequestStart = message.Start
 
@@ -145,11 +185,19 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
 			}
 		}
 	} else {
+		if message.RequestAck == 0 {
+			if responseRequest, ok := t.respAliases[message.Ack]; ok {
+				message.RequestStart = responseRequest.start
+				message.RequestAck = responseRequest.ack
+			}
+		}
+
 		delete(t.respAliases, message.Ack)
 		delete(t.respWithoutReq, message.Ack)
 
 		// Do not track responses which have no associated requests
 		if message.RequestAck == 0 {
+			// log.Println("Can't dispatch resp", message.Seq, message.Ack, string(message.Bytes()))
 			return
 		}
 	}
@@ -157,6 +205,102 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
 	t.messagesChan <- message
 }
 
+// DeviceNotFoundError raised if user specified wrong ip
+type DeviceNotFoundError struct {
+	addr string
+}
+
+func (e *DeviceNotFoundError) Error() string {
+	devices, _ := pcap.FindAllDevs()
+
+	var msg string
+	msg += "Devices with addr: " + e.addr + " not found. Available devices: \n"
+	for _, device := range devices {
+		msg += "Name: " + device.Name + "\n"
+		msg += "Description: " + device.Description + "\n"
+		msg += "Devices addresses: " + device.Description + "\n"
+		for _, address := range device.Addresses {
+			msg += "- IP address: " + address.IP.String() + "\n"
+			msg += "- Subnet mask: " + address.Netmask.String() + "\n"
+		}
+	}
+
+	return msg
+}
+
+func findPcapDevice(addr string) (*pcap.Interface, error) {
+	devices, err := pcap.FindAllDevs()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	for _, device := range devices {
+		if device.Name == "any" && addr == "" || addr == "0.0.0.0" {
+			return &device, nil
+		}
+
+		for _, address := range device.Addresses {
+			if address.IP.String() == addr {
+				return &device, nil
+			}
+		}
+	}
+
+	return nil, &DeviceNotFoundError{addr}
+}
+
+func (t *Listener) readPcap() {
+	device, err := findPcapDevice(t.addr)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	handle, err := pcap.OpenLive(device.Name, 65536, true, t.messageExpire)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer handle.Close()
+
+	if err := handle.SetBPFFilter("tcp and port " + strconv.Itoa(int(t.port))); err != nil {
+		log.Fatal(err)
+	}
+
+	source := gopacket.NewPacketSource(handle, handle.LinkType())
+	source.Lazy = true
+	source.NoCopy = true
+
+	// log.Println(handle.Stats())
+
+	for {
+		packet, err := source.NextPacket()
+
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			continue
+		}
+
+		// Skip ethernet layer, 14 bytes
+		data := packet.Data()[14:]
+		ihl := uint8(data[0]) & 0x0F
+		srcIP := data[12:16]
+		data = data[ihl*4:]
+
+		dataOffset := (data[12] & 0xF0) >> 4
+
+		// We need only packets with data inside
+		// Check that the buffer is larger than the size of the TCP header
+		if len(data) > int(dataOffset*4) {
+			newBuf := make([]byte, len(data))
+			copy(newBuf, data)
+
+			go func(newBuf []byte) {
+				t.packetsChan <- ParseTCPPacket(net.IP(srcIP).String(), newBuf)
+			}(newBuf)
+		}
+	}
+}
+
 func (t *Listener) readRAWSocket() {
 	conn, e := net.ListenPacket("ip4:tcp", t.addr)
 	t.conn = conn
@@ -187,7 +331,7 @@ func (t *Listener) readRAWSocket() {
 				copy(newBuf, buf[:n])
 
 				go func(newBuf []byte) {
-					t.packetsChan <- ParseTCPPacket(addr, newBuf)
+					t.packetsChan <- ParseTCPPacket(addr.String(), newBuf)
 				}(newBuf)
 			}
 		}
@@ -230,14 +374,32 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 		}
 	}()
 
+	// log.Println("Processing packet:", packet.Ack, packet.Seq, string(packet.Data))
+
 	var message *TCPMessage
 
 	isIncoming := packet.DestPort == t.port
 
+	// Seek for 100-expect chunks
 	if parentAck, ok := t.seqWithData[packet.Seq]; ok {
+		// log.Println("Found data package with Ack:", packet.Ack)
+		// In case if non-first data chunks comes first
+		for _id, m := range t.messages {
+			// log.Println("Message ack:", m.Ack, m.packets[0].Addr, packet.Addr)
+			if m.Ack == packet.Ack && m.packets[0].Addr == packet.Addr {
+				delete(t.messages, _id)
+
+				for _, pkt := range m.packets {
+					pkt.Ack = parentAck
+					// Re-queue this packets
+					t.processTCPPacket(pkt)
+				}
+			}
+		}
+
+		delete(t.seqWithData, packet.Seq)
 		t.ackAliases[packet.Ack] = parentAck
 		packet.Ack = parentAck
-		delete(t.seqWithData, packet.Seq)
 	}
 
 	if alias, ok := t.ackAliases[packet.Ack]; ok {
@@ -250,7 +412,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 		responseRequest, _ = t.respAliases[packet.Ack]
 	}
 
-	mID := packet.Addr.String() + strconv.Itoa(int(packet.DestPort)) + strconv.Itoa(int(packet.Ack))
+	mID := packet.Addr + strconv.Itoa(int(packet.DestPort)) + strconv.Itoa(int(packet.Ack))
 
 	message, ok := t.messages[mID]
 
@@ -262,6 +424,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 			if responseRequest != nil {
 				message.RequestStart = responseRequest.start
 				message.RequestAck = responseRequest.ack
+				message.RequestID = responseRequest.id
 			} else {
 				t.respWithoutReq[packet.Ack] = mID
 			}
@@ -276,7 +439,9 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 			t.seqWithData[seq] = packet.Ack
 
 			// In case if sequence packet came first
+			// log.Println("Looking for sequences:", seq, t.messages)
 			for _id, m := range t.messages {
+				// log.Println("SeqSEQ", m.Seq, len(m.packets))
 				if m.Seq == seq {
 					t.ackAliases[m.Ack] = packet.Ack
 
@@ -290,27 +455,41 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 
 			// Removing `Expect: 100-continue` header
 			packet.Data = append(packet.Data[:len(packet.Data)-24], packet.Data[len(packet.Data)-2:]...)
+
+			// log.Println(string(packet.Data))
 		}
 	}
 
-	if isIncoming {
-		// If message have multiple packets, delete previous alias
-		if len(message.packets) > 0 {
-			delete(t.respAliases, message.ResponseAck)
-		}
-
-		responseAck := packet.Seq + uint32(message.BodySize()) + uint32(len(packet.Data))
-		t.respAliases[responseAck] = &request{message.Start, message.Ack}
-
-		message.ResponseAck = responseAck
-	}
-
 	// Adding packet to message
 	message.AddPacket(packet)
 
+	if isIncoming {
+		// If message have multiple packets, delete previous alias
+		if len(message.packets) > 1 {
+			delete(t.respAliases, message.ResponseAck)
+		}
+
+		message.UpdateResponseAck()
+		t.respAliases[message.ResponseAck] = &request{message.ID, message.Start, message.Ack}
+	}
+
 	// If message contains only single packet immediately dispatch it
 	if message.IsFinished() {
-		t.dispatchMessage(message)
+		if isIncoming {
+			if resp, ok := t.messages[message.ResponseID()]; ok {
+				t.dispatchMessage(message)
+				if resp.IsFinished() {
+					t.dispatchMessage(resp)
+				}
+			}
+		} else {
+			if req, ok := t.messages[message.RequestID]; ok {
+				if req.IsFinished() {
+					t.dispatchMessage(req)
+					t.dispatchMessage(message)
+				}
+			}
+		}
 	}
 }
 

raw_socket_listener/listener_test.go

@@ -10,21 +10,21 @@ import (
 func TestRawListenerInput(t *testing.T) {
 	var req, resp *TCPMessage
 
-	listener := NewListener("", "0", 10*time.Millisecond)
+	listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
 	defer listener.Close()
 
 	reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1"))
 
-	listener.packetsChan <- reqPacket
-
 	respAck := reqPacket.Seq + uint32(len(reqPacket.Data))
 	respPacket := buildPacket(false, respAck, reqPacket.Seq+1, []byte("HTTP/1.1 200 OK"))
-	listener.packetsChan <- respPacket
+
+	listener.processTCPPacket(reqPacket)
+	listener.processTCPPacket(respPacket)
 
 	select {
 	case req = <-listener.messagesChan:
 	case <-time.After(time.Millisecond):
-		t.Error("Should return respose immediately")
+		t.Error("Should return request immediately")
 		return
 	}
 
@@ -34,7 +34,7 @@ func TestRawListenerInput(t *testing.T) {
 
 	select {
 	case resp = <-listener.messagesChan:
-	case <-time.After(time.Millisecond):
+	case <-time.After(20 * time.Millisecond):
 		t.Error("Should return response immediately")
 		return
 	}
@@ -47,15 +47,15 @@ func TestRawListenerInput(t *testing.T) {
 func TestRawListenerResponse(t *testing.T) {
 	var req, resp *TCPMessage
 
-	listener := NewListener("", "0", 10*time.Millisecond)
+	listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
 	defer listener.Close()
 
 	reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1"))
 	respPacket := buildPacket(false, 1+uint32(len(reqPacket.Data)), 2, []byte("HTTP/1.1 200 OK"))
 
 	// If response packet comes before request
-	listener.packetsChan <- respPacket
-	listener.packetsChan <- reqPacket
+	listener.processTCPPacket(respPacket)
+	listener.processTCPPacket(reqPacket)
 
 	select {
 	case req = <-listener.messagesChan:
@@ -87,7 +87,7 @@ func TestRawListenerResponse(t *testing.T) {
 func TestRawListener100Continue(t *testing.T) {
 	var req, resp *TCPMessage
 
-	listener := NewListener("", "0", 10*time.Millisecond)
+	listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
 	defer listener.Close()
 
 	reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\nExpect: 100-continue\r\n\r\n"))
@@ -98,7 +98,7 @@ func TestRawListener100Continue(t *testing.T) {
 	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
 
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
-	respPacket2 := buildPacket(false, reqPacket3.Seq+2 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
 
 	listener.processTCPPacket(reqPacket1)
 	listener.processTCPPacket(reqPacket2)
@@ -107,6 +107,63 @@ func TestRawListener100Continue(t *testing.T) {
 	listener.processTCPPacket(respPacket1)
 	listener.processTCPPacket(respPacket2)
 
+	select {
+	case req = <-listener.messagesChan:
+		break
+	case <-time.After(11 * time.Millisecond):
+		t.Error("Should return request after expire time")
+		return
+	}
+
+	if !bytes.Equal(req.Bytes(), []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab")) {
+		t.Error("Should receive full message", string(req.Bytes()))
+	}
+
+	if !req.IsIncoming {
+		t.Error("Should be request")
+	}
+
+	select {
+	case resp = <-listener.messagesChan:
+		break
+	case <-time.After(21 * time.Millisecond):
+		t.Error("Should return response after expire time")
+		return
+	}
+
+	if resp.IsIncoming {
+		t.Error("Should be response")
+	}
+
+	if !bytes.Equal(resp.UUID(), req.UUID()) {
+		t.Error("Resp and Req UUID should be equal")
+	}
+}
+
+// Response comes before Request
+func TestRawListener100ContinueWrongOrder(t *testing.T) {
+	var req, resp *TCPMessage
+
+	listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
+	defer listener.Close()
+
+	reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\nExpect: 100-continue\r\n\r\n"))
+	// Packet with data have different Seq
+	reqPacket2 := buildPacket(true, 2, reqPacket1.Seq+uint32(len(reqPacket1.Data)), []byte("a"))
+	reqPacket3 := buildPacket(true, 2, reqPacket2.Seq+1, []byte("b"))
+
+	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
+
+	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
+	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+
+	listener.processTCPPacket(respPacket1)
+	listener.processTCPPacket(respPacket2)
+
+	listener.processTCPPacket(reqPacket1)
+	listener.processTCPPacket(reqPacket2)
+	listener.processTCPPacket(reqPacket3)
+
 	select {
 	case req = <-listener.messagesChan:
 		break
@@ -140,59 +197,95 @@ func TestRawListener100Continue(t *testing.T) {
 	}
 }
 
-// Response comes before Request
-func TestRawListener100ContinueWrongOrder(t *testing.T) {
-    var req, resp *TCPMessage
+func testChunkedSequence(t *testing.T, listener *Listener, packets ...*TCPPacket) {
+	var r, req, resp *TCPMessage
 
-    listener := NewListener("", "0", 10*time.Millisecond)
-    defer listener.Close()
+	for _, p := range packets {
+		listener.processTCPPacket(p)
+	}
 
-    reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\nExpect: 100-continue\r\n\r\n"))
-    // Packet with data have different Seq
-    reqPacket2 := buildPacket(true, 2, reqPacket1.Seq+uint32(len(reqPacket1.Data)), []byte("a"))
-    reqPacket3 := buildPacket(true, 2, reqPacket2.Seq+1, []byte("b"))
+	select {
+	case r = <-listener.messagesChan:
+		if r.IsIncoming {
+			req = r
+		} else {
+			resp = r
+		}
+		break
+	case <-time.After(25 * time.Millisecond):
+		t.Error("Should return request after expire time")
+		return
+	}
+	select {
+	case r = <-listener.messagesChan:
+		if r.IsIncoming {
+			req = r
+		} else {
+			resp = r
+		}
+		break
+	case <-time.After(25 * time.Millisecond):
+		t.Error("Should return request after expire time")
+		return
+	}
 
-    respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
+	if !bytes.Equal(req.Bytes(), []byte("POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n1\r\na\r\n1\r\nb\r\n0\r\n\r\n")) {
+		t.Error("Should receive full message", string(req.Bytes()))
+	}
 
-    // panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
-    respPacket2 := buildPacket(false, reqPacket3.Seq+2 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+	if !req.IsIncoming {
+		t.Error("Should be request")
+	}
 
-    listener.processTCPPacket(respPacket1)
-    listener.processTCPPacket(respPacket2)
+	if resp.IsIncoming {
+		t.Error("Should be response")
+	}
 
-    listener.processTCPPacket(reqPacket1)
-    listener.processTCPPacket(reqPacket2)
-    listener.processTCPPacket(reqPacket3)
+	if !bytes.Equal(resp.UUID(), req.UUID()) {
+		t.Error("Resp and Req UUID should be equal", string(resp.UUID()), string(req.UUID()))
+	}
 
-    select {
-    case req = <-listener.messagesChan:
-        break
-    case <-time.After(11 * time.Millisecond):
-        t.Error("Should return response after expire time")
-        return
-    }
+	time.Sleep(15 * time.Millisecond)
 
-    if !bytes.Equal(req.Bytes(), []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab")) {
-        t.Error("Should receive full message", string(req.Bytes()))
-    }
-
-    if !req.IsIncoming {
-        t.Error("Should be request")
-    }
-
-    select {
-    case resp = <-listener.messagesChan:
-        break
-    case <-time.After(21 * time.Millisecond):
-        t.Error("Should return response after expire time")
-        return
-    }
-
-    if resp.IsIncoming {
-        t.Error("Should be response")
-    }
-
-    if !bytes.Equal(resp.UUID(), req.UUID()) {
-        t.Error("Resp and Req UUID should be equal")
-    }
+	if len(listener.messages) != 0 {
+		t.Error("Messages non empty:", listener.messages)
+	}
+}
+
+func permutation(n int, list []*TCPPacket) []*TCPPacket {
+	if len(list) == 1 {
+		return list
+	}
+
+	k := n % len(list)
+
+	first := []*TCPPacket{list[k]}
+	next := make([]*TCPPacket, len(list)-1)
+
+	copy(next, append(list[:k], list[k+1:]...))
+
+	return append(first, permutation(n/len(list), next)...)
+}
+
+// Response comes before Request
+func TestRawListenerChunkedWrongOrder(t *testing.T) {
+	listener := NewListener("", "0", EnginePcap, 10*time.Millisecond)
+	defer listener.Close()
+
+	reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nExpect: 100-continue\r\n\r\n"))
+	// Packet with data have different Seq
+	reqPacket2 := buildPacket(true, 2, reqPacket1.Seq+uint32(len(reqPacket1.Data)), []byte("1\r\na\r\n"))
+	reqPacket3 := buildPacket(true, 2, reqPacket2.Seq+uint32(len(reqPacket2.Data)), []byte("1\r\nb\r\n"))
+	reqPacket4 := buildPacket(true, 2, reqPacket3.Seq+uint32(len(reqPacket3.Data)), []byte("0\r\n\r\n"))
+
+	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
+
+	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
+	respPacket2 := buildPacket(false, reqPacket4.Seq+5 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+
+	// Should re-construct message from all possible combinations
+	for i := 0; i < 6*5*4*3*2*1; i++ {
+		packets := permutation(i, []*TCPPacket{reqPacket1, reqPacket2, reqPacket3, reqPacket4, respPacket1, respPacket2})
+		testChunkedSequence(t, listener, packets...)
+	}
 }

raw_socket_listener/tcp_message.go

@@ -23,6 +23,7 @@ type TCPMessage struct {
 	ResponseAck  uint32
 	RequestStart time.Time
 	RequestAck   uint32
+	RequestID    string
 	Start        time.Time
 	End          time.Time
 	IsIncoming   bool
@@ -95,11 +96,23 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) {
 		// Packets not always captured in same Seq order, and sometimes we need to prepend
 		if len(t.packets) == 0 || packet.Seq > t.packets[len(t.packets)-1].Seq {
 			t.packets = append(t.packets, packet)
-		} else {
+		} else if packet.Seq < t.packets[0].Seq {
 			t.packets = append([]*TCPPacket{packet}, t.packets...)
+			t.Seq = packet.Seq // Message Seq should indicated starting seq
+		} else { // insert somewhere in the middle...
+			for i, p := range t.packets {
+				if packet.Seq < p.Seq {
+					t.packets = append(t.packets[:i], append([]*TCPPacket{packet}, t.packets[i:]...)...)
+					break
+				}
+			}
 		}
 
-		t.End = time.Now()
+		if t.IsIncoming {
+			t.End = time.Now()
+		} else {
+			t.End = time.Now().Add(time.Millisecond)
+		}
 	}
 }
 
@@ -175,3 +188,15 @@ func (t *TCPMessage) UUID() []byte {
 
 	return uuid
 }
+
+// UpdateResponseAck should be called after packet is added
+func (t *TCPMessage) UpdateResponseAck() uint32 {
+	lastPacket := t.packets[len(t.packets)-1]
+	t.ResponseAck = lastPacket.Seq + uint32(len(lastPacket.Data))
+	return t.ResponseAck
+}
+
+// ResponseID generate message ID for request response
+func (t *TCPMessage) ResponseID() string {
+	return t.packets[0].Addr + strconv.Itoa(int(t.packets[0].SrcPort)) + strconv.Itoa(int(t.ResponseAck))
+}

raw_socket_listener/tcp_message_test.go

@@ -3,14 +3,13 @@ package rawSocket
 import (
 	"bytes"
 	_ "log"
-	"net"
 	"strconv"
 	"testing"
 )
 
 func buildPacket(isIncoming bool, Ack, Seq uint32, Data []byte) (packet *TCPPacket) {
 	packet = &TCPPacket{
-		Addr: &net.IPAddr{net.IP{}, ""},
+		Addr: "",
 		Ack:  Ack,
 		Seq:  Seq,
 		Data: Data,
@@ -27,7 +26,7 @@ func buildPacket(isIncoming bool, Ack, Seq uint32, Data []byte) (packet *TCPPack
 }
 
 func buildMessage(p *TCPPacket) *TCPMessage {
-	id := p.Addr.String() + strconv.Itoa(int(p.DestPort)) + strconv.Itoa(int(p.Ack))
+	id := p.Addr + strconv.Itoa(int(p.DestPort)) + strconv.Itoa(int(p.Ack))
 
 	isIncoming := false
 	if p.SrcPort == 1 {

raw_socket_listener/tcp_packet.go

@@ -2,7 +2,6 @@ package rawSocket
 
 import (
 	"encoding/binary"
-	"net"
 	"strconv"
 	"strings"
 )
@@ -35,11 +34,11 @@ type TCPPacket struct {
 
 	Data []byte
 
-	Addr net.Addr
+	Addr string
 }
 
 // ParseTCPPacket takes address and tcp payload and returns parsed TCPPacket
-func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
+func ParseTCPPacket(addr string, b []byte) (p *TCPPacket) {
 	p = &TCPPacket{Data: b}
 	p.ParseBasic()
 	p.Addr = addr
@@ -75,7 +74,7 @@ func (t *TCPPacket) String() string {
 	}
 
 	return strings.Join([]string{
-		"Addr: " + t.Addr.String(),
+		"Addr: " + t.Addr,
 		"Source port: " + strconv.Itoa(int(t.SrcPort)),
 		"Dest port:" + strconv.Itoa(int(t.DestPort)),
 		"Sequence:" + strconv.Itoa(int(t.Seq)),

settings.go

@@ -42,6 +42,7 @@ type AppSettings struct {
 	outputFile MultiOption
 
 	inputRAW MultiOption
+	inputRAWEngine string
 
 	middleware string
 
@@ -82,6 +83,8 @@ func init() {
 
 	flag.Var(&Settings.inputRAW, "input-raw", "Capture traffic from given port (use RAW sockets and require *sudo* access):\n\t# Capture traffic from 8080 port\n\tgor --input-raw :8080 --output-http staging.com")
 
+	flag.StringVar(&Settings.inputRAWEngine, "input-raw-engine", "libpcap", "Intercept traffic using `libpcap` (default), and `raw_socket`")
+
 	flag.StringVar(&Settings.middleware, "middleware", "", "Used for modifying traffic using external command")
 
 	flag.Var(&Settings.inputHTTP, "input-http", "Read requests from HTTP, should be explicitly sent from your application:\n\t# Listen for http on 9000\n\tgor --input-http :9000 --output-http staging.com")