diff --git a/Dockerfile b/Dockerfile index 1f35919..914094e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,25 +1,5 @@ -FROM golang:latest - -RUN apt-get update && apt-get install ruby vim-common -y - -# Install Java for middleware testing -RUN echo "deb http://ppa.launchpad.net/webupd8team/java/ubuntu trusty main" | tee /etc/apt/sources.list.d/webupd8team-java.list -RUN echo "deb-src http://ppa.launchpad.net/webupd8team/java/ubuntu trusty main" | tee -a /etc/apt/sources.list.d/webupd8team-java.list -RUN apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv-keys EEA14886 -RUN apt-get update -y -RUN echo oracle-java7-installer shared/accepted-oracle-license-v1-1 select true | /usr/bin/debconf-set-selections -RUN apt-get install oracle-java8-installer -y - -RUN apt-get install flex bison -y -RUN wget http://www.tcpdump.org/release/libpcap-1.7.4.tar.gz && tar xzf libpcap-1.7.4.tar.gz && cd libpcap-1.7.4 && ./configure && make install -RUN go get github.com/google/gopacket -RUN go get -u github.com/golang/lint/golint -RUN go get -u github.com/aws/aws-sdk-go - -WORKDIR /go/src/github.com/buger/gor-pro/ -ADD . /go/src/github.com/buger/gor-pro/ - -RUN wget http://archive.apache.org/dist/commons/io/binaries/commons-io-2.4-bin.tar.gz && tar xzf commons-io-2.4-bin.tar.gz && cd commons-io-2.4 && mv commons-io-2.4.jar /tmp/ -RUN wget http://archive.apache.org/dist/commons/codec/binaries/commons-codec-1.9-bin.tar.gz && tar xzf commons-codec-1.9-bin.tar.gz -RUN javac -cp commons-io-2.4/commons-io-2.4.jar -cp commons-codec-1.9/commons-codec-1.9.jar ./examples/middleware/echo.java -RUN go get +FROM alpine:latest +RUN apk update && apk add ca-certificates && update-ca-certificates && apk add openssl +RUN wget https://github.com/buger/goreplay/releases/download/v0.16.0.2/gor_0.16.0_x64.tar.gz -O gor.tar.gz +RUN tar xzf gor.tar.gz +ENTRYPOINT ./gor diff --git a/Makefile b/Makefile index d8379de..7fc6c7e 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ +SOURCE = $(shell ls -1 *.go | grep -v _test.go) SOURCE_PATH = /go/src/github.com/buger/gor-pro/ - PORT = 8000 FADDR = :8000 CONTAINER=gor-pro @@ -14,7 +14,7 @@ FADDR = ":8000" release: release-x64 release-mac release-bin: - docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i gor go build -tags netgo $(LDFLAGS) + docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i gor go build -o gor -tags netgo $(LDFLAGS) release-x64: docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i $(CONTAINER) go build -o gor -tags netgo $(LDFLAGS) && tar -czf gor_$(VERSION)_PRO_x64.tar.gz gor && rm gor @@ -29,8 +29,7 @@ install: go install $(MAC_LDFLAGS) build: - docker build -t $(CONTAINER) . - + docker build -t $(CONTAINER) -f Dockerfile.dev . profile: go build && ./gor --output-http="http://localhost:9000" --input-dummy 0 --input-raw :9000 --input-http :9000 --memprofile=./mem.out --cpuprofile=./cpu.out --stats --output-http-stats --output-http-timeout 100ms @@ -72,7 +71,7 @@ run: $(RUN) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw-track-response --input-raw 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh" --output-file requests.gor run-2: - sudo -E go run $(SOURCE) --input-dummy="" --output-tcp localhost:27001 --verbose --debug + $(RUN) go run $(LDFLAGS) $(SOURCE) --input-raw :8000 --input-raw-bpf-filter "dst port 8000" --output-stdout --output-http "http://localhost:8000" --input-dummy=0 run-3: sudo -E go run $(SOURCE) --input-tcp :27001 --output-stdout diff --git a/byteutils/byteutils.go b/byteutils/byteutils.go index f1c1e15..f5c58e2 100644 --- a/byteutils/byteutils.go +++ b/byteutils/byteutils.go @@ -29,12 +29,15 @@ func Replace(a []byte, from, to int, new []byte) []byte { copy(a[from:from+len(new)], new) return a - } else if lenDiff < 0 { + } + + if lenDiff < 0 { copy(a[from:], new) copy(a[from+len(new):], a[to:]) return a[:len(a)+lenDiff] - } else { // same size - copy(a[from:], new) - return a } + + // same size + copy(a[from:], new) + return a } diff --git a/elasticsearch.go b/elasticsearch.go index 281f9cd..b46f277 100644 --- a/elasticsearch.go +++ b/elasticsearch.go @@ -1,18 +1,22 @@ package main import ( + "net/url" "encoding/json" - "github.com/buger/gor-pro/proto" - "github.com/mattbaird/elastigo/lib" "log" - "regexp" + "strings" + //"regexp" "time" + + "github.com/buger/gor-pro/proto" + + "github.com/mattbaird/elastigo/lib" ) type ESUriErorr struct{} func (e *ESUriErorr) Error() string { - return "Wrong ElasticSearch URL format. Expected to be: host:port/index_name" + return "Wrong ElasticSearch URL format. Expected to be: scheme://host/index_name" } type ESPlugin struct { @@ -52,17 +56,27 @@ type ESRequestResponse struct { // Parse ElasticSearch URI // -// Proper format is: host:port/index_name -func parseURI(URI string) (err error, host string, port string, index string) { - rURI := regexp.MustCompile("(.+):([0-9]+)/(.+)") - match := rURI.FindAllStringSubmatch(URI, -1) +// Proper format is: scheme://[userinfo@]host/index_name +// userinfo is: user[:password] +// net/url.Parse() does not fail if scheme is not provided but actualy does not +// handle URI properly. +// So we must 'validate' URI format to match requirements to use net/url.Parse() +func parseURI(URI string) (err error, index string) { - if len(match) == 0 { + parsedUrl, parseErr := url.Parse(URI) + + if parseErr != nil { + err = new(ESUriErorr) + } + + // check URL validity by extracting host and undex values. + host := parsedUrl.Host + urlPathParts := strings.Split(parsedUrl.Path, "/") + index = urlPathParts[len(urlPathParts) - 1 ] + + // force index specification in uri : ie no implicit index + if (host == "" || index == "") { err = new(ESUriErorr) - } else { - host = match[0][1] - port = match[0][2] - index = match[0][3] } return @@ -71,14 +85,15 @@ func parseURI(URI string) (err error, host string, port string, index string) { func (p *ESPlugin) Init(URI string) { var err error - err, p.Host, p.ApiPort, p.Index = parseURI(URI) + err, p.Index = parseURI(URI) if err != nil { log.Fatal("Can't initialize ElasticSearch plugin.", err) } + p.eConn = elastigo.NewConn() - p.eConn.SetPort(p.ApiPort) - p.eConn.SetHosts([]string{p.Host}) + + p.eConn.SetFromUrl(URI) p.indexor = p.eConn.NewBulkIndexerErrors(50, 60) p.done = make(chan bool) diff --git a/elasticsearch_test.go b/elasticsearch_test.go new file mode 100644 index 0000000..701c38f --- /dev/null +++ b/elasticsearch_test.go @@ -0,0 +1,140 @@ +package main + +import ( + "testing" +) + +const expectedIndex = "gor" + +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) { + if expectedIndex != index { + t.Fatalf("Expected index %s but got %s", expectedIndex, index) + } +} + +func assertExpectedError (returnedError error, t *testing.T) { + expectedError := new(ESUriErorr) + + if expectedError != returnedError { + t.Errorf("Expected err %s but got %s", expectedError, returnedError) + } +} + +func assertNoError (returnedError error, t *testing.T) { + if nil != returnedError { + t.Errorf("Expected err %s but got %s", nil, returnedError) + } +} + +// Argument host:port/index_name +// i.e : localhost:9200/gor +// Fail because scheme is mandatory +func TestElasticConnectionBuildFailWithoutScheme(t *testing.T) { + uri := "localhost:9200/" + expectedIndex + + err, _ := parseURI(uri) + assertExpectedError(err, t) +} + +// Argument scheme://host:port +// i.e : http://localhost:9200 +// Fail : explicit index is required +func TestElasticConnectionBuildFailWithoutIndex(t *testing.T) { + uri := "http://localhost:9200" + + err, index := parseURI(uri) + + assertExpectedIndex("", index, t) + + assertExpectedError(err, t) +} + +// Argument scheme://host/index_name +// i.e : http://localhost/gor +func TestElasticConnectionBuildFailWithoutPort(t *testing.T) { + uri := "http://localhost/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : http://localhost:9200/gor +func TestElasticLocalConnectionBuild(t *testing.T) { + uri := "http://localhost:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : http://localhost.local:9200/gor or https://localhost.local:9200/gor +func TestElasticSimpleLocalWithSchemeConnectionBuild(t *testing.T) { + uri := "http://localhost.local:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : http://localhost.local:9200/gor or https://localhost.local:9200/gor +func TestElasticSimpleLocalWithHTTPSConnectionBuild(t *testing.T) { + uri := "https://localhost.local:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/index_name +// i.e : localhost.local:9200/pathtoElastic/gor +func TestElasticLongPathConnectionBuild(t *testing.T) { + uri := "http://localhost.local:9200/pathtoElastic/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:userinfo@port/index_name +// i.e : http://user:password@localhost.local:9200/gor +func TestElasticBasicAuthConnectionBuild(t *testing.T) { + uri := "http://user:password@localhost.local:9200/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} + +// Argument scheme://host:port/path/index_name +// i.e : http://localhost.local:9200/path/gor or https://localhost.local:9200/path/gor +func TestElasticComplexPathConnectionBuild(t *testing.T) { + uri := "http://localhost.local:9200/path/" + expectedIndex + + err, index := parseURI(uri) + + assertNoError(err, t) + + assertExpectedGorIndex(index, t) +} diff --git a/examples/middleware/echo.py b/examples/middleware/echo.py new file mode 100644 index 0000000..f8b7c99 --- /dev/null +++ b/examples/middleware/echo.py @@ -0,0 +1,76 @@ +#! /usr/bin/env python3 +# -*- coding: utf-8 -*- + +import sys +import fileinput +import binascii + +# Used to find end of the Headers section +EMPTY_LINE = b'\r\n\r\n' + + +def log(msg): + """ + Logging to STDERR as STDOUT and STDIN used for data transfer + @type msg: str or byte string + @param msg: Message to log to STDERR + """ + try: + msg = str(msg) + '\n' + except: + pass + sys.stderr.write(msg) + sys.stderr.flush() + + +def find_end_of_headers(byte_data): + """ + Finds where the header portion ends and the content portion begins. + @type byte_data: str or byte string + @param byte_data: Hex decoded req or resp string + """ + return byte_data.index(EMPTY_LINE) + 4 + + +def process_stdin(): + """ + Process STDIN and output to STDOUT + """ + for raw_line in fileinput.input(): + + line = raw_line.rstrip() + + # Decode base64 encoded line + decoded = bytes.fromhex(line) + + # Split into metadata and payload, the payload is headers + body + (raw_metadata, payload) = decoded.split(b'\n', 1) + + # Split into headers and payload + headers_pos = find_end_of_headers(payload) + raw_headers = payload[:headers_pos] + raw_content = payload[headers_pos:] + + log('===================================') + request_type_id = int(raw_metadata.split(b' ')[0]) + log('Request type: {}'.format({ + 1: 'Request', + 2: 'Original Request', + 3: 'Replayed Response' + }[request_type_id])) + log('===================================') + + log('Original data:') + log(line) + + log('Decoded request:') + log(decoded) + + encoded = binascii.hexlify(raw_metadata + b'\n' + raw_headers + raw_content).decode('ascii') + log('Encoded data:') + log(encoded) + + sys.stdout.write(encoded + '\n') + +if __name__ == '__main__': + process_stdin() diff --git a/examples/middleware/token_modifier.go b/examples/middleware/token_modifier.go index b715d76..a0013cd 100644 --- a/examples/middleware/token_modifier.go +++ b/examples/middleware/token_modifier.go @@ -24,8 +24,9 @@ import ( "bytes" "encoding/hex" "fmt" - "github.com/buger/goreplay/proto" "os" + + "github.com/buger/goreplay/proto" ) // requestID -> originalToken diff --git a/http_client.go b/http_client.go index cdb0b5b..baa3953 100644 --- a/http_client.go +++ b/http_client.go @@ -122,10 +122,14 @@ func (c *HTTPClient) isAlive() bool { if err == nil { return true - } else if err == io.EOF { + } + + if err == io.EOF { Debug("[HTTPClient] connection closed, reconnecting") return false - } else if err == syscall.EPIPE { + } + + if err == syscall.EPIPE { Debug("Detected broken pipe.", err) return false } @@ -134,7 +138,7 @@ func (c *HTTPClient) isAlive() bool { } func (c *HTTPClient) Send(data []byte) (response []byte, err error) { - var payload []byte + var payload []byte // Don't exit on panic defer func() { @@ -143,7 +147,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { if _, ok := r.(error); ok { log.Println("[HTTPClient] Failed to send request: ", string(data)) - log.Println("[HTTPClient] Response: ", string(response)) + log.Println("[HTTPClient] Response: ", string(response)) log.Println("PANIC: pkg:", r, string(debug.Stack())) } } @@ -202,16 +206,16 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { } else { // If headers are finished - if bytes.Contains(c.respBuf[:readBytes], proto.EmptyLine) { - if bytes.Equal(proto.Header(c.respBuf[:readBytes], []byte("Transfer-Encoding")), []byte("chunked")) { + if bytes.Contains(c.respBuf[:readBytes], proto.EmptyLine) { + if bytes.Equal(proto.Header(c.respBuf[:readBytes], []byte("Transfer-Encoding")), []byte("chunked")) { chunked = true } else { - status, _ := strconv.Atoi(string(proto.Status(c.respBuf[:readBytes]))) + status, _ := strconv.Atoi(string(proto.Status(c.respBuf[:readBytes]))) if (status >= 100 && status < 200) || status == 204 || status == 304 { - contentLength = 0 - break + contentLength = 0 + break } else { - l := proto.Header(c.respBuf[:readBytes], []byte("Content-Length")) + l := proto.Header(c.respBuf[:readBytes], []byte("Content-Length")) if len(l) > 0 { contentLength, _ = strconv.Atoi(string(l)) } @@ -280,7 +284,6 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { break } - } if readBytes >= maxResponseSize { @@ -294,16 +297,16 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { } if err != nil && readBytes == 0 { - Debug("[HTTPClient] Response read timeout error", err, c.conn, readBytes, string(c.respBuf[:readBytes])) + Debug("[HTTPClient] Response read timeout error", err, c.conn, readBytes, string(c.respBuf[:readBytes])) response = errorPayload(HTTP_TIMEOUT) - c.Disconnect() + c.Disconnect() return } - if readBytes < 4 || string(c.respBuf[:4]) != "HTTP" { + if readBytes < 4 || string(c.respBuf[:4]) != "HTTP" { Debug("[HTTPClient] Response read unknown error", err, c.conn, readBytes, string(c.respBuf[:readBytes])) response = errorPayload(HTTP_UNKNOWN_ERROR) - c.Disconnect() + c.Disconnect() return } diff --git a/http_client_test.go b/http_client_test.go index 6d743ed..04283ab 100644 --- a/http_client_test.go +++ b/http_client_test.go @@ -3,7 +3,6 @@ package main import ( "bytes" "crypto/rand" - "github.com/buger/goreplay/proto" "io/ioutil" _ "log" "net" @@ -15,6 +14,8 @@ import ( "sync" "testing" "time" + + "github.com/buger/goreplay/proto" ) func TestHTTPClientURLPort(t *testing.T) { diff --git a/http_modifier.go b/http_modifier.go index cdbd7e4..60c4501 100644 --- a/http_modifier.go +++ b/http_modifier.go @@ -2,7 +2,9 @@ package main import ( "bytes" + "strings" "hash/fnv" + "encoding/base64" "github.com/buger/gor-pro/proto" ) @@ -19,6 +21,7 @@ func NewHTTPModifier(config *HTTPModifierConfig) *HTTPModifier { len(config.headerRewrite) == 0 && len(config.headerFilters) == 0 && len(config.headerNegativeFilters) == 0 && + len(config.headerBasicAuthFilters) == 0 && len(config.headerHashFilters) == 0 && len(config.paramHashFilters) == 0 && len(config.params) == 0 && @@ -111,6 +114,23 @@ func (m *HTTPModifier) Rewrite(payload []byte) (response []byte) { } } + if len(m.config.headerBasicAuthFilters) > 0 { + for _, f := range m.config.headerBasicAuthFilters { + value := proto.Header(payload, []byte("Authorization")) + + if len(value) > 0 { + valueString := string(value) + trimmedBasicAuthEncoded := strings.TrimPrefix(valueString, "Basic ") + if strings.Compare(valueString, trimmedBasicAuthEncoded) != 0 { + decodedAuth, _ := base64.StdEncoding.DecodeString(trimmedBasicAuthEncoded) + if !f.regexp.Match(decodedAuth) { + return + } + } + } + } + } + if len(m.config.headerHashFilters) > 0 { for _, f := range m.config.headerHashFilters { value := proto.Header(payload, f.name) diff --git a/http_modifier_settings.go b/http_modifier_settings.go index ba0f0e0..5ae70e6 100644 --- a/http_modifier_settings.go +++ b/http_modifier_settings.go @@ -10,14 +10,15 @@ import ( // HTTPModifierConfig holds configuration options for built-in traffic modifier type HTTPModifierConfig struct { - urlNegativeRegexp HTTPUrlRegexp - urlRegexp HTTPUrlRegexp - urlRewrite UrlRewriteMap - headerRewrite HeaderRewriteMap - headerFilters HTTPHeaderFilters - headerNegativeFilters HTTPHeaderFilters - headerHashFilters HTTPHashFilters - paramHashFilters HTTPHashFilters + urlNegativeRegexp HTTPUrlRegexp + urlRegexp HTTPUrlRegexp + urlRewrite UrlRewriteMap + headerRewrite HeaderRewriteMap + headerFilters HTTPHeaderFilters + headerNegativeFilters HTTPHeaderFilters + headerBasicAuthFilters HTTPHeaderBasicAuthFilters + headerHashFilters HTTPHashFilters + paramHashFilters HTTPHashFilters params HTTPParams headers HTTPHeaders @@ -55,6 +56,32 @@ func (h *HTTPHeaderFilters) Set(value string) error { return nil } +// +// Handling of --http-basic-auth-filter option +// +type basicAuthFilter struct { + regexp *regexp.Regexp +} + +// HTTPHeaderFilters holds list of headers and their regexps +type HTTPHeaderBasicAuthFilters []basicAuthFilter + +func (h *HTTPHeaderBasicAuthFilters) String() string { + return fmt.Sprint(*h) +} + +func (h *HTTPHeaderBasicAuthFilters) Set(value string) error { + r, err := regexp.Compile(value) + if err != nil { + return err + } + + *h = append(*h, basicAuthFilter{regexp: r}) + + return nil +} + + // // Handling of --http-allow-header-hash and --http-allow-param-hash options // @@ -233,7 +260,6 @@ func (r *HeaderRewriteMap) Set(value string) error { return nil } - // // Handling of --http-allow-url option // diff --git a/http_modifier_test.go b/http_modifier_test.go index 6e4ee48..e39e0af 100644 --- a/http_modifier_test.go +++ b/http_modifier_test.go @@ -2,8 +2,9 @@ package main import ( "bytes" - "github.com/buger/goreplay/proto" "testing" + + "github.com/buger/goreplay/proto" ) func TestHTTPModifierWithoutConfig(t *testing.T) { @@ -78,6 +79,45 @@ func TestHTTPModifierHeaderNegativeFilters(t *testing.T) { } } +func TestHTTPHeaderBasicAuthFilters(t *testing.T) { + filters := HTTPHeaderBasicAuthFilters{} + filters.Set("^customer[0-9].*") + + modifier := NewHTTPModifier(&HTTPModifierConfig{ + headerBasicAuthFilters: filters, + }) + + //Encoded UserId:Password = customer3:welcome + payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nAuthorization: Basic Y3VzdG9tZXIzOndlbGNvbWU=\r\n\r\na=1&b=2") + if len(modifier.Rewrite(payload)) == 0 { + t.Error("Request should pass filters") + } + + //customer6:rest@123^TEST + 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 + filters.Set("^(homer simpson|mickey mouse).*") + + modifier = NewHTTPModifier(&HTTPModifierConfig{ + headerBasicAuthFilters: filters, + }) + + if len(modifier.Rewrite(payload)) != 0 { + t.Error("Request should not pass filters") + } + + //mickey mouse:happy123 + 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) { var url, newURL []byte diff --git a/http_prettifier.go b/http_prettifier.go index b28a4ab..61ae23e 100644 --- a/http_prettifier.go +++ b/http_prettifier.go @@ -1,69 +1,70 @@ package main import ( - "github.com/buger/gor-pro/proto" - "bytes" - "compress/gzip" - "strconv" - "io/ioutil" - "net/http/httputil" + "bytes" + "compress/gzip" + "io/ioutil" + "net/http/httputil" + "strconv" + + "github.com/buger/gor-pro/proto" ) func prettifyHTTP(p []byte) []byte { - headSize := bytes.IndexByte(p, '\n') + 1 - head := p[:headSize] - body := p[headSize:] + headSize := bytes.IndexByte(p, '\n') + 1 + head := p[:headSize] + body := p[headSize:] - headersPos := proto.MIMEHeadersEndPos(body) - headers := body[:headersPos] - content := body[headersPos:] + headersPos := proto.MIMEHeadersEndPos(body) + headers := body[:headersPos] + content := body[headersPos:] - var tEnc, cEnc []byte - proto.ParseHeaders([][]byte{headers}, func(header, value []byte) bool { - if proto.HeadersEqual(header, []byte("Transfer-Encoding")) { - tEnc = value - } + var tEnc, cEnc []byte + proto.ParseHeaders([][]byte{headers}, func(header, value []byte) bool { + if proto.HeadersEqual(header, []byte("Transfer-Encoding")) { + tEnc = value + } - if proto.HeadersEqual(header, []byte("Content-Encoding")) { - cEnc = value - } + if proto.HeadersEqual(header, []byte("Content-Encoding")) { + cEnc = value + } - return true - }) + return true + }) - if len(tEnc) == 0 && len(cEnc) == 0 { - return p - } + if len(tEnc) == 0 && len(cEnc) == 0 { + return p + } - if bytes.Equal(tEnc, []byte("chunked")) { - buf := bytes.NewBuffer(content) - r := httputil.NewChunkedReader(buf) - content, _ = ioutil.ReadAll(r) + if bytes.Equal(tEnc, []byte("chunked")) { + buf := bytes.NewBuffer(content) + r := httputil.NewChunkedReader(buf) + content, _ = ioutil.ReadAll(r) - headers = proto.DeleteHeader(headers, []byte("Transfer-Encoding")) + headers = proto.DeleteHeader(headers, []byte("Transfer-Encoding")) - newLen := strconv.Itoa(len(content)) - headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen)) - } + newLen := strconv.Itoa(len(content)) + headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen)) + } - if bytes.Equal(cEnc, []byte("gzip")) { - buf := bytes.NewBuffer(content) - g, err := gzip.NewReader(buf) + if bytes.Equal(cEnc, []byte("gzip")) { + buf := bytes.NewBuffer(content) + g, err := gzip.NewReader(buf) - if err != nil { - Debug("[Prettifier] GZIP encoding error:", err) - return []byte{} - } + if err != nil { + Debug("[Prettifier] GZIP encoding error:", err) + return []byte{} + } - content, _ = ioutil.ReadAll(g) + content, _ = ioutil.ReadAll(g) - headers = proto.DeleteHeader(headers, []byte("Content-Encoding")) + headers = proto.DeleteHeader(headers, []byte("Content-Encoding")) - newLen := strconv.Itoa(len(content)) - headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen)) - } + newLen := strconv.Itoa(len(content)) + headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen)) + } - newPayload := append(append(head, headers...), content...) + newPayload := append(append(head, headers...), content...) - return newPayload -} \ No newline at end of file + return newPayload +} diff --git a/http_prettifier_test.go b/http_prettifier_test.go index 233492d..c48c97f 100644 --- a/http_prettifier_test.go +++ b/http_prettifier_test.go @@ -1,36 +1,36 @@ package main import ( - "compress/gzip" - "testing" - "strconv" - "bytes" + "bytes" + "compress/gzip" + "strconv" + "testing" ) func TestHTTPPrettifierGzip(t *testing.T) { - b := bytes.NewBufferString("") - w := gzip.NewWriter(b) - w.Write([]byte("test")) - w.Close() + b := bytes.NewBufferString("") + w := gzip.NewWriter(b) + w.Write([]byte("test")) + w.Close() - size := strconv.Itoa(len(b.Bytes())) + size := strconv.Itoa(len(b.Bytes())) - payload := []byte("2 1 1\nHTTP/1.1 200 OK\r\nContent-Length: " + size + "\r\nContent-Encoding: gzip\r\n\r\n") - payload = append(payload, b.Bytes()...) + payload := []byte("2 1 1\nHTTP/1.1 200 OK\r\nContent-Length: " + size + "\r\nContent-Encoding: gzip\r\n\r\n") + payload = append(payload, b.Bytes()...) - newPayload := prettifyHTTP(payload) + newPayload := prettifyHTTP(payload) - if string(newPayload) != "2 1 1\nHTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\ntest" { - t.Error("Payload not match:", string(newPayload)) - } + if string(newPayload) != "2 1 1\nHTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\ntest" { + t.Error("Payload not match:", string(newPayload)) + } } func TestHTTPPrettifierChunked(t *testing.T) { - payload := []byte("POST / HTTP/1.1\r\nHost: www.w3.org\r\nTransfer-Encoding: chunked\r\n\r\n4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n") + payload := []byte("POST / HTTP/1.1\r\nHost: www.w3.org\r\nTransfer-Encoding: chunked\r\n\r\n4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n") - newPayload := prettifyHTTP(payload) + newPayload := prettifyHTTP(payload) - if string(newPayload) != "POST / HTTP/1.1\r\nHost: www.w3.org\r\nContent-Length: 23\r\n\r\nWikipedia in\r\n\r\nchunks." { - t.Error("Payload not match:", string(newPayload)) - } -} \ No newline at end of file + if string(newPayload) != "POST / HTTP/1.1\r\nHost: www.w3.org\r\nContent-Length: 23\r\n\r\nWikipedia in\r\n\r\nchunks." { + t.Error("Payload not match:", string(newPayload)) + } +} diff --git a/input_http_test.go b/input_http_test.go index 5f6dd4c..0d09fd1 100644 --- a/input_http_test.go +++ b/input_http_test.go @@ -1,7 +1,6 @@ package main import ( - "github.com/buger/goreplay/proto" "io" "log" "net/http" @@ -9,6 +8,8 @@ import ( "strings" "sync" "testing" + + "github.com/buger/goreplay/proto" ) func TestHTTPInput(t *testing.T) { diff --git a/input_kafka.go b/input_kafka.go index 25e6d72..e03fdbb 100644 --- a/input_kafka.go +++ b/input_kafka.go @@ -2,9 +2,10 @@ package main import ( "encoding/json" + "log" + "github.com/Shopify/sarama" "github.com/Shopify/sarama/mocks" - "log" ) // KafkaInput is used for recieving Kafka messages and @@ -82,20 +83,21 @@ func (i *KafkaInput) Read(data []byte) (int, error) { if !i.config.useJSON { copy(data, message.Value) return len(message.Value), nil - } else { - var kafkaMessage KafkaMessage - json.Unmarshal(message.Value, &kafkaMessage) - - buf, err := kafkaMessage.Dump() - if err != nil { - log.Println("Failed to decode access log entry:", err) - return 0, err - } - - copy(data, buf) - - return len(buf), nil } + + var kafkaMessage KafkaMessage + json.Unmarshal(message.Value, &kafkaMessage) + + buf, err := kafkaMessage.Dump() + if err != nil { + log.Println("Failed to decode access log entry:", err) + return 0, err + } + + copy(data, buf) + + return len(buf), nil + } func (i *KafkaInput) String() string { diff --git a/input_kafka_test.go b/input_kafka_test.go index f987ccd..9c0c7f1 100644 --- a/input_kafka_test.go +++ b/input_kafka_test.go @@ -1,9 +1,10 @@ package main import ( + "testing" + "github.com/Shopify/sarama" "github.com/Shopify/sarama/mocks" - "testing" ) func TestInputKafkaRAW(t *testing.T) { diff --git a/input_raw.go b/input_raw.go index f733670..703b604 100644 --- a/input_raw.go +++ b/input_raw.go @@ -1,11 +1,12 @@ package main import ( - "github.com/buger/gor-pro/proto" - raw "github.com/buger/gor-pro/raw_socket_listener" "log" "net" "time" + + "github.com/buger/gor-pro/proto" + raw "github.com/buger/goreplay/raw_socket_listener" ) // RAWInput used for intercepting traffic for given address @@ -19,6 +20,7 @@ type RAWInput struct { trackResponse bool listener *raw.Listener protocol raw.TCPProtocol + bpfFilter string } // Available engines for intercepting traffic @@ -29,12 +31,13 @@ const ( ) // NewRAWInput constructor for RAWInput. Accepts address with port as argument. -func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration, realIPHeader string, protocol string) (i *RAWInput) { +func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration, realIPHeader string, protocol string, bpfFilter string) (i *RAWInput) { i = new(RAWInput) i.data = make(chan *raw.TCPMessage) i.address = address i.expire = expire i.engine = engine + i.bpfFilter = bpfFilter i.realIPHeader = []byte(realIPHeader) i.quit = make(chan bool) i.trackResponse = trackResponse @@ -80,17 +83,11 @@ func (i *RAWInput) listen(address string) { host, port, err := net.SplitHostPort(address) - if i.engine == EnginePcapFile { - host = address - port = "1" - err = nil - } - if err != nil { log.Fatal("input-raw: error while parsing address", err) } - i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.protocol) + i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.protocol, i.bpfFilter) ch := i.listener.Receiver() diff --git a/input_raw_test.go b/input_raw_test.go index 1c6ac30..2fc8082 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -2,7 +2,6 @@ package main import ( "bytes" - "github.com/buger/goreplay/proto" "io" "io/ioutil" "log" @@ -19,6 +18,8 @@ import ( "sync/atomic" "testing" "time" + + "github.com/buger/goreplay/proto" ) const testRawExpire = time.Millisecond * 200 @@ -43,7 +44,7 @@ func TestRAWInputIPv4(t *testing.T) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "X-Real-IP", "http") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "X-Real-IP", "http", "") defer input.Close() output := NewTestOutput(func(data []byte) { @@ -105,7 +106,11 @@ func TestRAWInputNoKeepAlive(t *testing.T) { originAddr := listener.Addr().String() +<<<<<<< HEAD input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "http") +======= + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() output := NewTestOutput(func(data []byte) { @@ -151,7 +156,11 @@ func TestRAWInputIPv6(t *testing.T) { var respCounter, reqCounter int64 +<<<<<<< HEAD input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "http") +======= + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() output := NewTestOutput(func(data []byte) { @@ -202,7 +211,11 @@ func TestInputRAW100Expect(t *testing.T) { originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) +<<<<<<< HEAD input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "http") +======= + input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() // We will use it to get content of raw HTTP request @@ -265,7 +278,11 @@ func TestInputRAWChunkedEncoding(t *testing.T) { })) originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) +<<<<<<< HEAD input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "http") +======= + input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -329,7 +346,11 @@ func TestInputRAWLargePayload(t *testing.T) { })) originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) +<<<<<<< HEAD input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "http") +======= + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { @@ -374,7 +395,11 @@ func BenchmarkRAWInput(b *testing.B) { var respCounter, reqCounter int64 +<<<<<<< HEAD input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "http") +======= + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() output := NewTestOutput(func(data []byte) { diff --git a/input_tcp.go b/input_tcp.go index 025312a..8230bad 100644 --- a/input_tcp.go +++ b/input_tcp.go @@ -3,8 +3,8 @@ package main import ( "bufio" "bytes" - "fmt" "crypto/tls" + "fmt" "io" "log" "net" @@ -46,17 +46,17 @@ func (i *TCPInput) Read(data []byte) (int, error) { func (i *TCPInput) listen(address string) { if i.config.secure { - cer, err := tls.LoadX509KeyPair(i.config.certificatePath, i.config.keyPath) - if err != nil { - log.Fatal("Error while loading --input-file certificate:", err) - } + cer, err := tls.LoadX509KeyPair(i.config.certificatePath, i.config.keyPath) + if err != nil { + log.Fatal("Error while loading --input-file certificate:", err) + } - config := &tls.Config{Certificates: []tls.Certificate{cer}} - listener, err := tls.Listen("tcp", address, config) - if err != nil { - log.Fatal("Can't start --input-tcp with secure connection:", err) - } - i.listener = listener + config := &tls.Config{Certificates: []tls.Certificate{cer}} + listener, err := tls.Listen("tcp", address, config) + if err != nil { + log.Fatal("Can't start --input-tcp with secure connection:", err) + } + i.listener = listener } else { listener, err := net.Listen("tcp", address) if err != nil { diff --git a/input_tcp_test.go b/input_tcp_test.go index ef62b6b..e5b5ba9 100644 --- a/input_tcp_test.go +++ b/input_tcp_test.go @@ -1,28 +1,28 @@ package main import ( - "io" - "os" - "log" - "net" - "io/ioutil" - "crypto/x509" - "crypto/rsa" - "crypto/rand" - "crypto/tls" - "encoding/pem" - "math/big" - "time" "bytes" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "encoding/pem" + "io" + "io/ioutil" + "log" + "math/big" + "net" + "os" "sync" "testing" + "time" ) func TestTCPInput(t *testing.T) { wg := new(sync.WaitGroup) quit := make(chan int) - input := NewTCPInput("127.0.0.1:0", &TCPInputConfig{}) + input := NewTCPInput("127.0.0.1:0", &TCPInputConfig{}) output := NewTestOutput(func(data []byte) { wg.Done() }) @@ -90,7 +90,7 @@ func TestTCPInputSecure(t *testing.T) { serverPrivPemFile.Write(serverPrivPem) serverPrivPemFile.Close() - defer func(){ + defer func() { os.Remove(serverPrivPemFile.Name()) os.Remove(serverCertPemFile.Name()) }() @@ -99,9 +99,9 @@ func TestTCPInputSecure(t *testing.T) { quit := make(chan int) input := NewTCPInput("127.0.0.1:0", &TCPInputConfig{ - secure: true, + secure: true, certificatePath: serverCertPemFile.Name(), - keyPath: serverPrivPemFile.Name(), + keyPath: serverPrivPemFile.Name(), }) output := NewTestOutput(func(data []byte) { wg.Done() @@ -112,15 +112,15 @@ func TestTCPInputSecure(t *testing.T) { go Start(quit) - conf := &tls.Config{ - InsecureSkipVerify: true, - } + conf := &tls.Config{ + InsecureSkipVerify: true, + } - conn, err := tls.Dial("tcp", input.listener.Addr().String(), conf) - if err != nil { - t.Fatal(err) - } - defer conn.Close() + conn, err := tls.Dial("tcp", input.listener.Addr().String(), conf) + if err != nil { + t.Fatal(err) + } + defer conn.Close() msg := []byte("1 1 1\nGET / HTTP/1.1\r\n\r\n") @@ -133,4 +133,4 @@ func TestTCPInputSecure(t *testing.T) { wg.Wait() close(quit) -} \ No newline at end of file +} diff --git a/kafka.go b/kafka.go index 893a732..7a6be1d 100644 --- a/kafka.go +++ b/kafka.go @@ -3,8 +3,9 @@ package main import ( "bytes" "fmt" - "github.com/Shopify/sarama" + "github.com/buger/gor-pro/proto" + "github.com/Shopify/sarama" ) // KafkaConfig should contains required information to diff --git a/middleware.go b/middleware.go index 311dddf..83ad8d1 100644 --- a/middleware.go +++ b/middleware.go @@ -34,7 +34,7 @@ func NewMiddleware(command string) *Middleware { m.Stdout, _ = cmd.StdoutPipe() m.Stdin, _ = cmd.StdinPipe() - cmd.Stderr = os.Stderr + cmd.Stderr = os.Stderr go m.read(m.Stdout) @@ -67,11 +67,11 @@ func (m *Middleware) copy(to io.Writer, from io.Reader) { for { nr, _ := from.Read(buf) if nr > 0 && len(buf) > nr { - payload := buf[0: nr] + payload := buf[0:nr] if Settings.prettifyHTTP { payload = prettifyHTTP(payload) - nr = len(payload) + nr = len(payload) } hex.Encode(dst, payload) diff --git a/middleware_test.go b/middleware_test.go index 1043dd0..5cfe342 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -4,7 +4,6 @@ import ( "bytes" "crypto/rand" "encoding/hex" - "github.com/buger/goreplay/proto" "io" "net/http" "net/http/httptest" @@ -12,6 +11,8 @@ import ( "sync" "testing" "time" + + "github.com/buger/goreplay/proto" ) type fakeServiceCb func(string, int, []byte) @@ -117,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, "", "http") + input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "http", "") defer input.Close() // And redirect to another @@ -179,7 +180,11 @@ func TestTokenMiddleware(t *testing.T) { fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1) // Catch traffic from one service +<<<<<<< HEAD input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "http") +======= + input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer input.Close() // And redirect to another diff --git a/output_file.go b/output_file.go index a565568..26bbc0d 100644 --- a/output_file.go +++ b/output_file.go @@ -191,9 +191,10 @@ func (o *FileOutput) Write(data []byte) (n int, err error) { meta := payloadMeta(data) o.currentID = meta[1] o.payloadType = meta[0] - o.updateName() } - + + o.updateName() + if o.file == nil || o.currentName != o.file.Name() { o.mu.Lock() o.Close() diff --git a/output_file_settings.go b/output_file_settings.go index e7dac3c..58cf656 100644 --- a/output_file_settings.go +++ b/output_file_settings.go @@ -20,11 +20,12 @@ func parseDataUnit(s string) int64 { if unit, ok := dataUnitMap[s[len(s)-1]]; ok { size, _ := strconv.ParseInt(s[:len(s)-1], 10, 64) return unit * size - } else { - // If no unit specified use bytes - size, _ := strconv.ParseInt(s, 10, 64) - return size } + + // If no unit specified use bytes + size, _ := strconv.ParseInt(s, 10, 64) + return size + } type unitSizeVar int64 @@ -37,4 +38,3 @@ func (u *unitSizeVar) Set(s string) error { *u = unitSizeVar(parseDataUnit(s)) return nil } - diff --git a/output_file_test.go b/output_file_test.go index 9a81102..3dfa564 100644 --- a/output_file_test.go +++ b/output_file_test.go @@ -321,7 +321,7 @@ func TestFileOutputAppendSizeLimitOverflow(t *testing.T) { messageSize := len(message) + len(payloadSeparator) - output := NewFileOutput(name, &FileOutputConfig{append: false, flushInterval: time.Minute, sizeLimit: unitSizeVar(2 * messageSize) }) + output := NewFileOutput(name, &FileOutputConfig{append: false, flushInterval: time.Minute, sizeLimit: unitSizeVar(2 * messageSize)}) output.Write([]byte("1 1 1\r\ntest")) name1 := output.file.Name() @@ -345,4 +345,4 @@ func TestFileOutputAppendSizeLimitOverflow(t *testing.T) { os.Remove(name1) os.Remove(name3) -} \ No newline at end of file +} diff --git a/output_http.go b/output_http.go index 1bec930..074b85a 100644 --- a/output_http.go +++ b/output_http.go @@ -246,9 +246,9 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) { func (o *HTTPOutput) Read(data []byte) (int, error) { resp := <-o.responses - if Settings.debug { - Debug("[OUTPUT-HTTP] Received response:", string(resp.payload)) - } + if Settings.debug { + Debug("[OUTPUT-HTTP] Received response:", string(resp.payload)) + } header := payloadHeader(ReplayedResponsePayload, resp.uuid, resp.roundTripTime, resp.startedAt) copy(data[0:len(header)], header) @@ -260,9 +260,9 @@ func (o *HTTPOutput) Read(data []byte) (int, error) { func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) { meta := payloadMeta(request) - if Settings.debug { - Debug(meta) - } + if Settings.debug { + Debug(meta) + } if len(meta) < 2 { return diff --git a/output_kafka.go b/output_kafka.go index cfce972..a82f036 100644 --- a/output_kafka.go +++ b/output_kafka.go @@ -2,13 +2,15 @@ package main import ( "encoding/json" - "github.com/Shopify/sarama" - "github.com/Shopify/sarama/mocks" - "github.com/buger/gor-pro/proto" "io" "log" "strings" "time" + + "github.com/buger/gor-pro/proto" + + "github.com/Shopify/sarama" + "github.com/Shopify/sarama/mocks" ) // KafkaOutput is used for sending payloads to kafka in JSON format. diff --git a/output_kafka_test.go b/output_kafka_test.go index cf9efe0..30a58e0 100644 --- a/output_kafka_test.go +++ b/output_kafka_test.go @@ -1,9 +1,10 @@ package main import ( + "testing" + "github.com/Shopify/sarama" "github.com/Shopify/sarama/mocks" - "testing" ) func TestOutputKafkaRAW(t *testing.T) { diff --git a/output_tcp.go b/output_tcp.go index 69ba05e..4c5b9ef 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -21,7 +21,7 @@ type TCPOutput struct { } type TCPOutputConfig struct { - secure bool + secure bool } // NewTCPOutput constructor for TCPOutput diff --git a/plugins.go b/plugins.go index d66e679..95cded3 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.inputRAWProtocol) + registerPlugin(NewRAWInput, options, engine, Settings.inputRAWTrackResponse, Settings.inputRAWExpire, Settings.inputRAWRealIPHeader, Settings.inputRAWProtocol, Settings.inputRAWBpfFilter) } for _, options := range Settings.inputTCP { diff --git a/proto/proto.go b/proto/proto.go index 8ca4cbb..9672050 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -18,7 +18,8 @@ package proto import ( "bytes" - "github.com/buger/gor-pro/byteutils" + + "github.com/buger/gor-pro/byteutils" ) // In HTTP newline defined by 2 bytes (for both windows and *nix support) @@ -348,18 +349,18 @@ func Path(payload []byte) []byte { if eol > 0 { if end == -1 || eol < end { - return payload[start : start + eol] + return payload[start : start+eol] } } else { // support for legacy clients eol = bytes.IndexByte(payload[start:], '\n') - if eol > 0 && (end == - 1 || eol < end) { - return payload[start : start + eol] + if eol > 0 && (end == -1 || eol < end) { + return payload[start : start+eol] } } if end < 0 { - return payload[start: len(payload)] + return payload[start:len(payload)] } return payload[start : start+end] diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 2629c2f..c3fbff2 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -16,10 +16,6 @@ import ( "bytes" "encoding/binary" "fmt" - "github.com/buger/gor-pro/proto" - "github.com/google/gopacket" - "github.com/google/gopacket/layers" - "github.com/google/gopacket/pcap" "io" "log" "net" @@ -29,6 +25,12 @@ import ( "strings" "sync" "time" + + "github.com/buger/gor-pro/proto" + + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + "github.com/google/gopacket/pcap" ) var _ = fmt.Println @@ -70,6 +72,8 @@ type Listener struct { trackResponse bool messageExpire time.Duration + bpfFilter string + conn net.PacketConn pcapHandles []*pcap.Handle @@ -93,7 +97,11 @@ const ( ) // NewListener creates and initializes new Listener object +<<<<<<< HEAD func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, protocol TCPProtocol) (l *Listener) { +======= +func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, bpfFilter string) (l *Listener) { +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 l = &Listener{} l.packetsChan = make(chan *packet, 10000) @@ -107,7 +115,11 @@ func NewListener(addr string, port string, engine int, trackResponse bool, expir l.respAliases = make(map[uint32]*TCPMessage) l.respWithoutReq = make(map[uint32]tcpID) l.trackResponse = trackResponse +<<<<<<< HEAD l.protocol = protocol +======= + l.bpfFilter = bpfFilter +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 l.addr = addr _port, _ := strconv.Atoi(port) @@ -306,9 +318,9 @@ func findPcapDevices(addr string) (interfaces []pcap.Interface, err error) { if len(interfaces) == 0 { return nil, &DeviceNotFoundError{addr} - } else { - return interfaces, nil } + + return interfaces, nil } func (t *Listener) readPcap() { @@ -371,6 +383,10 @@ func (t *Listener) readPcap() { bpf = "tcp dst port " + strconv.Itoa(int(t.port)) + " and (" + bpfDstHost + ")" } + if t.bpfFilter != "" { + bpf = t.bpfFilter + } + if err := handle.SetBPFFilter(bpf); err != nil { log.Println("BPF filter error:", err, "Device:", device.Name, bpf) wg.Done() @@ -545,6 +561,13 @@ func (t *Listener) readPcapFile() { if handle, err := pcap.OpenOffline(t.addr); err != nil { log.Fatal(err) } else { + if t.bpfFilter != "" { + if err := handle.SetBPFFilter(t.bpfFilter); err != nil { + log.Println("BPF filter error:", err) + return + } + } + t.readyCh <- true packetSource := gopacket.NewPacketSource(handle, handle.LinkType()) @@ -563,12 +586,12 @@ func (t *Listener) readPcapFile() { tcp, _ := tcpLayer.(*layers.TCP) data = append(tcp.LayerContents(), tcp.LayerPayload()...) - if tcp.SrcPort >= 32768 && tcp.SrcPort <= 61000 { - copy(data[0:2], []byte{0, 0}) - copy(data[2:4], []byte{0, 1}) + if uint16(tcp.DstPort) == t.port { + copy(data[0:2], []byte{byte(tcp.SrcPort >> 8), byte(tcp.SrcPort)}) + copy(data[2:4], []byte{byte(tcp.DstPort >> 8), byte(tcp.DstPort)}) } else { - copy(data[0:2], []byte{0, 1}) - copy(data[2:4], []byte{0, 0}) + copy(data[0:2], []byte{byte(tcp.DstPort >> 8), byte(tcp.DstPort)}) + copy(data[2:4], []byte{byte(tcp.SrcPort >> 8), byte(tcp.SrcPort)}) } } else { continue diff --git a/raw_socket_listener/listener_test.go b/raw_socket_listener/listener_test.go index ef11333..73f56d6 100644 --- a/raw_socket_listener/listener_test.go +++ b/raw_socket_listener/listener_test.go @@ -12,7 +12,7 @@ import ( func TestRawListenerInput(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "") defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -49,7 +49,7 @@ func TestRawListenerInput(t *testing.T) { func TestRawListenerInputResponseByClose(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -89,7 +89,11 @@ func TestRawListenerInputResponseByClose(t *testing.T) { func TestRawListenerInputWithoutResponse(t *testing.T) { var req *TCPMessage +<<<<<<< HEAD listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond, ProtocolHTTP) +======= + listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -111,7 +115,11 @@ func TestRawListenerInputWithoutResponse(t *testing.T) { func TestRawListenerResponse(t *testing.T) { var req, resp *TCPMessage +<<<<<<< HEAD listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP) +======= + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -149,7 +157,11 @@ func TestRawListenerResponse(t *testing.T) { } func TestShort100Continue(t *testing.T) { +<<<<<<< HEAD listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP) +======= + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 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"), time.Now()) @@ -169,7 +181,11 @@ func TestShort100Continue(t *testing.T) { // Response comes before Request func Test100ContinueWrongOrder(t *testing.T) { +<<<<<<< HEAD listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP) +======= + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 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"), time.Now()) @@ -188,7 +204,11 @@ func Test100ContinueWrongOrder(t *testing.T) { } func TestAlt100ContinueHeaderOrder(t *testing.T) { +<<<<<<< HEAD listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP) +======= + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer listener.Close() reqPacket1 := buildPacket(true, 1, 1, []byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 2\r\n\r\n"), time.Now()) @@ -349,7 +369,11 @@ func permutation(n int, list []*TCPPacket) []*TCPPacket { // Response comes before Request func TestRawListenerChunkedWrongOrder(t *testing.T) { +<<<<<<< HEAD listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP) +======= + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 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"), time.Now()) @@ -427,7 +451,11 @@ func getMessage() []*TCPPacket { // Response comes before Request func TestRawListenerBench(t *testing.T) { +<<<<<<< HEAD l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond, ProtocolHTTP) +======= + l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond, "") +>>>>>>> d309650589a56db2bfa93c533337e3db1cfb0425 defer l.Close() // Should re-construct message from all possible combinations diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index ff1193b..1e38e68 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -5,11 +5,12 @@ import ( "crypto/sha1" "encoding/binary" "encoding/hex" - "github.com/buger/gor-pro/proto" "log" "net" "strconv" "time" + + "github.com/buger/gor-pro/proto" ) var _ = log.Println @@ -246,31 +247,26 @@ func (t *TCPMessage) checkIfComplete() { return } - // If one GET, OPTIONS, or HEAD request - if t.methodType == httpMethodWithoutBody { + switch t.bodyType { + case httpBodyEmpty: t.complete = true - } else { - switch t.bodyType { - case httpBodyEmpty: + case httpBodyContentLength: + if t.contentLength == 0 || t.contentLength == t.BodySize() { t.complete = true - case httpBodyContentLength: - if t.contentLength == 0 || t.contentLength == t.BodySize() { - t.complete = true - } - case httpBodyChunked: - lastPacket := t.packets[len(t.packets)-1] - if bytes.LastIndex(lastPacket.Data, bChunkEnd) != -1 { - t.complete = true - } - default: - if len(t.packets) == 0 { - return - } + } + case httpBodyChunked: + lastPacket := t.packets[len(t.packets)-1] + if bytes.LastIndex(lastPacket.Data, bChunkEnd) != -1 { + t.complete = true + } + default: + if len(t.packets) == 0 { + return + } - last := t.packets[len(t.packets)-1] - if last.IsFIN { - t.complete = true - } + last := t.packets[len(t.packets)-1] + if last.IsFIN { + t.complete = true } } } @@ -278,19 +274,11 @@ func (t *TCPMessage) checkIfComplete() { type httpMethodType uint8 const ( - httpMethodNotSet httpMethodType = 0 - httpMethodWithBody httpMethodType = 1 - httpMethodWithoutBody httpMethodType = 2 - httpMethodNotFound httpMethodType = 3 + httpMethodNotSet httpMethodType = 0 + httpMethodKnown httpMethodType = 1 + httpMethodNotFound httpMethodType = 2 ) -var methodsWithBody = [][]byte{ - []byte("POST"), - []byte("PUT"), - []byte("PATCH"), - []byte("CONNECT"), -} - func (t *TCPMessage) updateMethodType() { // if there is cache if t.methodType != httpMethodNotSet && t.methodType != httpMethodNotFound { @@ -307,11 +295,7 @@ func (t *TCPMessage) updateMethodType() { } if t.IsIncoming { - var method []byte - if mIdx := bytes.IndexByte(d[:8], ' '); mIdx != -1 { - method = d[:mIdx] - // Check that after method we have absolute or relative path switch d[mIdx+1] { case '/', 'h', '*': @@ -324,21 +308,14 @@ func (t *TCPMessage) updateMethodType() { return } - for _, m := range methodsWithBody { - if len(m) == len(method) && bytes.Equal(m, method) { - t.methodType = httpMethodWithBody - return - } - } - - t.methodType = httpMethodWithoutBody + t.methodType = httpMethodKnown } else { if !bytes.Equal(d[:6], []byte("HTTP/1")) { t.methodType = httpMethodNotFound return } - t.methodType = httpMethodWithBody + t.methodType = httpMethodKnown } } @@ -386,10 +363,7 @@ func (t *TCPMessage) updateBodyType() { switch t.methodType { case httpMethodNotFound: return - case httpMethodWithoutBody: - t.bodyType = httpBodyEmpty - return - case httpMethodWithBody: + case httpMethodKnown: if len(lengthB) > 0 { t.contentLength, _ = strconv.Atoi(string(lengthB)) @@ -431,10 +405,6 @@ func (t *TCPMessage) check100Continue() { return } - if t.methodType != httpMethodWithBody { - return - } - if t.seqMissing || t.headerPacket == -1 { return } diff --git a/raw_socket_listener/tcp_message_test.go b/raw_socket_listener/tcp_message_test.go index e05f058..91f2b7b 100644 --- a/raw_socket_listener/tcp_message_test.go +++ b/raw_socket_listener/tcp_message_test.go @@ -170,16 +170,16 @@ func TestTCPMessageMethodType(t *testing.T) { payload string expectedMethodType httpMethodType }{ - {true, "GET / HTTP/1.1\r\n\r\n", httpMethodWithoutBody}, - {true, "GET * HTTP/1.1\r\n\r\n", httpMethodWithoutBody}, - {true, "UNKNOWN / HTTP/1.1\r\n\r\n", httpMethodWithoutBody}, - {true, "GET http://example.com HTTP/1.1\r\n\r\n", httpMethodWithoutBody}, - {true, "POST / HTTP/1.1\r\n\r\n", httpMethodWithBody}, - {true, "PUT / HTTP/1.1\r\n\r\n", httpMethodWithBody}, + {true, "GET / HTTP/1.1\r\n\r\n", httpMethodKnown}, + {true, "GET * HTTP/1.1\r\n\r\n", httpMethodKnown}, + {true, "UNKNOWN / HTTP/1.1\r\n\r\n", httpMethodKnown}, + {true, "GET http://example.com HTTP/1.1\r\n\r\n", httpMethodKnown}, + {true, "POST / HTTP/1.1\r\n\r\n", httpMethodKnown}, + {true, "PUT / HTTP/1.1\r\n\r\n", httpMethodKnown}, {true, "GET zxc HTTP/1.1\r\n\r\n", httpMethodNotFound}, {true, "GET / HTTP\r\n\r\n", httpMethodNotFound}, {true, "VERYLONGMETHOD / HTTP/1.1\r\n\r\n", httpMethodNotFound}, - {false, "HTTP/1.1 200 OK\r\n\r\n", httpMethodWithBody}, + {false, "HTTP/1.1 200 OK\r\n\r\n", httpMethodKnown}, {false, "HTTP /1.1 200 OK\r\n\r\n", httpMethodNotFound}, } @@ -199,6 +199,7 @@ func TestTCPMessageBodyType(t *testing.T) { expectedBodyType httpBodyType }{ {true, "GET / HTTP/1.1\r\n\r\n", httpBodyEmpty}, + {true, "GET / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab", httpBodyContentLength}, {true, "POST / HTTP/1.1\r\n\r\n", httpBodyEmpty}, {true, "POST / HTTP/1.1\r\nUser-Agent: zxc\r\n\r\n", httpBodyEmpty}, {false, "HTTP/1.1 200 OK\r\n\r\n", httpBodyEmpty}, diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index 19deda4..3649a89 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/raw_socket_listener/tcp_packet.go @@ -78,8 +78,8 @@ func (t *TCPPacket) ParseBasic() { t.IsFIN = t.Raw[13]&0x01 != 0 if len(t.Raw) >= int(t.DataOffset*4) { - t.Data = t.Raw[t.DataOffset*4:] - } + t.Data = t.Raw[t.DataOffset*4:] + } } func (t *TCPPacket) dump() *packet { diff --git a/settings.go b/settings.go index 57e5813..fab7307 100644 --- a/settings.go +++ b/settings.go @@ -55,6 +55,7 @@ type AppSettings struct { inputRAWRealIPHeader string inputRAWProtocol string inputRAWExpire time.Duration + inputRAWBpfFilter string middleware string @@ -106,7 +107,6 @@ func init() { flag.StringVar(&Settings.inputTCPConfig.certificatePath, "input-tcp-certificate", "", "Path to PEM encoded certificate file. Used when TLS turned on.") flag.StringVar(&Settings.inputTCPConfig.keyPath, "input-tcp-certificate-key", "", "Path to PEM encoded certificate key file. Used when TLS turned on.") - flag.Var(&Settings.outputTCP, "output-tcp", "Used for internal communication between Gor instances. Example: \n\t# Listen for requests on 80 port and forward them to other Gor instance on 28020 port\n\tgor --input-raw :80 --output-tcp replay.local:28020") flag.BoolVar(&Settings.outputTCPConfig.secure, "output-tcp-secure", false, "Use TLS secure connection. --input-file on another end should have TLS turned on as well.") flag.BoolVar(&Settings.outputTCPStats, "output-tcp-stats", false, "Report TCP output queue stats to console every 5 seconds.") @@ -139,6 +139,8 @@ func init() { flag.DurationVar(&Settings.inputRAWExpire, "input-raw-expire", time.Second*2, "How much it should wait for the last TCP packet, till consider that TCP message complete.") + 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.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") @@ -199,7 +201,10 @@ 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.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-imiter user-id:25%") + 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, "output-http-header-hash-filter", "WARNING: `output-http-header-hash-filter` DEPRECATED, use `--http-header-hash-limiter` instead") flag.Var(&Settings.modifierConfig.paramHashFilters, "http-param-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific GET param:\n\t gor --input-raw :8080 --output-http staging.com --http-param-limiter user_id:25%")