diff --git a/elasticsearch.go b/elasticsearch.go index b83dabf..7a8bd3f 100644 --- a/elasticsearch.go +++ b/elasticsearch.go @@ -1,9 +1,9 @@ package main import ( - "net/url" "encoding/json" "log" + "net/url" "strings" //"regexp" "time" @@ -72,10 +72,10 @@ func parseURI(URI string) (err error, index string) { // check URL validity by extracting host and undex values. host := parsedUrl.Host urlPathParts := strings.Split(parsedUrl.Path, "/") - index = urlPathParts[len(urlPathParts) - 1 ] + index = urlPathParts[len(urlPathParts)-1] // force index specification in uri : ie no implicit index - if (host == "" || index == "") { + if host == "" || index == "" { err = new(ESUriErorr) } diff --git a/elasticsearch_test.go b/elasticsearch_test.go index 701c38f..27e624b 100644 --- a/elasticsearch_test.go +++ b/elasticsearch_test.go @@ -6,19 +6,19 @@ import ( const expectedIndex = "gor" -func assertExpectedGorIndex (index string, t *testing.T) { +func assertExpectedGorIndex(index string, t *testing.T) { if expectedIndex != index { t.Fatalf("Expected index %s but got %s", expectedIndex, index) } } -func assertExpectedIndex (expectedIndex string, index string, t *testing.T) { +func assertExpectedIndex(expectedIndex string, index string, t *testing.T) { if expectedIndex != index { t.Fatalf("Expected index %s but got %s", expectedIndex, index) } } -func assertExpectedError (returnedError error, t *testing.T) { +func assertExpectedError(returnedError error, t *testing.T) { expectedError := new(ESUriErorr) if expectedError != returnedError { @@ -26,7 +26,7 @@ func assertExpectedError (returnedError error, t *testing.T) { } } -func assertNoError (returnedError error, t *testing.T) { +func assertNoError(returnedError error, t *testing.T) { if nil != returnedError { t.Errorf("Expected err %s but got %s", nil, returnedError) } @@ -38,7 +38,7 @@ func assertNoError (returnedError error, t *testing.T) { func TestElasticConnectionBuildFailWithoutScheme(t *testing.T) { uri := "localhost:9200/" + expectedIndex - err, _ := parseURI(uri) + err, _ := parseURI(uri) assertExpectedError(err, t) } @@ -48,7 +48,7 @@ func TestElasticConnectionBuildFailWithoutScheme(t *testing.T) { func TestElasticConnectionBuildFailWithoutIndex(t *testing.T) { uri := "http://localhost:9200" - err, index := parseURI(uri) + err, index := parseURI(uri) assertExpectedIndex("", index, t) diff --git a/http_client.go b/http_client.go index 6bc2c11..c264e4b 100644 --- a/http_client.go +++ b/http_client.go @@ -298,9 +298,9 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { if err != nil && readBytes == 0 { maxRead := 100 - if readBytes < maxRead { - maxRead = readBytes - } + if readBytes < maxRead { + maxRead = readBytes + } Debug("[HTTPClient] Response read timeout error", err, c.conn, readBytes, string(c.respBuf[:maxRead])) response = errorPayload(HTTP_TIMEOUT) c.Disconnect() @@ -309,9 +309,9 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { if readBytes < 4 || string(c.respBuf[:4]) != "HTTP" { maxRead := 100 - if readBytes < maxRead { - maxRead = readBytes - } + if readBytes < maxRead { + maxRead = readBytes + } Debug("[HTTPClient] Response read unknown error", err, c.conn, readBytes, string(c.respBuf[:maxRead])) response = errorPayload(HTTP_UNKNOWN_ERROR) c.Disconnect() diff --git a/http_modifier.go b/http_modifier.go index 7e90208..8ed611e 100644 --- a/http_modifier.go +++ b/http_modifier.go @@ -2,9 +2,9 @@ package main import ( "bytes" - "strings" - "hash/fnv" "encoding/base64" + "hash/fnv" + "strings" "github.com/buger/goreplay/proto" ) diff --git a/http_modifier_settings.go b/http_modifier_settings.go index 5ae70e6..7291162 100644 --- a/http_modifier_settings.go +++ b/http_modifier_settings.go @@ -81,7 +81,6 @@ func (h *HTTPHeaderBasicAuthFilters) Set(value string) error { return nil } - // // Handling of --http-allow-header-hash and --http-allow-param-hash options // diff --git a/http_modifier_test.go b/http_modifier_test.go index e39e0af..ac7baba 100644 --- a/http_modifier_test.go +++ b/http_modifier_test.go @@ -97,7 +97,7 @@ func TestHTTPHeaderBasicAuthFilters(t *testing.T) { payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 88\r\nAuthorization: Basic Y3VzdG9tZXI2OnJlc3RAMTIzXlRFU1Q==\r\n\r\na=1&b=2") if len(modifier.Rewrite(payload)) == 0 { t.Error("Request should pass filters") - } + } filters = HTTPHeaderBasicAuthFilters{} // Setting filter that not match our header @@ -115,7 +115,7 @@ func TestHTTPHeaderBasicAuthFilters(t *testing.T) { payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 88\r\nAuthorization: Basic bWlja2V5IG1vdXNlOmhhcHB5MTIz\r\n\r\na=1&b=2") if len(modifier.Rewrite(payload)) == 0 { t.Error("Request should pass filters") - } + } } func TestHTTPModifierURLRewrite(t *testing.T) { diff --git a/input_raw.go b/input_raw.go index eec5002..7dddce1 100644 --- a/input_raw.go +++ b/input_raw.go @@ -20,6 +20,7 @@ type RAWInput struct { trackResponse bool listener *raw.Listener bpfFilter string + timestampType string } // Available engines for intercepting traffic @@ -30,7 +31,7 @@ const ( ) // NewRAWInput constructor for RAWInput. Accepts address with port as argument. -func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration, realIPHeader string, bpfFilter string) (i *RAWInput) { +func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration, realIPHeader string, bpfFilter string, timestampType string) (i *RAWInput) { i = new(RAWInput) i.data = make(chan *raw.TCPMessage) i.address = address @@ -40,6 +41,7 @@ func NewRAWInput(address string, engine int, trackResponse bool, expire time.Dur i.realIPHeader = []byte(realIPHeader) i.quit = make(chan bool) i.trackResponse = trackResponse + i.timestampType = timestampType i.listen(address) i.listener.IsReady() @@ -59,7 +61,7 @@ func (i *RAWInput) Read(data []byte) (int, error) { buf = proto.SetHeader(buf, i.realIPHeader, []byte(msg.IP().String())) } } else { - header = payloadHeader(ResponsePayload, msg.UUID(), msg.AssocMessage.Start.UnixNano(), msg.End.UnixNano()-msg.AssocMessage.Start.UnixNano()) + header = payloadHeader(ResponsePayload, msg.UUID(), msg.Start.UnixNano(), msg.End.UnixNano()-msg.AssocMessage.End.UnixNano()) } copy(data[0:len(header)], header) @@ -77,7 +79,7 @@ func (i *RAWInput) listen(address string) { log.Fatal("input-raw: error while parsing address", err) } - i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.bpfFilter) + i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.bpfFilter, i.timestampType) ch := i.listener.Receiver() diff --git a/input_raw_test.go b/input_raw_test.go index 099b393..cd0b319 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -44,7 +44,7 @@ func TestRAWInputIPv4(t *testing.T) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "X-Real-IP", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "X-Real-IP", "", "") defer input.Close() output := NewTestOutput(func(data []byte) { @@ -106,7 +106,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) { originAddr := listener.Addr().String() - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") defer input.Close() output := NewTestOutput(func(data []byte) { @@ -152,7 +152,7 @@ func TestRAWInputIPv6(t *testing.T) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") defer input.Close() output := NewTestOutput(func(data []byte) { @@ -203,7 +203,7 @@ func TestInputRAW100Expect(t *testing.T) { originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "") + input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "", "") defer input.Close() // We will use it to get content of raw HTTP request @@ -266,7 +266,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) { })) originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "") + input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "", "") defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -330,7 +330,7 @@ func TestInputRAWLargePayload(t *testing.T) { })) originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { @@ -375,7 +375,7 @@ func BenchmarkRAWInput(b *testing.B) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") defer input.Close() output := NewTestOutput(func(data []byte) { diff --git a/middleware_test.go b/middleware_test.go index c828b85..b8d3c9a 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -118,7 +118,7 @@ func TestEchoMiddleware(t *testing.T) { // Catch traffic from one service fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "") + input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "", "") defer input.Close() // And redirect to another @@ -180,7 +180,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, EnginePcap, true, testRawExpire, "", "") + input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "", "") defer input.Close() // And redirect to another diff --git a/output_file.go b/output_file.go index 07471f6..0145344 100644 --- a/output_file.go +++ b/output_file.go @@ -186,9 +186,9 @@ func (o *FileOutput) Write(data []byte) (n int, err error) { o.currentID = meta[1] o.payloadType = meta[0] } - + o.updateName() - + if o.file == nil || o.currentName != o.file.Name() { o.mu.Lock() o.Close() diff --git a/plugins.go b/plugins.go index 504abb7..1cb0637 100644 --- a/plugins.go +++ b/plugins.go @@ -106,7 +106,7 @@ func InitPlugins() { } for _, options := range Settings.inputRAW { - registerPlugin(NewRAWInput, options, engine, Settings.inputRAWTrackResponse, Settings.inputRAWExpire, Settings.inputRAWRealIPHeader, Settings.inputRAWBpfFilter) + registerPlugin(NewRAWInput, options, engine, Settings.inputRAWTrackResponse, Settings.inputRAWExpire, Settings.inputRAWRealIPHeader, Settings.inputRAWBpfFilter, Settings.inputRAWTimestampType) } for _, options := range Settings.inputTCP { diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 518cc06..47a1aba 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -72,7 +72,8 @@ type Listener struct { trackResponse bool messageExpire time.Duration - bpfFilter string + bpfFilter string + timestampType string conn net.PacketConn pcapHandles []*pcap.Handle @@ -95,7 +96,7 @@ const ( ) // NewListener creates and initializes new Listener object -func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, bpfFilter string) (l *Listener) { +func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, bpfFilter string, timestampType string) (l *Listener) { l = &Listener{} l.packetsChan = make(chan *packet, 10000) @@ -110,6 +111,7 @@ func NewListener(addr string, port string, engine int, trackResponse bool, expir l.respWithoutReq = make(map[uint32]tcpID) l.trackResponse = trackResponse l.bpfFilter = bpfFilter + l.timestampType = timestampType l.addr = addr _port, _ := strconv.Atoi(port) @@ -329,12 +331,31 @@ func (t *Listener) readPcap() { for _, d := range devices { go func(device pcap.Interface) { - handle, err := pcap.OpenLive(device.Name, 65536, true, t.messageExpire) + inactive, err := pcap.NewInactiveHandle(device.Name) if err != nil { log.Println("Pcap Error while opening device", device.Name, err) wg.Done() return } + + if t.timestampType != "" { + if tt, terr := pcap.TimestampSourceFromString(t.timestampType); terr != nil { + log.Println("Supported timestamp types: ", inactive.SupportedTimestamps(), device.Name) + } else if terr := inactive.SetTimestampSource(tt); terr != nil { + log.Println("Supported timestamp types: ", inactive.SupportedTimestamps(), device.Name) + } + } + inactive.SetSnapLen(65536) + inactive.SetTimeout(t.messageExpire) + inactive.SetPromisc(true) + + handle, herr := inactive.Activate() + if herr != nil { + log.Println("PCAP Activate error:", herr) + wg.Done() + return + } + defer handle.Close() t.mu.Lock() diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 839a47a..440624c 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -131,12 +131,6 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { } } - if t.IsIncoming { - t.End = time.Now() - } else { - t.End = time.Now().Add(time.Millisecond) - } - if packet.OrigAck != 0 { t.DataAck = packet.OrigAck } @@ -144,6 +138,10 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { if packet.timestamp.Before(t.Start) { t.Start = packet.timestamp } + + if t.End.IsZero() || t.End.Before(packet.timestamp) { + t.End = packet.timestamp + } } t.checkSeqIntegrity() @@ -226,7 +224,7 @@ func (t *TCPMessage) updateHeadersPacket() { return } -// checkIfComplete returns true if all of the packets that compse the message arrived. +// checkIfComplete returns true if all of the packets that compse the message arrived. func (t *TCPMessage) checkIfComplete() { if t.seqMissing || t.headerPacket == -1 { return diff --git a/settings.go b/settings.go index 4815282..3ae0204 100644 --- a/settings.go +++ b/settings.go @@ -54,6 +54,7 @@ type AppSettings struct { inputRAWRealIPHeader string inputRAWExpire time.Duration inputRAWBpfFilter string + inputRAWTimestampType string middleware string @@ -130,6 +131,8 @@ func init() { flag.StringVar(&Settings.inputRAWBpfFilter, "input-raw-bpf-filter", "", "BPF filter to write custom expressions. Can be useful in case of non standard network interfaces like tunneling or SPAN port. Example: --input-raw-bpf-filter 'dst port 80'") + flag.StringVar(&Settings.inputRAWTimestampType, "input-raw-timestamp-type", "", "Possible values: PCAP_TSTAMP_HOST, PCAP_TSTAMP_HOST_LOWPREC, PCAP_TSTAMP_HOST_HIPREC, PCAP_TSTAMP_ADAPTER, PCAP_TSTAMP_ADAPTER_UNSYNCED. This values not supported on all systems, GoReplay will tell you available values of you put wrong one.") + 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") @@ -178,9 +181,9 @@ func init() { flag.Var(&Settings.modifierConfig.headerNegativeFilters, "http-disallow-header", "A regexp to match a specific header against. Requests with matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-disallow-header \"User-Agent: Replayed by Gor\"") - flag.Var(&Settings.modifierConfig.headerBasicAuthFilters, "http-basic-auth-filter", "A regexp to match the decoded basic auth string against. Requests with non-matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-basic-auth-filter \"^customer[0-9].*\"") + flag.Var(&Settings.modifierConfig.headerBasicAuthFilters, "http-basic-auth-filter", "A regexp to match the decoded basic auth string against. Requests with non-matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-basic-auth-filter \"^customer[0-9].*\"") - flag.Var(&Settings.modifierConfig.headerHashFilters, "http-header-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header:\n\t gor --input-raw :8080 --output-http staging.com --http-header-limiter user-id:25%") + flag.Var(&Settings.modifierConfig.headerHashFilters, "http-header-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header:\n\t gor --input-raw :8080 --output-http staging.com --http-header-limiter user-id:25%") flag.Var(&Settings.modifierConfig.headerHashFilters, "output-http-header-hash-filter", "WARNING: `output-http-header-hash-filter` DEPRECATED, use `--http-header-hash-limiter` instead")