From afc8f7852e716212237965fccd38cdff69e9a95e Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Fri, 29 Apr 2016 14:48:17 +0500 Subject: [PATCH] Libpcap support and vairous bugfixes Merge #260 --- .gitignore | 2 + .travis.yml | 5 +- Dockerfile | 6 +- LICENSE.txt | 6 +- Makefile | 13 +- README.md | 16 +- circle.yml | 7 + examples/middleware/token_modifier.go | 2 +- http_client.go | 2 +- input_raw.go | 14 +- input_raw_test.go | 53 +++--- middleware_test.go | 9 +- plugins.go | 7 +- raw_socket_listener/listener.go | 235 +++++++++++++++++++++--- raw_socket_listener/listener_test.go | 209 +++++++++++++++------ raw_socket_listener/tcp_message.go | 29 ++- raw_socket_listener/tcp_message_test.go | 5 +- raw_socket_listener/tcp_packet.go | 7 +- settings.go | 3 + 19 files changed, 485 insertions(+), 145 deletions(-) create mode 100644 circle.yml diff --git a/.gitignore b/.gitignore index b9d19c7..de9ee5a 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,5 @@ .idea *.iml gor + +*.mprof diff --git a/.travis.yml b/.travis.yml index 77472cf..53df214 100644 --- a/.travis.yml +++ b/.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" \ No newline at end of file +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 \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index d915390..c4be6b0 100644 --- a/Dockerfile +++ b/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 \ No newline at end of file diff --git a/LICENSE.txt b/LICENSE.txt index b19014a..4d69dfd 100644 --- a/LICENSE.txt +++ b/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 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. diff --git a/Makefile b/Makefile index a48466f..28c760d 100644 --- a/Makefile +++ b/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: diff --git a/README.md b/README.md index 3e11c52..efffb71 100644 --- a/README.md +++ b/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 diff --git a/circle.yml b/circle.yml new file mode 100644 index 0000000..0b2ee36 --- /dev/null +++ b/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" \ No newline at end of file diff --git a/examples/middleware/token_modifier.go b/examples/middleware/token_modifier.go index 021408d..6536c0d 100644 --- a/examples/middleware/token_modifier.go +++ b/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) } } diff --git a/http_client.go b/http_client.go index 70ad56e..1b3a915 100644 --- a/http_client.go +++ b/http_client.go @@ -8,9 +8,9 @@ import ( "net" "net/url" "runtime/debug" + "strconv" "strings" "time" - "strconv" ) var defaultPorts = map[string]string{ diff --git a/input_raw.go b/input_raw.go index 3b36a20..60a495b 100644 --- a/input_raw.go +++ b/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() { diff --git a/input_raw_test.go b/input_raw_test.go index aada58a..9630921 100644 --- a/input_raw_test.go +++ b/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) diff --git a/middleware_test.go b/middleware_test.go index af7655b..5f63cf4 100644 --- a/middleware_test.go +++ b/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 diff --git a/plugins.go b/plugins.go index 8c08e2f..a958a0e 100644 --- a/plugins.go +++ b/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 { diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index b81c23b..3f6683c 100644 --- a/raw_socket_listener/listener.go +++ b/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) + } + } + } } } diff --git a/raw_socket_listener/listener_test.go b/raw_socket_listener/listener_test.go index 7b8cf54..e52796b 100644 --- a/raw_socket_listener/listener_test.go +++ b/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...) + } } diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 448e486..df8caa9 100644 --- a/raw_socket_listener/tcp_message.go +++ b/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)) +} diff --git a/raw_socket_listener/tcp_message_test.go b/raw_socket_listener/tcp_message_test.go index 5165467..f404541 100644 --- a/raw_socket_listener/tcp_message_test.go +++ b/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 { diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index ca3925a..bce68f4 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/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)), diff --git a/settings.go b/settings.go index 3671367..d308087 100644 --- a/settings.go +++ b/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")