diff --git a/Dockerfile b/Dockerfile index 3f03461..bbb0dab 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,8 @@ -FROM alpine:3.6 -RUN apk update && apk add ca-certificates && update-ca-certificates && apk add openssl +FROM alpine:latest as builder +RUN apk add --no-cache ca-certificates openssl RUN wget https://github.com/buger/goreplay/releases/download/v0.16.1/gor_0.16.1_x64.tar.gz -O gor.tar.gz RUN tar xzf gor.tar.gz -ENTRYPOINT ["./goreplay"] \ No newline at end of file + +FROM scratch +COPY --from=builder /goreplay . +ENTRYPOINT ["./goreplay"] diff --git a/Dockerfile.dev b/Dockerfile.dev index b47f638..46b159c 100644 --- a/Dockerfile.dev +++ b/Dockerfile.dev @@ -13,7 +13,7 @@ RUN apt-get install oracle-java8-installer -y RUN apt-get install flex bison -y RUN wget http://www.tcpdump.org/release/libpcap-1.8.1.tar.gz && tar xzf libpcap-1.8.1.tar.gz && cd libpcap-1.8.1 && ./configure && make install RUN go get github.com/google/gopacket -RUN go get -u github.com/golang/lint/golint +RUN go get -u golang.org/x/lint/golint WORKDIR /go/src/github.com/buger/gor-pro/ ADD . /go/src/github.com/buger/gor-pro/ diff --git a/emitter.go b/emitter.go index 2b92553..a201433 100644 --- a/emitter.go +++ b/emitter.go @@ -4,6 +4,7 @@ import ( "bytes" "hash/fnv" "io" + "log" "time" ) @@ -23,15 +24,30 @@ func Start(stop chan int) { } } - go CopyMulty(middleware, Plugins.Outputs...) + go func() { + if err := CopyMulty(middleware, Plugins.Outputs...); err != nil { + log.Println("Error during copy: ", err) + close(stop) + } + }() } else { for _, in := range Plugins.Inputs { - go CopyMulty(in, Plugins.Outputs...) + go func() { + if err := CopyMulty(in, Plugins.Outputs...); err != nil { + log.Println("Error during copy: ", err) + close(stop) + } + }() } for _, out := range Plugins.Outputs { if r, ok := out.(io.Reader); ok { - go CopyMulty(r, Plugins.Outputs...) + go func() { + if err := CopyMulty(r, Plugins.Outputs...); err != nil { + log.Println("Error during copy: ", err) + close(stop) + } + }() } } } @@ -48,7 +64,7 @@ func Start(stop chan int) { // CopyMulty copies from 1 reader to multiple writers func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { - buf := make([]byte, 5*1024*1024) + buf := make([]byte, Settings.copyBufferSize) wIndex := 0 modifier := NewHTTPModifier(&Settings.modifierConfig) filteredRequests := make(map[string]time.Time) @@ -58,11 +74,18 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { for { nr, er := src.Read(buf) + + if er == io.EOF { + return nil + } + if er != nil { + return err + } + _maxN := nr if nr > 500 { _maxN = 500 } - if nr > 0 && len(buf) > nr { payload := buf[:nr] meta := payloadMeta(payload) @@ -74,6 +97,10 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { } requestID := string(meta[1]) + if nr >= 5*1024*1024 { + log.Println("INFO: Large packet... We received ", len(payload), " bytes from ", src) + } + if Settings.debug { Debug("[EMITTER] input:", string(payload[0:_maxN]), nr, "from:", src) } @@ -123,8 +150,10 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { wIndex = int(hasher.Sum32()) % len(writers) writers[wIndex].Write(payload) } else { - // Simple round robin - writers[wIndex].Write(payload) + // Simple round robin + if _, err := writers[wIndex].Write(payload); err != nil { + return err + } wIndex++ @@ -134,17 +163,13 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { } } else { for _, dst := range writers { - dst.Write(payload) + if _, err := dst.Write(payload); err != nil { + return err + } } } - - } - if er == io.EOF { - break - } - if er != nil { - err = er - break + } else if nr > 0 { + log.Println("WARN: Packet", nr, "bytes is too large to process. Consider increasing --copy-buffer-size") } // Run GC on each 1000 request diff --git a/gor.go b/gor.go index f035d4c..0126d21 100644 --- a/gor.go +++ b/gor.go @@ -9,6 +9,7 @@ import ( "log" "net/http" "net/http/httputil" + _ "net/http/pprof" "os" "os/signal" "runtime" @@ -77,6 +78,12 @@ func main() { profileCPU(*cpuprofile) } + if Settings.pprof != "" { + go func() { + log.Println(http.ListenAndServe(Settings.pprof, nil)) + }() + } + c := make(chan os.Signal, 1) signal.Notify(c, os.Interrupt, syscall.SIGTERM) go func() { @@ -87,7 +94,6 @@ func main() { if Settings.exitAfter > 0 { log.Println("Running gor for a duration of", Settings.exitAfter) - closeCh = make(chan int) time.AfterFunc(Settings.exitAfter, func() { log.Println("Stopping gor after", Settings.exitAfter) diff --git a/gor_stat.go b/gor_stat.go index 2273a5c..8f28e6f 100644 --- a/gor_stat.go +++ b/gor_stat.go @@ -7,21 +7,19 @@ import ( "time" ) -const ( - rate = 5 -) - type GorStat struct { statName string + rateMs int latest int mean int max int count int } -func NewGorStat(statName string) (s *GorStat) { +func NewGorStat(statName string, rateMs int) (s *GorStat) { s = new(GorStat) s.statName = statName + s.rateMs = rateMs s.latest = 0 s.mean = 0 s.max = 0 @@ -55,13 +53,13 @@ func (s *GorStat) Reset() { } func (s *GorStat) String() string { - return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count/rate) + "," + strconv.Itoa(runtime.NumGoroutine()) + return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count/(s.rateMs/1000.0)) + "," + strconv.Itoa(runtime.NumGoroutine()) } func (s *GorStat) reportStats() { for { log.Println(s) s.Reset() - time.Sleep(rate * time.Second) + time.Sleep(time.Duration(s.rateMs) * time.Millisecond) } } diff --git a/http_client.go b/http_client.go index 0f76308..2826ad3 100644 --- a/http_client.go +++ b/http_client.go @@ -1,12 +1,14 @@ package main import ( + "bufio" "bytes" "crypto/tls" "encoding/base64" "io" "log" "net" + "net/http" "net/url" "runtime/debug" "strconv" @@ -47,6 +49,8 @@ type HTTPClient struct { host string auth string conn net.Conn + proxy *url.URL + proxyAuth string respBuf []byte config *HTTPClientConfig redirectsCount int @@ -80,19 +84,74 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient { client.auth = "Basic " + base64.StdEncoding.EncodeToString([]byte(u.User.String())) } + client.proxy, _ = http.ProxyFromEnvironment(&http.Request{URL: u}) + + if client.isProxy() && client.proxy.User != nil { + client.proxyAuth = "Basic " + base64.StdEncoding.EncodeToString([]byte(client.proxy.User.String())) + } + return client } func (c *HTTPClient) Connect() (err error) { c.Disconnect() + var toDial string if !strings.Contains(c.host, ":") { - c.conn, err = net.DialTimeout("tcp", c.host+":"+defaultPorts[c.scheme], c.config.ConnectionTimeout) + toDial = c.host + ":" + defaultPorts[c.scheme] } else { - c.conn, err = net.DialTimeout("tcp", c.host, c.config.ConnectionTimeout) + toDial = c.host + } + + if c.isProxy() { + if c.proxy.Scheme != "http" { + panic("Unsupported HTTP Proxy method") + } + Debug("[HTTPClient] Connecting to proxy", c.proxy.String(), "<>", toDial) + c.conn, err = net.DialTimeout("tcp", c.proxy.Host, c.config.ConnectionTimeout) + if err != nil { + return + } + if c.scheme == "https" { + c.conn.Write([]byte("CONNECT " + toDial + " HTTP/1.1\r\n")) + if c.proxyAuth != "" { + c.conn.Write([]byte("Proxy-Authorization: " + c.proxyAuth + "\r\n")) + } + c.conn.Write([]byte("\r\n")) + br := bufio.NewReader(c.conn) + l, _, err := br.ReadLine() + if err != nil { + return err + } + if len(l) < 12 { + panic("HTTP proxy did not respond correctly") + } + status := l[9:12] + if !bytes.Equal(status, []byte("200")) { + panic("HTTP proxy did not respond correctly") + } + for { + // Read until we find the empty line + l, _, err := br.ReadLine() + if err != nil { + return err + } + if len(l) == 0 { + break + } + } + } + Debug("[HTTPClient] Proxy successfully connected") + } else { + c.conn, err = net.DialTimeout("tcp", toDial, c.config.ConnectionTimeout) + if err != nil { + return + } } if c.scheme == "https" { + // Wrap our socket in TLS + Debug("[HTTPClient] Wrapping socket in TLS", c.host) tlsConn := tls.Client(c.conn, &tls.Config{InsecureSkipVerify: true, ServerName: c.host}) if err = tlsConn.Handshake(); err != nil { @@ -100,6 +159,7 @@ func (c *HTTPClient) Connect() (err error) { } c.conn = tlsConn + Debug("[HTTPClient] Successfully wrapped in TLS") } return @@ -135,8 +195,6 @@ func (c *HTTPClient) isAlive(readBytes *int) bool { } func (c *HTTPClient) Send(data []byte) (response []byte, err error) { - var payload []byte - // Don't exit on panic defer func() { if r := recover(); r != nil { @@ -150,7 +208,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { } }() - var readBytes, n int + var readBytes int if c.conn == nil || !c.isAlive(&readBytes) { Debug("[HTTPClient] Connecting:", c.baseURL) if err = c.Connect(); err != nil { @@ -168,6 +226,16 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { data = proto.SetHost(data, []byte(c.baseURL), []byte(c.host)) } + if c.isProxy() && c.scheme == "http" { + path := proto.Path(data) + if len(path) > 0 && path[0] == '/' { + data = proto.SetPath(data, c.proxyPath(path)) + if c.proxyAuth != "" { + data = proto.SetHeader(data, []byte("Proxy-Authorization"), []byte(c.proxyAuth)) + } + } + } + if c.auth != "" { data = proto.SetHeader(data, []byte("Authorization"), []byte(c.auth)) } @@ -176,6 +244,12 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { Debug("[HTTPClient] Sending:", string(data)) } + return c.send(data, readBytes, timeout) +} + +func (c *HTTPClient) send(data []byte, readBytes int, timeout time.Time) (response []byte, err error) { + var payload []byte + var n int if _, err = c.conn.Write(data); err != nil { Debug("[HTTPClient] Write error:", err, c.baseURL) response = errorPayload(HTTP_TIMEOUT) @@ -376,6 +450,14 @@ func (c *HTTPClient) Post(path string, body []byte) (response []byte, err error) return c.Send([]byte(payload)) } +func (c *HTTPClient) proxyPath(path []byte) []byte { + return append([]byte(c.scheme+"://"+c.host), path...) +} + +func (c *HTTPClient) isProxy() bool { + return c.proxy != nil +} + const ( // https://support.cloudflare.com/hc/en-us/articles/200171936-Error-520-Web-server-is-returning-an-unknown-error HTTP_UNKNOWN_ERROR = "520" diff --git a/http_client_test.go b/http_client_test.go index 963c816..1af3ab8 100644 --- a/http_client_test.go +++ b/http_client_test.go @@ -15,7 +15,7 @@ import ( "testing" "time" - "github.com/buger/goreplay/proto" + "github.com/buger/gor-pro/proto" ) func TestHTTPClientURLPort(t *testing.T) { diff --git a/http_modifier.go b/http_modifier.go index e406a80..e18a68f 100644 --- a/http_modifier.go +++ b/http_modifier.go @@ -99,10 +99,10 @@ func (m *HTTPModifier) Rewrite(payload []byte) (response []byte) { value := proto.Header(payload, f.name) if len(value) == 0 { - return - } + return + } - if !f.regexp.Match(value) { + if !f.regexp.Match(value) { return } } diff --git a/http_modifier_test.go b/http_modifier_test.go index ac7baba..f35511a 100644 --- a/http_modifier_test.go +++ b/http_modifier_test.go @@ -4,7 +4,7 @@ import ( "bytes" "testing" - "github.com/buger/goreplay/proto" + "github.com/buger/gor-pro/proto" ) func TestHTTPModifierWithoutConfig(t *testing.T) { diff --git a/input_http_test.go b/input_http_test.go index 0d09fd1..5eb8adb 100644 --- a/input_http_test.go +++ b/input_http_test.go @@ -9,7 +9,7 @@ import ( "sync" "testing" - "github.com/buger/goreplay/proto" + "github.com/buger/gor-pro/proto" ) func TestHTTPInput(t *testing.T) { diff --git a/input_raw.go b/input_raw.go index c18b83c..7b98e8b 100644 --- a/input_raw.go +++ b/input_raw.go @@ -91,7 +91,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.protocol, i.bpfFilter, i.timestampType, i.bufferSize) + i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.protocol, i.bpfFilter, i.timestampType, i.bufferSize, Settings.inputRAWOverrideSnapLen, Settings.inputRAWImmediateMode) ch := i.listener.Receiver() diff --git a/input_raw_test.go b/input_raw_test.go index f295901..6f8018f 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -19,7 +19,7 @@ import ( "testing" "time" - "github.com/buger/goreplay/proto" + "github.com/buger/gor-pro/proto" ) const testRawExpire = time.Millisecond * 200 diff --git a/middleware/middleware.js b/middleware/middleware.js index c869362..5c7b4ac 100755 --- a/middleware/middleware.js +++ b/middleware/middleware.js @@ -45,7 +45,7 @@ function init() { if (proxy.ch[chanID]) { proxy.ch[chanID].forEach(function(ch){ let r = ch.cb(msg); - if (r) resp = r; // If one of callback decided not to send response back, do not override it in global callbacks + if (resp) resp = r; // If one of callback decided not to send response back, do not override it in global callbacks }) // Cleanup Individual message channels to avoid memory leaks @@ -58,6 +58,8 @@ function init() { if (resp) { process.stdout.write(`${resp.rawMeta.toString('hex')}${Buffer.from("\n").toString("hex")}${resp.http.toString('hex')}\n`) } + + return resp } } @@ -421,7 +423,7 @@ module.exports = { // =========== Tests ============== function testRunner(){ - ["init", "parseMessage", "httpMethod", "httpPath", "setHttpHeader", "deleteHttpHeader", "httpPathParam", "httpHeader", "httpBody", "setHttpBody", "httpBodyParam", "httpCookie", "setHttpCookie", "httpHeaders"].forEach(function(t){ + ["init", "filter", "parseMessage", "httpMethod", "httpPath", "setHttpHeader", "deleteHttpHeader", "httpPathParam", "httpHeader", "httpBody", "setHttpBody", "httpBodyParam", "httpCookie", "setHttpCookie", "httpHeaders"].forEach(function(t){ console.log(`====== Start ${t} =======`) eval(`TEST_${t}()`) console.log(`====== End ${t} =======`) @@ -495,6 +497,7 @@ function TEST_init() { let req = parseMessage(Buffer.from("1 2 3\nGET / HTTP/1.1\r\n\r\n").toString('hex')); let resp = parseMessage(Buffer.from("2 2 3\nHTTP/1.1 200 OK\r\n\r\n").toString('hex')); let resp2 = parseMessage(Buffer.from("2 3 3\nHTTP/1.1 200 OK\r\n\r\n").toString('hex')); + gor.emit(req); gor.emit(resp); gor.emit(resp2); @@ -506,6 +509,34 @@ function TEST_init() { } } +function TEST_filter() { + const child_process = require('child_process'); + + let gor = init(); + gor.on("request", function(req){ + if (httpPath(req.http) != "/filter") { + return req + } + }); + + gor.on("request", function(req){ + return req + }); + + + let reqPass = parseMessage(Buffer.from("1 2 3\nGET / HTTP/1.1\r\n\r\n").toString('hex')); + let reqFilter = parseMessage(Buffer.from("1 2 3\nGET /filter HTTP/1.1\r\n\r\n").toString('hex')); + + if (!gor.emit(reqPass)) { + return fail("Should not filter request") + } + + if (gor.emit(reqFilter)) { + return fail("Should filter request even if one middleware rejected it") + } + +} + function TEST_parseMessage() { const exampleMessage = Buffer.from("1 2 3\nGET / HTTP/1.1\r\n\r\n").toString('hex') let msg = parseMessage(exampleMessage) diff --git a/middleware/package.json b/middleware/package.json index 9748460..c7ca3bf 100644 --- a/middleware/package.json +++ b/middleware/package.json @@ -1,6 +1,6 @@ { "name": "goreplay_middleware", - "version": "0.1.15", + "version": "0.1.19", "description": "Package for writing middleware for GoReplay https://goreplay.org", "main": "middleware.js", "scripts": { diff --git a/middleware_test.go b/middleware_test.go index 931eca2..0119d39 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -12,7 +12,7 @@ import ( "testing" "time" - "github.com/buger/goreplay/proto" + "github.com/buger/gor-pro/proto" ) type fakeServiceCb func(string, int, []byte) diff --git a/output_file.go b/output_file.go index 395524c..3feac48 100644 --- a/output_file.go +++ b/output_file.go @@ -3,6 +3,7 @@ package main import ( "bufio" "compress/gzip" + "errors" "fmt" "io" "log" @@ -31,6 +32,7 @@ var dateFileNameFuncs = map[string]func(*FileOutput) string{ type FileOutputConfig struct { flushInterval time.Duration sizeLimit unitSizeVar + outputFileMaxSize unitSizeVar queueLimit int append bool bufferPath string @@ -50,6 +52,7 @@ type FileOutput struct { currentID []byte payloadType []byte closed bool + totalFileSize int64 config *FileOutputConfig } @@ -219,8 +222,13 @@ func (o *FileOutput) Write(data []byte) (n int, err error) { o.writer.Write(data) o.writer.Write([]byte(payloadSeparator)) + o.totalFileSize += int64(len(data) + len(payloadSeparator)) o.queueLength++ + if Settings.outputFileConfig.outputFileMaxSize > 0 && o.totalFileSize >= int64(Settings.outputFileConfig.outputFileMaxSize) { + return len(data), errors.New("File output reached size limit") + } + return len(data), nil } @@ -244,7 +252,9 @@ func (o *FileOutput) flush() { if stat, err := o.file.Stat(); err == nil { o.chunkSize = int(stat.Size()) - } + } else { + log.Println("Error accessing file sats", err) + } } } diff --git a/output_http.go b/output_http.go index 074b85a..0ab4fe2 100644 --- a/output_http.go +++ b/output_http.go @@ -3,6 +3,7 @@ package main import ( "fmt" "io" + "log" "sync/atomic" "time" @@ -63,8 +64,12 @@ type response struct { type HTTPOutputConfig struct { redirectLimit int - stats bool + stats bool + workersMin int + workersMax int + statsMs int workers int + queueLen int elasticSearch string @@ -112,18 +117,18 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o.config = config if o.config.stats { - o.queueStats = NewGorStat("output_http") + o.queueStats = NewGorStat("output_http", o.config.statsMs) } - o.queue = make(chan []byte, 1000) - o.responses = make(chan response, 1000) + o.queue = make(chan []byte, o.config.queueLen) + o.responses = make(chan response, o.config.queueLen) o.needWorker = make(chan int, 1) // Initial workers count - if o.config.workers == 0 { + if o.config.workersMax == 0 { o.needWorker <- initialDynamicWorkers } else { - o.needWorker <- o.config.workers + o.needWorker <- o.config.workersMax } if o.config.elasticSearch != "" { @@ -148,11 +153,6 @@ func (o *HTTPOutput) workerMaster() { for i := 0; i < newWorkers; i++ { go o.startWorker() } - - // Disable dynamic scaling if workers poll fixed size - if o.config.workers != 0 { - return - } } } @@ -203,17 +203,17 @@ func (o *HTTPOutput) startWorker() { o.sendRequest(client, data) case <-time.After(2 * time.Second): // When dynamic scaling enabled workers die after 2s of inactivity - if o.config.workers > 0 { + if o.config.workersMin == o.config.workersMax { continue } - workersCount := atomic.LoadInt64(&o.activeWorkers) + workersCount := int(atomic.LoadInt64(&o.activeWorkers)) - // At least 1 startWorker should be alive - if workersCount != 1 { - atomic.AddInt64(&o.activeWorkers, -1) - return - } + // At least 1 startWorker should be alive + if workersCount != 1 && workersCount > o.config.workersMin { + atomic.AddInt64(&o.activeWorkers, -1) + return + } } } } @@ -232,11 +232,18 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) { o.queueStats.Write(len(o.queue)) } - if !Settings.recognizeTCPSessions && o.config.workers == 0 { - workersCount := atomic.LoadInt64(&o.activeWorkers) + if !Settings.recognizeTCPSessions && o.config.workersMax != o.config.workersMin { + workersCount := int(atomic.LoadInt64(&o.activeWorkers)) - if len(o.queue) > int(workersCount) { - o.needWorker <- len(o.queue) + if len(o.queue) > workersCount { + extraWorkersReq := len(o.queue) - workersCount + 1 + maxWorkersAvailable := o.config.workersMax - workersCount + if extraWorkersReq > maxWorkersAvailable { + extraWorkersReq = maxWorkersAvailable + } + if extraWorkersReq > 0 { + o.needWorker <- extraWorkersReq + } } } @@ -279,6 +286,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) { stop := time.Now() if err != nil { + log.Println("Error when sending ", err, time.Now()) Debug("Request error:", err) } diff --git a/output_tcp.go b/output_tcp.go index 916ba7d..23b55ba 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -34,7 +34,7 @@ func NewTCPOutput(address string, config *TCPOutputConfig) io.Writer { o.buf = make(chan []byte, 1000) if Settings.outputTCPStats { - o.bufStats = NewGorStat("output_tcp") + o.bufStats = NewGorStat("output_tcp", 5000) } for i := 0; i < 10; i++ { diff --git a/proto/proto.go b/proto/proto.go index a6d148a..4887456 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -42,87 +42,23 @@ func MIMEHeadersStartPos(payload []byte) int { return bytes.Index(payload, CLRF) + 2 // Find first line end } -func isLower(b byte) bool { - if 'a' <= b && b <= 'z' { - return true - } - - return false -} - -func toUpper(b byte) byte { - if 'a' <= b && b <= 'z' { - b -= 'a' - 'A' - } - return b -} - -func toLower(b byte) byte { - if 'A' <= b && b <= 'Z' { - b += 'a' - 'A' - } - return b -} - func headerIndex(payload []byte, name []byte) int { - isLower := isLower(name[0]) i := 0 - for { - if i >= len(payload) { + // we need enough space for at least '\n' and the header name + if i >= (len(payload) - len(name) - 1) { return -1 } if payload[i] == '\n' { i++ - - // We are at the end - if i == len(payload) { - return -1 - } - - if payload[i] == name[0] || - (!isLower && payload[i] == toLower(name[0])) || - (isLower && payload[i] == toUpper(name[0])) { - - i++ - j := 1 - for { - if j == len(name) { - // Matched, and return start of the header - return i - len(name) - } - - // We are at the end - if i == len(payload) { - return -1 - } - - if payload[i] != name[j] { - break - } - - // If compound header name do one more case check: Content-Length or Transfer-Encoding - if name[j] == '-' { - i++ - j++ - - if !(payload[i] == name[j] || - (!isLower && payload[i] == toLower(name[j])) || - (isLower && payload[i] == toUpper(name[j]))) { - break - } - } - - j++ - i++ - } + if bytes.EqualFold(name, payload[i:i+len(name)]) { + return i } } - i++ - } + } return -1 } diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 36d6425..bd85f56 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -74,6 +74,8 @@ type Listener struct { bpfFilter string timestampType string + overrideSnapLen bool + immediateMode bool bufferSize int @@ -99,7 +101,8 @@ const ( EnginePcapFile ) -func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, protocol TCPProtocol, bpfFilter string, timestampType string, bufferSize int) (l *Listener) { +// NewListener creates and initializes new Listener object +func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, protocol TCPProtocol, bpfFilter string, timestampType string, bufferSize int, overrideSnapLen bool, immediateMode bool) (l *Listener) { l = &Listener{} l.packetsChan = make(chan *packet, 10000) @@ -116,7 +119,9 @@ func NewListener(addr string, port string, engine int, trackResponse bool, expir l.protocol = protocol l.bpfFilter = bpfFilter l.timestampType = timestampType + l.immediateMode = immediateMode l.bufferSize = bufferSize + l.overrideSnapLen = overrideSnapLen l.addr = addr _port, _ := strconv.Atoi(port) @@ -351,7 +356,7 @@ func (t *Listener) readPcap() { } } - if it, err := net.InterfaceByName(device.Name); err == nil { + if it, err := net.InterfaceByName(device.Name); err == nil && !t.overrideSnapLen { // Auto-guess max length of packet to capture inactive.SetSnapLen(it.MTU + 68*2) } else { @@ -360,7 +365,10 @@ func (t *Listener) readPcap() { inactive.SetTimeout(t.messageExpire) inactive.SetPromisc(true) - + inactive.SetImmediateMode(t.immediateMode) + if t.immediateMode { + log.Println("Setting immediate mode") + } if t.bufferSize > 0 { inactive.SetBufferSize(t.bufferSize) } diff --git a/raw_socket_listener/listener_test.go b/raw_socket_listener/listener_test.go index fc736a2..cff6375 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, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -77,7 +77,7 @@ func responsePacket(prev *TCPPacket, payload []byte) *TCPPacket { } func TestHEADRequestNoBody(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket := firstPacket([]byte("HEAD / HTTP/1.1\r\nContent-Length: 0\r\n\r\n")) @@ -111,7 +111,7 @@ func TestHEADRequestNoBody(t *testing.T) { } func TestSingleAck100Continue(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 4\r\n\r\n")) @@ -130,7 +130,7 @@ func TestSingleAck100Continue(t *testing.T) { } func Test100ContinueWithoutWaiting(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() req1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 4\r\n\r\n")) @@ -146,7 +146,7 @@ func Test100ContinueWithoutWaiting(t *testing.T) { // Client first sends data without waiting 100-continue, but once response received, generate packets based on Ack payload func Test100ContinueMixed(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() req1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 12\r\n\r\n")) @@ -164,7 +164,7 @@ func Test100ContinueMixed(t *testing.T) { } func TestDoubleAck100Continue(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 4\r\n\r\n")) @@ -187,7 +187,7 @@ func TestDoubleAck100Continue(t *testing.T) { func TestRawListenerInputResponseByClose(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -227,7 +227,7 @@ func TestRawListenerInputResponseByClose(t *testing.T) { func TestRawListenerInputWithoutResponse(t *testing.T) { var req *TCPMessage - listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -249,7 +249,7 @@ func TestRawListenerInputWithoutResponse(t *testing.T) { func TestRawListenerResponse(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket := firstPacket([]byte("GET / HTTP/1.1\r\n\r\n")) @@ -297,7 +297,7 @@ func get100ContinuePackets() (req []*TCPPacket, resp []*TCPPacket) { } func TestShort100Continue(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() req, resp := get100ContinuePackets() @@ -309,7 +309,7 @@ func TestShort100Continue(t *testing.T) { // Response comes before Request func Test100ContinueWrongOrder(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() req, resp := get100ContinuePackets() @@ -462,7 +462,7 @@ func permutation(n int, list []*TCPPacket) []*TCPPacket { // Response comes before Request func TestRawListenerChunkedWrongOrder(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket1 := firstPacket([]byte("POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nExpect: 100-continue\r\n\r\n")) @@ -532,7 +532,7 @@ func getMessage() []*TCPPacket { // Response comes before Request func TestRawListenerBench(t *testing.T) { - l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond, ProtocolHTTP, "", "", 0) + l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer l.Close() // Should re-construct message from all possible combinations @@ -583,7 +583,7 @@ func TestRawListenerBench(t *testing.T) { func TestResponseZeroContentLength(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0) + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, ProtocolHTTP, "", "", 0, false) defer listener.Close() reqPacket := firstPacket([]byte("POST /api/setup/install HTTP/1.1\r\nHost: localhost:22936\r\nUser-Agent: curl/7.57.0\r\nAccept: */*\r\nContent-Length: 0\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n")) @@ -620,4 +620,4 @@ func TestResponseZeroContentLength(t *testing.T) { if !bytes.Equal(resp.UUID(), req.UUID()) { t.Error("Resp and Req UUID should be equal") } -} \ No newline at end of file +} diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 446ca08..22f636e 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -207,7 +207,7 @@ func (t *TCPMessage) checkSeqIntegrity() { var bEmptyLine = []byte("\r\n\r\n") var bBR = []byte("\r\n") -var bChunkEnd = []byte("0\r\n\r\n") +var bChunkEnd = []byte("\r\n0\r\n\r\n") func (t *TCPMessage) updateHeadersPacket() { if len(t.packets) == 1 { @@ -228,8 +228,9 @@ func (t *TCPMessage) updateHeadersPacket() { t.headerPacket = i return } - } else if bytes.Equal(p.Data, bBR) { - if bytes.LastIndex(t.packets[i-1].Data, bBR) != -1 { + } else if i > 0 && bytes.Equal(p.Data, bBR) { + idx := bytes.LastIndex(t.packets[i-1].Data, bBR) + if idx != -1 && idx == len(t.packets[i-1].Data)-len(bBR) { t.headerPacket = i return } diff --git a/settings.go b/settings.go index a682ba6..831c138 100644 --- a/settings.go +++ b/settings.go @@ -33,6 +33,7 @@ type AppSettings struct { splitOutput bool recognizeTCPSessions bool + pprof string inputDummy MultiOption outputDummy MultiOption @@ -58,7 +59,10 @@ type AppSettings struct { inputRAWExpire time.Duration inputRAWBpfFilter string inputRAWTimestampType string + copyBufferSize int + inputRAWImmediateMode bool inputRawBufferSize int + inputRAWOverrideSnapLen bool middleware string @@ -89,6 +93,7 @@ func usage() { func init() { flag.Usage = usage + flag.StringVar(&Settings.pprof, "http-pprof", "", "Enable profiling. Starts http server on specified port, exposing special /debug/pprof endpoint. Example: `:8181`") flag.BoolVar(&Settings.verbose, "verbose", false, "Turn on more verbose output") flag.BoolVar(&Settings.debug, "debug", false, "Turn on debug output, shows all intercepted traffic. Works only when with `verbose` flag") flag.BoolVar(&Settings.stats, "stats", false, "Turn on queue stats output") @@ -130,6 +135,8 @@ func init() { Settings.outputFileConfig.sizeLimit.Set("32mb") flag.Var(&Settings.outputFileConfig.sizeLimit, "output-file-size-limit", "Size of each chunk. Default: 32mb") flag.IntVar(&Settings.outputFileConfig.queueLimit, "output-file-queue-limit", 256, "The length of the chunk queue. Default: 256") + Settings.outputFileConfig.outputFileMaxSize.Set("-1") + flag.Var(&Settings.outputFileConfig.outputFileMaxSize, "output-file-max-size-limit", "Max size of output file, Default: 1TB") flag.StringVar(&Settings.outputFileConfig.bufferPath, "output-file-buffer", "/tmp", "The path for temporary storing current buffer: \n\tgor --input-raw :80 --output-file s3://mybucket/logs/%Y-%m-%d.gz --output-file-buffer /mnt/logs") @@ -150,6 +157,9 @@ 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.IntVar(&Settings.copyBufferSize, "copy-buffer-size", 5*1024*1024, "Set the buffer size for an individual request (default 5M)") + flag.BoolVar(&Settings.inputRAWOverrideSnapLen, "input-raw-override-snaplen", false, "Override the capture snaplen to be 64k. Required for some Virtualized environments") + flag.BoolVar(&Settings.inputRAWImmediateMode, "input-raw-immediate-mode", false, "Set pcap interface to immediate mode.") flag.IntVar(&Settings.inputRawBufferSize, "input-raw-buffer-size", 0, "Controls size of the OS buffer (in bytes) which holds packets until they dispatched. Default value depends by system: in Linux around 2MB. If you see big package drop, increase this value.") @@ -161,12 +171,17 @@ func init() { /* outputHTTPConfig */ flag.IntVar(&Settings.outputHTTPConfig.BufferSize, "output-http-response-buffer", 0, "HTTP response buffer size, all data after this size will be discarded.") - flag.IntVar(&Settings.outputHTTPConfig.workers, "output-http-workers", 0, "Gor uses dynamic worker scaling by default. Enter a number to run a set number of workers.") + + flag.IntVar(&Settings.outputHTTPConfig.workersMin, "output-http-workers-min", 0, "Gor uses dynamic worker scaling. Enter a number to set a minimum number of workers. default = 1.") + flag.IntVar(&Settings.outputHTTPConfig.workersMax, "output-http-workers", 0, "Gor uses dynamic worker scaling. Enter a number to set a maximum number of workers. default = 0 = unlimited.") + flag.IntVar(&Settings.outputHTTPConfig.queueLen, "output-http-queue-len", 1000, "Number of requests that can be queued for output, if all workers are busy. default = 1000") + flag.IntVar(&Settings.outputHTTPConfig.redirectLimit, "output-http-redirects", 0, "Enable how often redirects should be followed.") flag.DurationVar(&Settings.outputHTTPConfig.Timeout, "output-http-timeout", 5*time.Second, "Specify HTTP request/response timeout. By default 5s. Example: --output-http-timeout 30s") flag.BoolVar(&Settings.outputHTTPConfig.TrackResponses, "output-http-track-response", false, "If turned on, HTTP output responses will be set to all outputs like stdout, file and etc.") - flag.BoolVar(&Settings.outputHTTPConfig.stats, "output-http-stats", false, "Report http output queue stats to console every 5 seconds.") + flag.BoolVar(&Settings.outputHTTPConfig.stats, "output-http-stats", false, "Report http output queue stats to console every N milliseconds. See output-http-stats-ms") + flag.IntVar(&Settings.outputHTTPConfig.statsMs, "output-http-stats-ms", 5000, "Report http output queue stats to console every N milliseconds. default: 5000") flag.BoolVar(&Settings.outputHTTPConfig.OriginalHost, "http-original-host", false, "Normally gor replaces the Host http header with the host supplied with --output-http. This option disables that behavior, preserving the original Host header.") flag.BoolVar(&Settings.outputHTTPConfig.Debug, "output-http-debug", false, "Enables http debug output.") flag.StringVar(&Settings.outputHTTPConfig.elasticSearch, "output-http-elasticsearch", "", "Send request and response stats to ElasticSearch:\n\tgor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'") diff --git a/snapcraft.yaml b/snapcraft.yaml new file mode 100644 index 0000000..bf3e8f4 --- /dev/null +++ b/snapcraft.yaml @@ -0,0 +1,39 @@ +name: goreplay +version: '1.0' +summary: GoReplay is an open-source tool for capturing and replaying live HTTP traffic +description: | + GoReplay is an open-source tool for capturing and replaying + live HTTP traffic into a test environment in order to continuously + test your system with real data. It can be used to increase confidence + in code deployments, configuration changes and infrastructure changes. +grade: stable +confinement: strict +base: core18 +parts: + goreplay: + plugin: go + source: https://github.com/buger/goreplay.git + go-importpath: github.com/buger/goreplay + build-packages: + - build-essential + - libpcap-dev + stage-packages: + - libpcap0.8 + +apps: + goreplay: + command: bin/goreplay + daemon: simple + restart-condition: on-abnormal + plugs: + - home + - network + - network-bind + - network-control + - network-observe + - netlink-connector + - netlink-audit + - bluetooth-control + - firewall-control + - x11 + diff --git a/vendor/vendor.json b/vendor/vendor.json index 5db34eb..94a443d 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -45,10 +45,10 @@ "revisionTime": "2016-05-29T05:00:41Z" }, { - "checksumSHA1": "WT6lYgJhoWbXLpnFOxPISxrL2/o=", + "checksumSHA1": "2TqkKFm8mR2RAtE6JRPcVixW20w=", "path": "github.com/google/gopacket", - "revision": "60ab61cd59496fcfa4d208b265ba79b1e37c1476", - "revisionTime": "2018-05-13T17:29:36Z" + "revision": "f86faeb88894864d4fa0652c24f3413b1e42f4a1", + "revisionTime": "2019-02-11T01:39:29Z" }, { "checksumSHA1": "BM6ZlNJmtKy3GBoWwg2X55gnZ4A=",