diff --git a/.gitignore b/.gitignore index c59f5ba..0acc83e 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,7 @@ *.out *.bin + +*.gz + +*.class diff --git a/.travis.yml b/.travis.yml index d491b84..a1eaa09 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,3 +1,3 @@ language: go go: 1.4.2 -script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4.2)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test ./... -v -timeout 15s" \ No newline at end of file +script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4.2)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test ./... -v -timeout 60s -race" \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index f09d7f1..bf86e03 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,9 +2,23 @@ FROM google/golang:1.4 RUN cd /goroot/src/ && GOOS=linux GOARCH=386 ./make.bash --no-clean -WORKDIR /gopath/src/github.com/buger/gor/ +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 wget http://apache-mirror.rbc.ru/pub/apache//commons/io/binaries/commons-io-2.4-bin.tar.gz -P /tmp +RUN tar xzf /tmp/commons-io-2.4-bin.tar.gz -C /tmp + +WORKDIR /gopath/src/github.com/buger/gor/ ADD . /gopath/src/github.com/buger/gor/ +RUN javac -cp /tmp/commons-io-2.4/commons-io-2.4.jar ./examples/middleware/echo.java + RUN go get -u github.com/golang/lint/golint RUN go get \ No newline at end of file diff --git a/Makefile b/Makefile index 0e39f2d..339df03 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go input_http.go output_http.go output_tcp.go plugins.go settings.go test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.go +SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go input_http.go output_http.go output_tcp.go plugins.go settings.go test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.go middleware.go protocol.go SOURCE_PATH = /gopath/src/github.com/buger/gor/ @@ -20,7 +20,7 @@ drace: docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test ./... $(ARGS) -v -race -timeout 15s dtest: - docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go test ./... $(ARGS) -v -timeout 10s + docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go test ./... $(ARGS) -v -timeout 60s dcover: docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test $(ARGS) -race -v -timeout 15s -coverprofile=coverage.out @@ -37,13 +37,16 @@ dbench: # Used mainly for debugging, because docker container do not have access to parent machine ports drun: - docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --verbose + docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw :9000 --input-http :9000 --verbose --debug --middleware "./examples/middleware/echo.sh" + +drun-2: + docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-file="./fixtures/requests.gor" --output-dummy=0 --verbose --debug --middleware "java -cp ./examples/middleware echo" drecord: - docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-file=requests.bin --verbose + docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug dreplay: docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-file=requests.bin --output-tcp=:9000 --verbose -h dbash: - docker run -v `pwd`:$(SOURCE_PATH) -t -i gor /bin/bash \ No newline at end of file + docker run -v `pwd`:$(SOURCE_PATH) -t -i gor /bin/bash diff --git a/README.md b/README.md index 2ea19c3..11415f4 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ gor --input-tcp :28020 --output-http "http://staging.com" --output-http "http:/ ``` ### HTTP output workers -By default Gor creates dynamic pull of workers: it starts with 10 and create more http output workers when the http output queue length is greater than 10. The number of workers created (N) is equal to the queue length at the time which it is checked and found to have a length greater than 10. The queue length is checked every time a message is written to the http output queue. No more workers will be spawned until that request to spawn N workers is satisfied. If a dynamic worker cannot process a message at that time, it will sleep for 100 milliseconds. If a dynamic worker cannot process a message for 2 seconds it dies. +By default Gor creates dynamic pull of workers: it starts with 10 and create more http output workers when the http output queue length is greater than 10. The number of workers created (N) is equal to the queue length at the time which it is checked and found to have a length greater than 10. The queue length is checked every time a message is written to the http output queue. No more workers will be spawned until that request to spawn N workers is satisfied. If a dynamic worker cannot process a message at that time, it will sleep for 100 milliseconds. If a dynamic worker cannot process a message for 2 seconds it dies. You may specify fixed number of workers using `--output-http-workers=20` option. ### Follow redirects @@ -149,7 +149,7 @@ gor --input-raw :80 --output-http "http://staging.server" \ ``` ### Rewriting original request -Gor supports built-in basic rewriting support, for complex logic see https://github.com/buger/gor/pull/162 +Gor supports some basic request rewriting support. For complex logic you can use middleware, see below. #### Rewrite URL based on a mapping ``` @@ -177,6 +177,74 @@ Host header gets special treatment. By default Host get set to the value specifi If you app accepts traffic from multiple domain, and you want to keep original headers, there is specific `--http-original-host` with tells Gor do not touch Host header at all. +### Middleware +Middleware is a program that accepts request and response payload at STDIN and emits modified requests at STDOUT. You can implement any custom logic like stripping private data, advanced rewriting, support for oAuth and etc. + +``` + Original request +--------------+ ++-------------+----------STDIN---------->+ | +| Gor input | | Middleware | ++-------------+----------STDIN---------->+ | + Original response +------+---+---+ + | ^ ++-------------+ Modified request v | +| Gor output +<---------STDOUT-----------------+ | ++-----+-------+ | + | | + | Replayed response | + +------------------STDIN----------------->----+ +``` + +Middleware can be written in any language, see `examples/middleware` folder for examples. +Middleware program should accept the fact that all communication with Gor is asynchronous, there is no guarantee that original request and response messages will come one after each other. Your app should take care of the state if logic depends on original or replayed response, see `examples/middleware/token_modifier.go` as example. + +Simple bash echo middleware (returns same request) will look like this: +```bash +while read line; do + echo $line +end +``` + +Middleware can be enabled using `--middleware` option, by specifying path to executable file: +``` +gor --input-raw :80 --middleware "/opt/middleware_executable" --output-http "http://staging.server" +``` + +#### Communication protocol +All messages should be hex encoded, new line character specifieds the end of the message, eg. new message per line. + +Decoded payload consist of 2 parts: header and HTTP payload, separated by new line character. + +Example request payload: + +``` +1 932079936fa4306fc308d67588178d17d823647c 1439818823587396305 +GET /a HTTP/1.1 +Host: 127.0.0.1 + +``` + +Example response payload: + +``` +2 8e091765ae902fef8a2b7d9dd960e9d52222bd8c 2782013 +HTTP/1.1 200 OK +Date: Mon, 17 Aug 2015 13:40:23 GMT +Content-Length: 0 +Content-Type: text/plain; charset=utf-8 + +``` + +Header contains request meta information separated by spaces. First value is payload type, possible values: `1` - request, `2` - original response, `3` - replayed response. +Next goes request id: unique among all requests (sha1 of time and Ack), but remain same for original and replayed response, so you can create associations between request and responses. Third argument varies depending on payload type: for request - start time, for responses - round-trip time. + +HTTP payload is unmodified HTTP requests/responses intercepted from network. You can read more about request format [here](http://www.jmarshall.com/easy/http/), [here](https://en.wikipedia.org/wiki/Hypertext_Transfer_Protocol) and [here](http://www.w3.org/Protocols/rfc2616/rfc2616.html). You can operate with payload as you want, add headers, change path, and etc. Basically you just editing a string, just ensure that it is RCF compliant. + +At the end modified (or untouched) request should be emitted back to STDOUT, keeping original header, and hex-encoded. If you want to filter request, just not send it. Emitting responses back is required, even if you did not touch them. + +#### Advanced example +Imagine that you have auth system that randomly generate access tokens, which used later for accessing secure content. Since there is no pre-defined token value, naive approach without middleware (or if middleware use only request payloads) will fail, because replayed server have own tokens, not synced with origin. To fix this, our middleware should take in account responses of replayed and origin server, store `originalToken -> replayedToken` aliases and rewrite all requests using this token to use replayed alias. See `examples/middleware/token_modifier.go` and `middleware_test.go#TestTokenMiddleware` as example of described scheme. + ### Saving requests to file and replaying them You can save requests to file, and replay them later: ``` diff --git a/emitter.go b/emitter.go index 1f7caa0..d767d45 100644 --- a/emitter.go +++ b/emitter.go @@ -1,21 +1,39 @@ package main import ( + "bytes" "io" "time" ) // Start initialize loop for sending data from inputs to outputs func Start(stop chan int) { - for _, in := range Plugins.Inputs { - go CopyMulty(in, Plugins.Outputs...) + if Settings.middleware != "" { + middleware := NewMiddleware(Settings.middleware) + + for _, in := range Plugins.Inputs { + middleware.ReadFrom(in) + } + + // We going only to read responses, so using same ReadFrom method + for _, out := range Plugins.Outputs { + if r, ok := out.(io.Reader); ok { + middleware.ReadFrom(r) + } + } + + go CopyMulty(middleware, Plugins.Outputs...) + } else { + for _, in := range Plugins.Inputs { + go CopyMulty(in, Plugins.Outputs...) + } } for { select { case <-stop: return - case <-time.After(1 * time.Second): + case <-time.After(time.Second): } } } @@ -28,20 +46,37 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { for { nr, er := src.Read(buf) + if nr > 0 && len(buf) > nr { payload := buf[0:nr] - if modifier != nil { - payload = modifier.Rewrite(payload) - - // If modifier tells to skip request - if len(payload) == 0 { - continue - } + _maxN := nr + if nr > 500 { + _maxN = 500 } if Settings.debug { - Debug("[EMITTER] Sending payload, size:", len(payload), "First 500 bytes:", string(payload[0:500])) + Debug("[EMITTER] input:", string(payload[0:_maxN])) + } + + if modifier != nil && isRequestPayload(payload) { + headSize := bytes.IndexByte(payload, '\n') + 1 + body := payload[headSize:] + originalBodyLen := len(body) + body = modifier.Rewrite(body) + + // If modifier tells to skip request + if len(body) == 0 { + continue + } + + if originalBodyLen != len(body) { + payload = append(payload[:headSize], body...) + } + + if Settings.debug { + Debug("[EMITTER] Rewrittern input:", len(payload), "First 500 bytes:", string(payload[0:_maxN])) + } } if Settings.splitOutput { diff --git a/examples/middleware/echo.java b/examples/middleware/echo.java new file mode 100644 index 0000000..707193e --- /dev/null +++ b/examples/middleware/echo.java @@ -0,0 +1,30 @@ +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; + +import org.apache.commons.io.IOUtils; + +public class echo { + public static void main(String[] args) { + if(args != null){ + for(String arg : args){ + System.out.println(arg); + } + + } + + BufferedReader stdin = new BufferedReader(new InputStreamReader( + System.in)); + String line = null; + + try { + while ((line = stdin.readLine()) != null) { + + System.out.println(line); + + } + } catch (IOException e) { + IOUtils.closeQuietly(stdin); + } + } +} \ No newline at end of file diff --git a/examples/middleware/echo.rb b/examples/middleware/echo.rb new file mode 100755 index 0000000..084acc1 --- /dev/null +++ b/examples/middleware/echo.rb @@ -0,0 +1,15 @@ +#!/usr/bin/env ruby +# encoding: utf-8 +while data = STDIN.gets + next unless data + data = data.chomp + + decoded = [data].pack("H*") + encoded = decoded.unpack("H*").first + + STDOUT.puts encoded + + STDERR.puts "[DEBUG][MIDDLEWARE] Original data: #{data}" + STDERR.puts "[DEBUG][MIDDLEWARE] Decoded request: #{decoded}" + STDERR.puts "[DEBUG][MIDDLEWARE] Encoded data: #{encoded}" +end diff --git a/examples/middleware/echo.sh b/examples/middleware/echo.sh new file mode 100755 index 0000000..ec946ee --- /dev/null +++ b/examples/middleware/echo.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +# +# `xxd` utility included into vim-common package +# It allow hex decoding/encoding + +function log { + # Logging to stderr, because stdout/stdin used for data transfer + >&2 echo "[DEBUG][ECHO] $1" +} + +while read line; do + decoded=$(echo -e "$line" | xxd -r -p) + + header=$(echo -e "$decoded" | head -n +1) + payload=$(echo -e "$decoded" | tail -n +2) + + encoded=$(echo -e "$header\n$payload" | xxd -p | tr -d "\\n") + + log "" + log "===================================" + + case ${header:0:1} in + "1") + log "Request type: Request" + ;; + "2") + log "Request type: Original Response" + ;; + "3") + log "Request type: Replayed Response" + ;; + *) + log "Unknown request type $header" + esac + echo "$encoded" + + log "===================================" + + log "Original data: $line" + log "Decoded request: $decoded" + log "Encoded data: $encoded" +done; diff --git a/examples/middleware/token_modifier.go b/examples/middleware/token_modifier.go new file mode 100644 index 0000000..021408d --- /dev/null +++ b/examples/middleware/token_modifier.go @@ -0,0 +1,119 @@ +/* +This middleware made for auth system that randomly generate access tokens, which used later for accessing secure content. Since there is no pre-defined token value, naive approach without middleware (or if middleware use only request payloads) will fail, because replayed server have own tokens, not synced with origin. To fix this, our middleware should take in account responses of replayed and origin server, store `originalToken -> replayedToken` aliases and rewrite all requests using this token to use replayed alias. See `middleware_test.go#TestTokenMiddleware` test for examples of using this middleware. + +How middleware works: + + Original request +--------------+ ++-------------+----------STDIN---------->+ | +| Gor input | | Middleware | ++-------------+----------STDIN---------->+ | + Original response +------+---+---+ + | ^ ++-------------+ Modified request v | +| Gor output +<---------STDOUT-----------------+ | ++-----+-------+ | + | | + | Replayed response | + +------------------STDIN----------------->----+ +*/ + +package main + +import ( + "bufio" + "bytes" + "encoding/hex" + "fmt" + "github.com/buger/gor/proto" + "os" +) + +// requestID -> originalToken +var originalTokens map[string][]byte + +// originalToken -> replayedToken +var tokenAliases map[string][]byte + +func main() { + originalTokens = make(map[string][]byte) + tokenAliases = make(map[string][]byte) + + scanner := bufio.NewScanner(os.Stdin) + + for scanner.Scan() { + encoded := scanner.Bytes() + buf := make([]byte, len(encoded)/2) + hex.Decode(buf, encoded) + + go process(buf) + } +} + +func process(buf []byte) { + // First byte indicate payload type, possible values: + // 1 - Request + // 2 - Response + // 3 - ReplayedResponse + payloadType := buf[0] + headerSize := bytes.IndexByte(buf, '\n') + 1 + header := buf[:headerSize-1] + + // Header contains space separated values of: request type, request id, and request start time (or round-trip time for responses) + meta := bytes.Split(header, []byte(" ")) + // For each request you should receive 3 payloads (request, response, replayed response) with same request id + reqID := string(meta[1]) + payload := buf[headerSize:] + + Debug("Received payload:", string(buf)) + + switch payloadType { + case '1': // Request + if bytes.Equal(proto.Path(payload), []byte("/token")) { + originalTokens[reqID] = []byte{} + Debug("Found token request:", reqID) + } else { + token, vs, _ := proto.PathParam(payload, []byte("token")) + + if vs != -1 { // If there is GET token param + if alias, ok := tokenAliases[string(token)]; ok { + // Rewrite original token to alias + payload = proto.SetPathParam(payload, []byte("token"), alias) + + // Copy modified payload to our buffer + buf = append(buf[:headerSize], payload...) + } + } + } + + // Emitting data back + os.Stdout.Write(encode(buf)) + case '2': // Original response + if _, ok := originalTokens[reqID]; ok { + // Token is inside response body + secureToken := proto.Body(payload) + originalTokens[reqID] = secureToken + Debug("Remember origial token:", string(secureToken)) + } + case '3': // Replayed response + if originalToken, ok := originalTokens[reqID]; ok { + delete(originalTokens, reqID) + secureToken := proto.Body(payload) + tokenAliases[string(originalToken)] = secureToken + + Debug("Create alias for new token token, was:", string(originalToken), "now:", string(secureToken)) + } + } +} + +func encode(buf []byte) []byte { + dst := make([]byte, len(buf)*2+1) + hex.Encode(dst, buf) + dst[len(dst)-1] = '\n' + + return dst +} + +func Debug(args ...interface{}) { + fmt.Fprint(os.Stderr, "[DEBUG][TOKEN-MOD] ") + fmt.Fprintln(os.Stderr, args...) +} diff --git a/gor.go b/gor.go index af9bf5f..4cd8596 100644 --- a/gor.go +++ b/gor.go @@ -8,7 +8,7 @@ import ( "log" "os" "runtime" - "runtime/debug" + _ "runtime/debug" "runtime/pprof" "time" ) @@ -20,20 +20,18 @@ var ( ) func main() { + // // Don't exit on panic + // defer func() { + // if r := recover(); r != nil { + // fmt.Printf("PANIC: pkg: %v %s \n", r, debug.Stack()) + // } + // }() + // If not set via env cariable if len(os.Getenv("GOMAXPROCS")) == 0 { runtime.GOMAXPROCS(runtime.NumCPU() * 2) } - // Don't exit on panic - defer func() { - if r := recover(); r != nil { - if _, ok := r.(error); !ok { - fmt.Printf("PANIC: pkg: %v %s \n", r, debug.Stack()) - } - } - }() - fmt.Println("Version:", VERSION) flag.Parse() diff --git a/http_client.go b/http_client.go index f9918b7..6ec1227 100644 --- a/http_client.go +++ b/http_client.go @@ -18,11 +18,12 @@ var defaultPorts = map[string]string{ } type HTTPClientConfig struct { - FollowRedirects int - Debug bool - OriginalHost bool - Timeout time.Duration - ResponseBufferSize int + FollowRedirects int + Debug bool + OriginalHost bool + ConnectionTimeout time.Duration + Timeout time.Duration + ResponseBufferSize int } type HTTPClient struct { @@ -52,7 +53,7 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient { } if config.ResponseBufferSize == 0 { - config.ResponseBufferSize = 512*1024 // 500kb + config.ResponseBufferSize = 512 * 1024 // 500kb } client := new(HTTPClient) @@ -69,9 +70,9 @@ func (c *HTTPClient) Connect() (err error) { c.Disconnect() if !strings.Contains(c.host, ":") { - c.conn, err = net.Dial("tcp", c.host + ":80") + c.conn, err = net.DialTimeout("tcp", c.host+":80", c.config.ConnectionTimeout) } else { - c.conn, err = net.Dial("tcp", c.host) + c.conn, err = net.DialTimeout("tcp", c.host, c.config.ConnectionTimeout) } if c.scheme == "https" { @@ -91,7 +92,7 @@ func (c *HTTPClient) Disconnect() { if c.conn != nil { c.conn.Close() c.conn = nil - Debug("Disconnected: ", c.baseURL) + Debug("[HTTP] Disconnected: ", c.baseURL) } } @@ -124,6 +125,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { Debug("[HTTPClient] Connecting:", c.baseURL) if err = c.Connect(); err != nil { log.Println("[HTTPClient] Connection error:", err) + response = errorPayload(HTTP_CONNECTION_ERROR) return } } @@ -142,6 +144,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { if _, err = c.conn.Write(data); err != nil { Debug("[HTTPClient] Write error:", err, c.baseURL) + response = errorPayload(HTTP_TIMEOUT) return } @@ -160,13 +163,12 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { if err != nil { Debug("[HTTPClient] Response read error", err, c.conn) + response = errorPayload(HTTP_TIMEOUT) return } payload := c.respBuf[:n] - Debug("[HTTPClient] Received:", n) - if c.config.Debug { Debug("[HTTPClient] Received:", string(payload)) } @@ -199,3 +201,28 @@ func (c *HTTPClient) Get(path string) (response []byte, err error) { return c.Send([]byte(payload)) } + +const ( + // https://support.cloudflare.com/hc/en-us/articles/200171936-Error-520-Web-server-is-returning-an-unknown-error + HTTP_UNKNOWN_ERROR = "520" + // https://support.cloudflare.com/hc/en-us/articles/200171916-Error-521-Web-server-is-down + HTTP_CONNECTION_ERROR = "521" + // https://support.cloudflare.com/hc/en-us/articles/200171906-Error-522-Connection-timed-out + HTTP_CONNECTION_TIMEOUT = "522" + // https://support.cloudflare.com/hc/en-us/articles/200171946-Error-523-Origin-is-unreachable + HTTP_UNREACHABLE = "523" + // https://support.cloudflare.com/hc/en-us/articles/200171926-Error-524-A-timeout-occurred + HTTP_TIMEOUT = "524" +) + +var errorPayloadTemplate = "HTTP/1.1 202 Accepted\r\nDate: Mon, 17 Aug 2015 14:10:11 GMT\r\nContent-Length: 0\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n" + +func errorPayload(errorCode string) []byte { + payload := make([]byte, len(errorPayloadTemplate)) + copy(payload, errorPayloadTemplate) + + copy(payload[29:58], []byte(time.Now().Format(time.RFC1123))) + copy(payload[9:12], errorCode) + + return payload +} diff --git a/http_client_test.go b/http_client_test.go index 6109a90..e91284e 100644 --- a/http_client_test.go +++ b/http_client_test.go @@ -3,13 +3,17 @@ package main import ( "bytes" "crypto/rand" + "github.com/buger/gor/proto" "io/ioutil" + _ "log" "net" "net/http" "net/http/httptest" "net/http/httputil" + _ "reflect" "sync" "testing" + "time" _ "time" ) @@ -71,6 +75,7 @@ func TestHTTPClientSend(t *testing.T) { wg.Done() })) + defer server.Close() client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true}) @@ -99,6 +104,7 @@ func TestHTTPClientResponseBuffer(t *testing.T) { wg.Done() })) + defer server.Close() client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: false, ResponseBufferSize: 1024}) @@ -149,6 +155,7 @@ func TestHTTPClientHTTPSSend(t *testing.T) { wg.Done() })) + defer server.Close() client := NewHTTPClient(server.URL, &HTTPClientConfig{}) @@ -167,12 +174,16 @@ func TestHTTPClientServerInstantDisconnect(t *testing.T) { GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") ln, _ := net.Listen("tcp", ":0") + defer ln.Close() go func() { for { - conn, _ := ln.Accept() - conn.Close() + conn, err := ln.Accept() + if err != nil { + break + } + conn.Close() wg.Done() } }() @@ -192,12 +203,13 @@ func TestHTTPClientServerNoKeepAlive(t *testing.T) { GETPayload := []byte("GET / HTTP/1.1\r\n\r\n") ln, _ := net.Listen("tcp", ":0") + defer ln.Close() go func() { for { conn, err := ln.Accept() if err != nil { - // handle error + break } buf := make([]byte, 4096) @@ -237,6 +249,7 @@ func TestHTTPClientRedirect(t *testing.T) { wg.Done() })) + defer server.Close() client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 1, Debug: false}) @@ -268,6 +281,7 @@ func TestHTTPClientRedirectLimit(t *testing.T) { wg.Done() })) + defer server.Close() client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 2, Debug: false}) @@ -291,6 +305,7 @@ func TestHTTPClientHandleHTTP10(t *testing.T) { wg.Done() })) + defer server.Close() client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true}) @@ -299,3 +314,87 @@ func TestHTTPClientHandleHTTP10(t *testing.T) { wg.Wait() } + +func TestHTTPClientErrors(t *testing.T) { + req := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n") + + // Port not exists + client := NewHTTPClient("http://127.0.0.1:1", &HTTPClientConfig{Debug: true}) + if resp, err := client.Send(req); err != nil { + if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) { + t.Error("Should return status 521 for connection refused, instead:", string(s)) + } + } else { + t.Error("Should throw error") + } + + client = NewHTTPClient("http://not.existing", &HTTPClientConfig{Debug: true}) + if resp, err := client.Send(req); err != nil { + if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) { + t.Error("Should return status 521 for no such host, instead:", string(s)) + } + } else { + t.Error("Should throw error") + } + + // Non routable IP address to simulate connection timeout + client = NewHTTPClient("http://10.255.255.1", &HTTPClientConfig{Debug: true, ConnectionTimeout: 100 * time.Millisecond}) + + if resp, err := client.Send(req); err != nil { + if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) { + t.Error("Should return status 521 for io/timeout:", string(s)) + } + } else { + t.Error("Should throw error") + } + + // Connecting but io timeout on read + ln, _ := net.Listen("tcp", ":0") + client = NewHTTPClient("http://"+ln.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond}) + defer ln.Close() + + if resp, err := client.Send(req); err != nil { + if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) { + t.Error("Should return status 524 for io read, instead:", string(s)) + } + } else { + t.Error("Should throw error") + } + + // Response read error read tcp [::1]:51128: connection reset by peer &{{0xc20802a000}} + ln1, _ := net.Listen("tcp", ":0") + go func() { + ln1.Accept() + }() + defer ln1.Close() + + client = NewHTTPClient("http://"+ln1.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond}) + + if resp, err := client.Send(req); err != nil { + if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) { + t.Error("Should return status 524 for connection reset by peer, instead:", string(s)) + } + } else { + t.Error("Should throw error") + } + + ln2, _ := net.Listen("tcp", ":0") + go func() { + buf := make([]byte, 64*1024) + conn, _ := ln2.Accept() + + conn.Read(buf) + defer conn.Close() + }() + defer ln2.Close() + + client = NewHTTPClient("http://"+ln2.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond}) + + if resp, err := client.Send(req); err != nil { + if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) { + t.Error("Should return status 524 for connection reset by peer, instead:", string(s)) + } + } else { + t.Error("Should throw error") + } +} diff --git a/http_modifier.go b/http_modifier.go index 14cdf85..fbca0c3 100644 --- a/http_modifier.go +++ b/http_modifier.go @@ -30,6 +30,10 @@ func NewHTTPModifier(config *HTTPModifierConfig) *HTTPModifier { } func (m *HTTPModifier) Rewrite(payload []byte) (response []byte) { + if !proto.IsHTTPPayload(payload) { + return payload + } + if len(m.config.methods) > 0 { method := proto.Method(payload) diff --git a/input_dummy.go b/input_dummy.go index 7f6a7f8..17907d3 100644 --- a/input_dummy.go +++ b/input_dummy.go @@ -21,6 +21,7 @@ func NewDummyInput(options string) (di *DummyInput) { func (i *DummyInput) Read(data []byte) (int, error) { buf := <-i.data + copy(data, buf) return len(buf), nil @@ -32,7 +33,12 @@ func (i *DummyInput) emit() { for { select { case <-ticker.C: - i.data <- []byte("GET / HTTP/1.1\r\n\r\n") + uuid := uuid() + reqh := payloadHeader(RequestPayload, uuid, time.Now().UnixNano()) + i.data <- append(reqh, []byte("POST /pub/WWW/Ć„ HTTP/1.1\nHost: www.w3.org\r\nContent-Length: 7\r\n\r\na=1&b=2")...) + + resh := payloadHeader(ResponsePayload, uuid, 1) + i.data <- append(resh, []byte("HTTP/1.1 200 OK\r\n\r\n")...) } } } diff --git a/input_file.go b/input_file.go index 60c93ec..60ec958 100644 --- a/input_file.go +++ b/input_file.go @@ -1,9 +1,10 @@ package main import ( - "encoding/gob" + "bufio" "log" "os" + "strconv" "time" ) @@ -11,7 +12,7 @@ import ( type FileInput struct { data chan []byte path string - decoder *gob.Decoder + file *os.File speedFactor float64 } @@ -35,7 +36,7 @@ func (i *FileInput) init(path string) { log.Fatal(i, "Cannot open file %q. Error: %s", path, err) } - i.decoder = gob.NewDecoder(file) + i.file = file } func (i *FileInput) Read(data []byte) (int, error) { @@ -52,27 +53,33 @@ func (i *FileInput) String() string { func (i *FileInput) emit() { var lastTime int64 - for { - raw := new(RawRequest) - err := i.decoder.Decode(raw) + // reader := bufio.NewReader(conn) + scanner := bufio.NewScanner(i.file) + scanner.Split(payloadScanner) - if err != nil { - return - } + for scanner.Scan() { + buf := scanner.Bytes() + meta := payloadMeta(buf) - if lastTime != 0 { - timeDiff := raw.Timestamp - lastTime + if meta[0][0] == RequestPayload && lastTime != 0 { + ts, _ := strconv.ParseInt(string(meta[2]), 10, 64) + timeDiff := ts - lastTime - // We can speedup or slowdown execution based on speedFactor if i.speedFactor != 1 { - timeDiff = int64(float64(raw.Timestamp-lastTime) / i.speedFactor) + timeDiff = int64(float64(timeDiff) / i.speedFactor) } time.Sleep(time.Duration(timeDiff)) + + lastTime = ts } - lastTime = raw.Timestamp + // scanner returs only pointer, so to remove data-race we have to allocate new array + newBuf := make([]byte, len(buf)) + copy(newBuf, buf) - i.data <- raw.Request + i.data <- newBuf } + + log.Println("FileInput: end of file") } diff --git a/input_http.go b/input_http.go index 5150c4e..f7717c5 100644 --- a/input_http.go +++ b/input_http.go @@ -5,6 +5,7 @@ import ( "net" "net/http" "net/http/httputil" + "time" ) // HTTPInput used for sending requests to Gor via http @@ -27,9 +28,13 @@ func NewHTTPInput(address string) (i *HTTPInput) { func (i *HTTPInput) Read(data []byte) (int, error) { buf := <-i.data - copy(data, buf) - return len(buf), nil + header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano()) + + copy(data[0:len(header)], header) + copy(data[len(header):], buf) + + return len(buf) + len(header), nil } func (i *HTTPInput) handler(w http.ResponseWriter, r *http.Request) { diff --git a/input_raw.go b/input_raw.go index 93b007d..ed14707 100644 --- a/input_raw.go +++ b/input_raw.go @@ -5,19 +5,25 @@ import ( "log" "net" "strings" + "time" ) // RAWInput used for intercepting traffic for given address type RAWInput struct { - data chan []byte - address string + data chan *raw.TCPMessage + address string + expire time.Duration + quit chan bool + listener *raw.Listener } // NewRAWInput constructor for RAWInput. Accepts address with port as argument. -func NewRAWInput(address string) (i *RAWInput) { +func NewRAWInput(address string, expire time.Duration) (i *RAWInput) { i = new(RAWInput) - i.data = make(chan []byte) + i.data = make(chan *raw.TCPMessage) i.address = address + i.expire = expire + i.quit = make(chan bool) go i.listen(address) @@ -25,31 +31,55 @@ func NewRAWInput(address string) (i *RAWInput) { } func (i *RAWInput) Read(data []byte) (int, error) { - buf := <-i.data - copy(data, buf) + msg := <-i.data + buf := msg.Bytes() - return len(buf), nil + var header []byte + + if msg.IsIncoming { + header = payloadHeader(RequestPayload, msg.UUID(), msg.Start) + } else { + header = payloadHeader(ResponsePayload, msg.UUID(), msg.End-msg.RequestStart) + } + + copy(data[0:len(header)], header) + copy(data[len(header):], buf) + + return len(buf) + len(header), nil } func (i *RAWInput) listen(address string) { address = strings.Replace(address, "[::]", "127.0.0.1", -1) + Debug("Listening for traffic on: " + address) + host, port, err := net.SplitHostPort(address) if err != nil { log.Fatal("input-raw: error while parsing address", err) } - listener := raw.NewListener(host, port) + i.listener = raw.NewListener(host, port, i.expire, true) for { - // Receiving TCPMessage object - m := listener.Receive() + select { + case <-i.quit: + return + default: + } - i.data <- m.Bytes() + // Receiving TCPMessage object + m := i.listener.Receive() + + i.data <- m } } func (i *RAWInput) String() string { return "RAW Socket input: " + i.address } + +func (i *RAWInput) Close() { + i.listener.Close() + close(i.quit) +} diff --git a/input_raw_test.go b/input_raw_test.go index 4bcca9a..0d75443 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -11,40 +11,56 @@ import ( "os/exec" "strings" "sync" + "sync/atomic" "testing" "time" ) -func TestRAWInput(t *testing.T) { +const testRawExpire = time.Millisecond * 200 +func TestRAWInput(t *testing.T) { wg := new(sync.WaitGroup) quit := make(chan int) - listener := startHTTP(func(req *http.Request) {}) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + defer origin.Close() + originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) + + var respCounter, reqCounter int64 + + input := NewRAWInput(originAddr, testRawExpire) + defer input.Close() - input := NewRAWInput(listener.Addr().String()) output := NewTestOutput(func(data []byte) { + if data[0] == '1' { + atomic.AddInt64(&reqCounter, 1) + } else { + atomic.AddInt64(&respCounter, 1) + } + + if Settings.debug { + log.Println(reqCounter, respCounter) + } + wg.Done() }) Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{output} - address := strings.Replace(listener.Addr().String(), "[::]", "127.0.0.1", -1) - - client := NewHTTPClient(address, &HTTPClientConfig{}) + client := NewHTTPClient(origin.URL, &HTTPClientConfig{}) time.Sleep(time.Millisecond) go Start(quit) for i := 0; i < 100; i++ { - wg.Add(1) + // request + response + wg.Add(2) client.Get("/") } wg.Wait() - close(quit) } @@ -55,46 +71,54 @@ func TestInputRAW100Expect(t *testing.T) { fileContent, _ := ioutil.ReadFile("README.md") // Origing and Replay server initialization - origin := startHTTP(func(req *http.Request) { - defer req.Body.Close() - ioutil.ReadAll(req.Body) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + ioutil.ReadAll(r.Body) wg.Done() - }) + })) + defer origin.Close() - originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) + originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr) + input := NewRAWInput(originAddr, time.Second) + defer input.Close() // We will use it to get content of raw HTTP request testOutput := NewTestOutput(func(data []byte) { - if strings.Contains(string(data), "Expect: 100-continue") { - t.Error("Should not contain 100-continue header") + switch data[0] { + case RequestPayload: + if strings.Contains(string(data), "Expect: 100-continue") { + t.Error("Should not contain 100-continue header") + } + wg.Done() + case ResponsePayload: + wg.Done() } - wg.Done() }) - listener := startHTTP(func(req *http.Request) { - defer req.Body.Close() - body, _ := ioutil.ReadAll(req.Body) + replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + body, _ := ioutil.ReadAll(r.Body) if !bytes.Equal(body, fileContent) { - buf, _ := httputil.DumpRequest(req, true) + buf, _ := httputil.DumpRequest(r, true) t.Error("Wrong POST body:", string(buf)) } wg.Done() - }) - replayAddr := listener.Addr().String() + })) + defer replay.Close() - httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{}) + httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{}) Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{testOutput, httpOutput} go Start(quit) - wg.Add(3) + // Origin + Response/Request Test Output + Request Http Output + wg.Add(4) curl := exec.Command("curl", "http://"+originAddr, "--data-binary", "@README.md") err := curl.Run() if err != nil { @@ -112,31 +136,31 @@ func TestInputRAWChunkedEncoding(t *testing.T) { fileContent, _ := ioutil.ReadFile("README.md") // Origing and Replay server initialization - origin := startHTTP(func(req *http.Request) { - defer req.Body.Close() - ioutil.ReadAll(req.Body) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + ioutil.ReadAll(r.Body) wg.Done() - }) + })) - originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) + originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) + input := NewRAWInput(originAddr, time.Second) + defer input.Close() - input := NewRAWInput(originAddr) - - listener := startHTTP(func(req *http.Request) { - defer req.Body.Close() - body, _ := ioutil.ReadAll(req.Body) + replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + body, _ := ioutil.ReadAll(r.Body) if !bytes.Equal(body, fileContent) { - buf, _ := httputil.DumpRequest(req, true) + buf, _ := httputil.DumpRequest(r, true) t.Error("Wrong POST body:", string(buf)) } wg.Done() - }) - replayAddr := listener.Addr().String() + })) + defer replay.Close() - httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: true}) + httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: true}) Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{httpOutput} @@ -167,8 +191,7 @@ func TestInputRAWLargePayload(t *testing.T) { log.Fatal("dd error:", err) } - // Origing and Replay server initialization - origin := startHTTP(func(req *http.Request) { + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { defer req.Body.Close() body, _ := ioutil.ReadAll(req.Body) @@ -177,10 +200,11 @@ func TestInputRAWLargePayload(t *testing.T) { } wg.Done() - }) - originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1) + })) + originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr) + input := NewRAWInput(originAddr, time.Second) + defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024) diff --git a/input_tcp.go b/input_tcp.go index 47d2c3d..15dffa5 100644 --- a/input_tcp.go +++ b/input_tcp.go @@ -2,7 +2,6 @@ package main import ( "bufio" - "encoding/hex" "fmt" "log" "net" @@ -10,7 +9,6 @@ import ( ) // TCPInput used for internal communication -// It expected hex encoded data type TCPInput struct { data chan []byte address string @@ -62,16 +60,10 @@ func (i *TCPInput) handleConnection(conn net.Conn) { reader := bufio.NewReader(conn) scanner := bufio.NewScanner(reader) + scanner.Split(payloadScanner) for scanner.Scan() { - encodedPayload := scanner.Bytes() - // Hex encoding always 2x number of bytes - decoded := make([]byte, len(encodedPayload)/2) - _, err := hex.Decode(decoded, encodedPayload) - if err != nil { - log.Println("[TCPInput] failed to hex decode TCP payload:", err) - } - i.data <- decoded + i.data <- scanner.Bytes() } if err := scanner.Err(); err != nil { diff --git a/input_tcp_test.go b/input_tcp_test.go index f440c57..d844023 100644 --- a/input_tcp_test.go +++ b/input_tcp_test.go @@ -1,7 +1,6 @@ package main import ( - "encoding/hex" "io" "log" "net" @@ -35,14 +34,12 @@ func TestTCPInput(t *testing.T) { log.Fatal(err) } - msg := []byte("GET / HTTP/1.1\r\n\r\n") + msg := []byte("1 1 1\nGET / HTTP/1.1\r\n\r\n") for i := 0; i < 100; i++ { wg.Add(1) - - encoded := make([]byte, len(msg)*2) - hex.Encode(encoded, msg) - conn.Write(append(encoded, '\n')) + conn.Write(msg) + conn.Write([]byte(payloadSeparator)) } wg.Wait() diff --git a/middleware.go b/middleware.go new file mode 100644 index 0000000..8655e85 --- /dev/null +++ b/middleware.go @@ -0,0 +1,116 @@ +package main + +import ( + "bufio" + "encoding/hex" + "fmt" + "io" + "log" + "os" + "os/exec" + "strings" + "sync" +) + +type Middleware struct { + command string + + data chan []byte + + mu sync.Mutex + + Stdin io.Writer + Stdout io.Reader +} + +func NewMiddleware(command string) *Middleware { + m := new(Middleware) + m.command = command + m.data = make(chan []byte, 1000) + + commands := strings.Split(command, " ") + cmd := exec.Command(commands[0], commands[1:]...) + + m.Stdout, _ = cmd.StdoutPipe() + m.Stdin, _ = cmd.StdinPipe() + + if Settings.verbose { + cmd.Stderr = os.Stderr + } + + go m.read(m.Stdout) + + go func() { + err := cmd.Start() + + if err != nil { + log.Fatal(err) + } + + cmd.Wait() + }() + + return m +} + +func (m *Middleware) ReadFrom(plugin io.Reader) { + Debug("[MIDDLEWARE-MASTER] Starting reading from", plugin) + go m.copy(m.Stdin, plugin) +} + +func (m *Middleware) copy(to io.Writer, from io.Reader) { + buf := make([]byte, 5*1024*1024) + dst := make([]byte, len(buf)*2) + + for { + nr, _ := from.Read(buf) + if nr > 0 && len(buf) > nr { + + hex.Encode(dst, buf[0:nr]) + dst[nr*2] = '\n' + + m.mu.Lock() + to.Write(dst[0 : nr*2+1]) + m.mu.Unlock() + + if Settings.debug { + Debug("[MIDDLEWARE-MASTER] Sending:", string(buf[0:nr]), "From:", from) + } + } + } +} + +func (m *Middleware) read(from io.Reader) { + scanner := bufio.NewScanner(from) + + for scanner.Scan() { + bytes := scanner.Bytes() + buf := make([]byte, len(bytes)/2) + if _, err := hex.Decode(buf, bytes); err != nil { + fmt.Fprintln(os.Stderr, "Failed to decode input payload", err, len(bytes)) + } + + if Settings.debug { + Debug("[MIDDLEWARE-MASTER] Received:", string(buf)) + } + + m.data <- buf + } + + if err := scanner.Err(); err != nil { + fmt.Fprintln(os.Stderr, "Traffic modifier command failed:", err) + } + + return +} + +func (m *Middleware) Read(data []byte) (int, error) { + buf := <-m.data + copy(data, buf) + + return len(buf), nil +} + +func (m *Middleware) String() string { + return fmt.Sprintf("Modifying traffic using '%s' command", m.command) +} diff --git a/middleware_test.go b/middleware_test.go new file mode 100644 index 0000000..8474eeb --- /dev/null +++ b/middleware_test.go @@ -0,0 +1,215 @@ +package main + +import ( + "bytes" + "crypto/rand" + "encoding/hex" + "github.com/buger/gor/proto" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +type fakeServiceCb func(string, int, []byte) + +// Simple service that generate token on request, and require this token for accesing to secure area +func NewFakeSecureService(wg *sync.WaitGroup, cb fakeServiceCb) *httptest.Server { + active_tokens := make([]string, 0) + var mu sync.Mutex + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + mu.Lock() + defer mu.Unlock() + Debug("Received request: " + req.URL.String()) + + switch req.URL.Path { + case "/token": + // Generate random token + token_length := 10 + buf := make([]byte, token_length) + rand.Read(buf) + token := hex.EncodeToString(buf) + active_tokens = append(active_tokens, token) + + w.Write([]byte(token)) + + cb(req.URL.Path, 200, []byte(token)) + case "/secure": + token := req.URL.Query().Get("token") + token_found := false + + for _, t := range active_tokens { + if t == token { + token_found = true + break + } + } + + if token_found { + w.WriteHeader(http.StatusAccepted) + cb(req.URL.Path, 202, []byte(nil)) + } else { + w.WriteHeader(http.StatusForbidden) + cb(req.URL.Path, 403, []byte(nil)) + } + } + + wg.Done() + })) + + return server +} + +func TestFakeSecureService(t *testing.T) { + var resp, token []byte + + wg := new(sync.WaitGroup) + + server := NewFakeSecureService(wg, func(path string, status int, resp []byte) { + }) + defer server.Close() + + wg.Add(3) + + client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true}) + resp, _ = client.Get("/token") + token = proto.Body(resp) + + // Right token + resp, _ = client.Get("/secure?token=" + string(token)) + if !bytes.Equal(proto.Status(resp), []byte("202")) { + t.Error("Valid token should return status 202:", string(proto.Status(resp))) + } + + // Wrong tokens forbidden + resp, _ = client.Get("/secure?token=wrong") + if !bytes.Equal(proto.Status(resp), []byte("403")) { + t.Error("Wrong token should returns status 403:", string(proto.Status(resp))) + } + + wg.Wait() +} + +func TestEchoMiddleware(t *testing.T) { + wg := new(sync.WaitGroup) + + from := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Env", "prod") + w.Header().Set("RequestPath", r.URL.Path) + wg.Done() + })) + defer from.Close() + + to := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Env", "test") + w.Header().Set("RequestPath", r.URL.Path) + wg.Done() + })) + defer to.Close() + + quit := make(chan int) + + Settings.middleware = "./examples/middleware/echo.sh" + + // Catch traffic from one service + input := NewRAWInput(from.Listener.Addr().String(), testRawExpire) + defer input.Close() + + // And redirect to another + output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: false}) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + // Start Gor + go Start(quit) + + time.Sleep(time.Millisecond) + + // Should receive 2 requests from original + 2 from replayed + wg.Add(4) + + client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false}) + + // Request should be echoed + client.Get("/a") + client.Get("/b") + + wg.Wait() + close(quit) + time.Sleep(100 * time.Millisecond) + + Settings.middleware = "" +} + +func TestTokenMiddleware(t *testing.T) { + var resp, token []byte + + wg := new(sync.WaitGroup) + + from := NewFakeSecureService(wg, func(path string, status int, tok []byte) { + time.Sleep(10 * time.Millisecond) + }) + defer from.Close() + + to := NewFakeSecureService(wg, func(path string, status int, tok []byte) { + switch path { + case "/secure": + if status != 202 { + t.Error("Server should receive valid rewritten token") + } + } + + time.Sleep(10 * time.Millisecond) + }) + defer to.Close() + + quit := make(chan int) + + Settings.middleware = "go run ./examples/middleware/token_modifier.go" + + fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1) + // Catch traffic from one service + input := NewRAWInput(fromAddr, testRawExpire) + defer input.Close() + + // And redirect to another + output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: true}) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + // Start Gor + go Start(quit) + + // Wait for middleware to initialize + // Give go compiller time to build programm + time.Sleep(500 * time.Millisecond) + + // Should receive 2 requests from original + 2 from replayed + wg.Add(4) + + client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false}) + + // Sending traffic to original service + resp, _ = client.Get("/token") + token = proto.Body(resp) + + // When delay is too smal, middleware does not always rewrite requests in time + // Hopefuly client will have delay more then 10ms :) + time.Sleep(10 * time.Millisecond) + + resp, _ = client.Get("/secure?token=" + string(token)) + if !bytes.Equal(proto.Status(resp), []byte("202")) { + t.Error("Valid token should return 202:", proto.Status(resp)) + } + + wg.Wait() + close(quit) + time.Sleep(100 * time.Millisecond) + Settings.middleware = "" +} diff --git a/output_file.go b/output_file.go index f6a81ae..5ccc93c 100644 --- a/output_file.go +++ b/output_file.go @@ -1,24 +1,15 @@ package main import ( - "encoding/gob" "io" "log" "os" - "time" ) -// RawRequest stores original start time and request payload -type RawRequest struct { - Timestamp int64 - Request []byte -} - // FileOutput output plugin type FileOutput struct { - path string - encoder *gob.Encoder - file *os.File + path string + file *os.File } // NewFileOutput constructor for FileOutput, accepts path @@ -38,14 +29,15 @@ func (o *FileOutput) init(path string) { if err != nil { log.Fatal(o, "Cannot open file %q. Error: %s", path, err) } - - o.encoder = gob.NewEncoder(o.file) } func (o *FileOutput) Write(data []byte) (n int, err error) { - raw := RawRequest{time.Now().UnixNano(), data} + if !isOriginPayload(data) { + return len(data), nil + } - o.encoder.Encode(raw) + o.file.Write(data) + o.file.Write([]byte(payloadSeparator)) return len(data), nil } diff --git a/output_http.go b/output_http.go index 005ecd8..9f82d37 100644 --- a/output_http.go +++ b/output_http.go @@ -1,6 +1,7 @@ package main import ( + "github.com/buger/gor/proto" "io" "log" "sync/atomic" @@ -9,6 +10,12 @@ import ( const initialDynamicWorkers = 10 +type response struct { + payload []byte + uuid []byte + roundTripTime int64 +} + // HTTPOutputConfig struct for holding http output configuration type HTTPOutputConfig struct { redirectLimit int @@ -22,6 +29,8 @@ type HTTPOutputConfig struct { OriginalHost bool Debug bool + + TrackResponses bool } // HTTPOutput plugin manage pool of workers which send request to replayed server @@ -37,6 +46,8 @@ type HTTPOutput struct { limit int queue chan []byte + responses chan response + needWorker chan int config *HTTPOutputConfig @@ -49,7 +60,6 @@ type HTTPOutput struct { // NewHTTPOutput constructor for HTTPOutput // Initialize workers func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { - o := new(HTTPOutput) o.address = address @@ -60,6 +70,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { } o.queue = make(chan []byte, 100) + o.responses = make(chan response, 100) o.needWorker = make(chan int, 1) // Initial workers count @@ -74,6 +85,10 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o.elasticSearch.Init(o.config.elasticSearch) } + if len(Settings.middleware) > 0 { + o.config.TrackResponses = true + } + go o.workerMaster() return o @@ -132,6 +147,10 @@ func (o *HTTPOutput) startWorker() { } func (o *HTTPOutput) Write(data []byte) (n int, err error) { + if !isRequestPayload(data) { + return len(data), nil + } + buf := make([]byte, len(data)) copy(buf, data) @@ -152,15 +171,39 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) { return len(data), nil } +func (o *HTTPOutput) Read(data []byte) (int, error) { + resp := <-o.responses + + Debug("[OUTPUT-HTTP] Received response:", string(resp.payload)) + + header := payloadHeader(ReplayedResponsePayload, resp.uuid, resp.roundTripTime) + copy(data[0:len(header)], header) + copy(data[len(header):], resp.payload) + + return len(resp.payload) + len(header), nil +} + func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) { + meta := payloadMeta(request) + uuid := meta[1] + + body := payloadBody(request) + if !proto.IsHTTPPayload(body) { + return + } + start := time.Now() - resp, err := client.Send(request) + resp, err := client.Send(body) stop := time.Now() if err != nil { log.Println("Request error:", err) } + if o.config.TrackResponses { + o.responses <- response{resp, uuid, stop.UnixNano() - start.UnixNano()} + } + if o.elasticSearch != nil { o.elasticSearch.ResponseAnalyze(request, resp, start, stop) } diff --git a/output_http_test.go b/output_http_test.go index 3d32734..b97399e 100644 --- a/output_http_test.go +++ b/output_http_test.go @@ -3,7 +3,6 @@ package main import ( "io" "io/ioutil" - "net" "net/http" "net/http/httptest" _ "net/http/httputil" @@ -12,25 +11,13 @@ import ( "time" ) -func startHTTP(cb func(*http.Request)) net.Listener { - handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - cb(r) - }) - - listener, _ := net.Listen("tcp", ":0") - - go http.Serve(listener, handler) - - return listener -} - func TestHTTPOutput(t *testing.T) { wg := new(sync.WaitGroup) quit := make(chan int) input := NewTestInput() - listener := startHTTP(func(req *http.Request) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { if req.Header.Get("User-Agent") != "Gor" { t.Error("Wrong header") } @@ -49,13 +36,14 @@ func TestHTTPOutput(t *testing.T) { } wg.Done() - }) + })) + defer server.Close() headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}} methods := HTTPMethods{[]byte("GET"), []byte("PUT"), []byte("POST")} Settings.modifierConfig = HTTPModifierConfig{headers: headers, methods: methods} - output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: true}) Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{output} @@ -82,18 +70,19 @@ func TestHTTPOutputKeepOriginalHost(t *testing.T) { input := NewTestInput() - listener := startHTTP(func(req *http.Request) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { if req.Host != "custom-host.com" { t.Error("Wrong header", req.Host) } wg.Done() - }) + })) + defer server.Close() headers := HTTPHeaders{HTTPHeader{"Host", "custom-host.com"}} Settings.modifierConfig = HTTPModifierConfig{headers: headers} - output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{Debug: false, OriginalHost: true}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: false, OriginalHost: true}) Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{output} @@ -140,13 +129,14 @@ func BenchmarkHTTPOutput(b *testing.B) { wg := new(sync.WaitGroup) quit := make(chan int) - listener := startHTTP(func(req *http.Request) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { time.Sleep(50 * time.Millisecond) wg.Done() - }) + })) + defer server.Close() input := NewTestInput() - output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{output} diff --git a/output_tcp.go b/output_tcp.go index 2656a25..9d46fa7 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -1,7 +1,6 @@ package main import ( - "encoding/hex" "fmt" "io" "log" @@ -47,7 +46,9 @@ func (o *TCPOutput) worker() { defer conn.Close() for { - _, err := conn.Write(<-o.buf) + conn.Write(<-o.buf) + _, err := conn.Write([]byte(payloadSeparator)) + if err != nil { log.Println("Worker failed on write, exitings and starting new worker") go o.worker() @@ -57,11 +58,15 @@ func (o *TCPOutput) worker() { } func (o *TCPOutput) Write(data []byte) (n int, err error) { - // Hex encoding always 2x number of bytes - encoded := make([]byte, len(data)*2+1) - hex.Encode(encoded, data) - encoded[len(encoded)-1] = '\n' - o.buf <- encoded + if !isOriginPayload(data) { + return len(data), nil + } + + // We have to copy, because sending data in multiple threads + newBuf := make([]byte, len(data)) + copy(newBuf, data) + + o.buf <- newBuf if Settings.outputTCPStats { o.bufStats.Write(len(o.buf)) diff --git a/output_tcp_test.go b/output_tcp_test.go index 84c66b8..b295d7a 100644 --- a/output_tcp_test.go +++ b/output_tcp_test.go @@ -2,7 +2,6 @@ package main import ( "bufio" - "encoding/hex" "io" "log" "net" @@ -50,13 +49,10 @@ func startTCP(cb func([]byte)) net.Listener { go func() { reader := bufio.NewReader(conn) scanner := bufio.NewScanner(reader) + scanner.Split(payloadScanner) for scanner.Scan() { - encodedPayload := scanner.Bytes() - // Hex encoding always 2x number of bytes - decoded := make([]byte, len(encodedPayload)/2) - hex.Decode(decoded, encodedPayload) - cb(decoded) + cb(scanner.Bytes()) } }() } diff --git a/plugins.go b/plugins.go index cb189a4..8c08e2f 100644 --- a/plugins.go +++ b/plugins.go @@ -4,6 +4,7 @@ import ( "io" "reflect" "strings" + "time" ) // InOutPlugins struct for holding references to plugins @@ -55,11 +56,15 @@ func registerPlugin(constructor interface{}, options ...interface{}) { pluginWrapper = plugin } - if _, ok := plugin.(io.Reader); ok { + _, isR := plugin.(io.Reader) + _, isW := plugin.(io.Writer) + + // Some of the output can be Readers as well because return responses + if isR && !isW { Plugins.Inputs = append(Plugins.Inputs, pluginWrapper.(io.Reader)) } - if _, ok := plugin.(io.Writer); ok { + if isW { Plugins.Outputs = append(Plugins.Outputs, pluginWrapper.(io.Writer)) } } @@ -75,7 +80,7 @@ func InitPlugins() { } for _, options := range Settings.inputRAW { - registerPlugin(NewRAWInput, options) + registerPlugin(NewRAWInput, options, time.Duration(0)) } for _, options := range Settings.inputTCP { diff --git a/proto/proto.go b/proto/proto.go index f595234..549cec3 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -95,6 +95,12 @@ func AddHeader(payload, name, value []byte) []byte { return byteutils.Insert(payload, mimeStart, header) } +// Body returns request/response body +func Body(payload []byte) []byte { + // 4 -> len(EMPTY_LINE) + return payload[MIMEHeadersEndPos(payload)+4:] +} + // Path takes payload and retuns request path: Split(firstLine, ' ')[1] func Path(payload []byte) []byte { start := bytes.IndexByte(payload, ' ') + 1 @@ -202,3 +208,18 @@ func Method(payload []byte) []byte { func Status(payload []byte) []byte { return Path(payload) } + +var httpMethods []string = []string{ + "GET ", "OPTI", "HEAD", "POST", "PUT ", "DELE", "TRAC", "CONN", +} + +func IsHTTPPayload(payload []byte) bool { + method := string(payload[0:4]) + + for _, m := range httpMethods { + if method == m { + return true + } + } + return false +} diff --git a/protocol.go b/protocol.go new file mode 100644 index 0000000..1705f58 --- /dev/null +++ b/protocol.go @@ -0,0 +1,84 @@ +package main + +import ( + "bytes" + "crypto/rand" + "encoding/hex" + "strconv" +) + +const ( + RequestPayload = '1' + ResponsePayload = '2' + ReplayedResponsePayload = '3' +) + +func uuid() []byte { + b := make([]byte, 20) + rand.Read(b) + + uuid := make([]byte, 40) + hex.Encode(uuid, b) + + return uuid +} + +var payloadSeparator = "\nšŸµšŸ™ˆšŸ™‰\n" + +func payloadScanner(data []byte, atEOF bool) (advance int, token []byte, err error) { + if atEOF && len(data) == 0 { + return 0, nil, nil + } + + if i := bytes.Index(data, []byte(payloadSeparator)); i >= 0 { + // We have a full newline-terminated line. + return i + len([]byte(payloadSeparator)), data[0:i], nil + } + + if atEOF { + return len(data), data, nil + } + return 0, nil, nil +} + +// Timing is request start or round-trip time, depending on payloadType +func payloadHeader(payloadType byte, uuid []byte, timing int64) (header []byte) { + sTime := strconv.FormatInt(timing, 10) + + //Example: + // 3 f45590522cd1838b4a0d5c5aab80b77929dea3b3 1231\n + // `+ 1` indicates space characters or end of line + header = make([]byte, 1+1+len(uuid)+1+len(sTime)+1) + header[0] = payloadType + header[1] = ' ' + header[2+len(uuid)] = ' ' + header[len(header)-1] = '\n' + + copy(header[2:], uuid) + copy(header[3+len(uuid):], sTime) + + return header +} + +func payloadBody(payload []byte) []byte { + headerSize := bytes.IndexByte(payload, '\n') + return payload[headerSize+1:] +} + +func payloadMeta(payload []byte) [][]byte { + headerSize := bytes.IndexByte(payload, '\n') + return bytes.Split(payload[:headerSize], []byte{' '}) +} + +func isOriginPayload(payload []byte) bool { + switch payload[0] { + case RequestPayload, ResponsePayload: + return true + default: + return false + } +} + +func isRequestPayload(payload []byte) bool { + return payload[0] == RequestPayload +} diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index f82ae6f..4fe232d 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -17,7 +17,10 @@ import ( "encoding/binary" "log" "net" + "runtime/debug" "strconv" + "strings" + "time" ) // Listener handle traffic capture @@ -31,6 +34,8 @@ type Listener struct { // To get ACK of second message we need to compute its Seq and wait for them message seqWithData map[uint32]uint32 + respAliases map[uint32]*request + // Messages ready to be send to client packetsChan chan *TCPPacket @@ -41,26 +46,47 @@ type Listener struct { messageDelChan chan *TCPMessage addr string // IP to listen - port int // Port to listen + port uint16 // Port to listen + + messageExpire time.Duration + + captureResponse bool + + conn net.PacketConn + quit chan bool +} + +type request struct { + start int64 + ack uint32 } // NewListener creates and initializes new Listener object -func NewListener(addr string, port string) (rawListener *Listener) { - rawListener = &Listener{} +func NewListener(addr string, port string, expire time.Duration, captureResponse bool) (l *Listener) { + l = &Listener{captureResponse: captureResponse} - rawListener.packetsChan = make(chan *TCPPacket, 10000) - rawListener.messagesChan = make(chan *TCPMessage, 10000) - rawListener.messageDelChan = make(chan *TCPMessage, 10000) + l.packetsChan = make(chan *TCPPacket, 10000) + l.messagesChan = make(chan *TCPMessage, 10000) + l.messageDelChan = make(chan *TCPMessage, 10000) + l.quit = make(chan bool) - rawListener.messages = make(map[string]*TCPMessage) - rawListener.ackAliases = make(map[uint32]uint32) - rawListener.seqWithData = make(map[uint32]uint32) + l.messages = make(map[string]*TCPMessage) + l.ackAliases = make(map[uint32]uint32) + l.seqWithData = make(map[uint32]uint32) + l.respAliases = make(map[uint32]*request) - rawListener.addr = addr - rawListener.port, _ = strconv.Atoi(port) + l.addr = addr + _port, _ := strconv.Atoi(port) + l.port = uint16(_port) - go rawListener.listen() - go rawListener.readRAWSocket() + if expire.Nanoseconds() == 0 { + expire = 2000 * time.Millisecond + } + + l.messageExpire = expire + + go l.listen() + go l.readRAWSocket() return } @@ -68,12 +94,20 @@ func NewListener(addr string, port string) (rawListener *Listener) { func (t *Listener) listen() { for { select { + case <-t.quit: + t.conn.Close() + return // If message ready for deletion it means that its also complete or expired by timeout case message := <-t.messageDelChan: - t.messagesChan <- message delete(t.ackAliases, message.Ack) delete(t.messages, message.ID) + if !message.IsIncoming { + delete(t.respAliases, message.Ack) + } + + t.messagesChan <- message + // We need to use channels to process each packet to avoid data races case packet := <-t.packetsChan: t.processTCPPacket(packet) @@ -82,21 +116,26 @@ func (t *Listener) listen() { } func (t *Listener) readRAWSocket() { conn, e := net.ListenPacket("ip4:tcp", t.addr) + t.conn = conn if e != nil { log.Fatal(e) } - defer conn.Close() + defer t.conn.Close() for { buf := make([]byte, 64*1024) // 64kb // Note: ReadFrom receive messages without IP header - n, addr, err := conn.ReadFrom(buf) + n, addr, err := t.conn.ReadFrom(buf) if err != nil { - log.Println("Error:", err) - continue + if strings.HasSuffix(err.Error(), "closed network connection") { + return + } else { + log.Println("Raw listener error:", err) + continue + } } if n > 0 { @@ -106,18 +145,19 @@ func (t *Listener) readRAWSocket() { } func (t *Listener) parsePacket(addr net.Addr, buf []byte) { - if t.isIncomingDataPacket(buf) { + if t.isValidPacket(buf) { t.packetsChan <- ParseTCPPacket(addr, buf) } } -func (t *Listener) isIncomingDataPacket(buf []byte) bool { +func (t *Listener) isValidPacket(buf []byte) bool { // To avoid full packet parsing every time, we manually parsing values needed for packet filtering // http://en.wikipedia.org/wiki/Transmission_Control_Protocol destPort := binary.BigEndian.Uint16(buf[2:4]) + srcPort := binary.BigEndian.Uint16(buf[0:2]) // Because RAW_SOCKET can't be bound to port, we have to control it by ourself - if int(destPort) == t.port { + if destPort == t.port || (t.captureResponse && srcPort == t.port) { // Get the 'data offset' (size of the TCP header in 32-bit words) dataOffset := (buf[12] & 0xF0) >> 4 @@ -139,10 +179,17 @@ var bPOST = []byte("POST") // // For TCP message unique id is Acknowledgment number (see tcp_packet.go) func (t *Listener) processTCPPacket(packet *TCPPacket) { - defer func() { recover() }() + // Don't exit on panic + defer func() { + if r := recover(); r != nil { + log.Println("PANIC: pkg:", r, string(debug.Stack())) + } + }() var message *TCPMessage + isIncoming := packet.DestPort == t.port + if parentAck, ok := t.seqWithData[packet.Seq]; ok { t.ackAliases[packet.Ack] = parentAck delete(t.seqWithData, packet.Seq) @@ -152,16 +199,30 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { packet.Ack = alias } - mID := packet.Addr.String() + strconv.Itoa(int(packet.SrcPort)) + strconv.Itoa(int(packet.Ack)) + var responseRequest *request + + if t.captureResponse && !isIncoming { + responseRequest, _ = t.respAliases[packet.Ack] + } + + mID := packet.Addr.String() + strconv.Itoa(int(packet.DestPort)) + strconv.Itoa(int(packet.Ack)) + message, ok := t.messages[mID] if !ok { // We sending messageDelChan channel, so message object can communicate with Listener and notify it if message completed - message = NewTCPMessage(mID, t.messageDelChan, packet.Ack) + message = NewTCPMessage(mID, t.messageDelChan, packet.Ack, &t.messageExpire, isIncoming) t.messages[mID] = message + + if !isIncoming && responseRequest != nil { + message.RequestStart = responseRequest.start + message.RequestAck = responseRequest.ack + } } + // Handling Expect: 100-continue requests if bytes.Equal(packet.Data[0:4], bPOST) { + // reading last 20 bytes (not counting CRLF): last header value (if no body presented) if bytes.Equal(packet.Data[len(packet.Data)-24:len(packet.Data)-4], bExpect100ContinueCheck) { t.seqWithData[packet.Seq+uint32(len(packet.Data))] = packet.Ack @@ -170,6 +231,19 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { } } + if t.captureResponse && isIncoming { + message.mu.Lock() + // If message have multiple packets, delete previous alias + if len(message.packets) > 0 { + delete(t.respAliases, message.ResponseAck) + } + message.mu.Unlock() + + responseAck := packet.Seq + uint32(len(packet.Data)) + t.respAliases[responseAck] = &request{message.Start, message.Ack} + message.ResponseAck = responseAck + } + // Adding packet to message message.packetsChan <- packet } @@ -178,3 +252,9 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { func (t *Listener) Receive() *TCPMessage { return <-t.messagesChan } + +func (t *Listener) Close() { + close(t.quit) + t.conn.Close() + return +} diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index ddfb0f2..cabac1d 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -1,14 +1,16 @@ package rawSocket import ( + "bytes" + "crypto/sha1" + "encoding/hex" + "github.com/buger/gor/proto" "log" - "sort" + "strconv" + "sync" "time" ) -// MsgExpire specify period that message should wait before it considered as finished -const MsgExpire = 2000 * time.Millisecond - // TCPMessage ensure that all TCP packets for given request is received, and processed in right sequence // Its needed because all TCP message can be fragmented or re-transmitted // @@ -16,8 +18,15 @@ const MsgExpire = 2000 * time.Millisecond // Message can be compiled from unique packets with same message_id which sorted by sequence // Message is received if we didn't receive any packets for 2000ms type TCPMessage struct { - ID string // Message ID - Ack uint32 + ID string // Message ID + Ack uint32 + ResponseAck uint32 + RequestStart int64 + RequestAck uint32 + Start int64 + End int64 + IsIncoming bool + packets []*TCPPacket timer *time.Timer // Used for expire check @@ -25,17 +34,19 @@ type TCPMessage struct { packetsChan chan *TCPPacket delChan chan *TCPMessage + + expire *time.Duration + + mu sync.Mutex } // NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted -func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32) (msg *TCPMessage) { - msg = &TCPMessage{ID: ID, Ack: Ack} - +func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32, expire *time.Duration, IsIncoming bool) (msg *TCPMessage) { + msg = &TCPMessage{ID: ID, Ack: Ack, expire: expire, IsIncoming: IsIncoming} + msg.Start = time.Now().UnixNano() msg.packetsChan = make(chan *TCPPacket) msg.delChan = delChan // used for notifying that message completed or expired - - // Every time we receive packet we reset this timer - msg.timer = time.AfterFunc(MsgExpire, msg.Timeout) + msg.timer = time.NewTimer(0) go msg.listen() @@ -64,29 +75,44 @@ func (t *TCPMessage) Timeout() { case packet, ok := <-t.packetsChan: if ok { t.AddPacket(packet) + t.Timeout() } else { return } default: close(t.packetsChan) - t.delChan <- t // Notify RAWListener that message is ready to be send to replay server + // Notify RAWListener that message is ready to be send to replay server + // Responses without requests gets discarded + if t.IsIncoming || t.RequestStart != 0 { + t.delChan <- t + } } } -// Bytes sorts packets in right orders and return message content +// Bytes return message content func (t *TCPMessage) Bytes() (output []byte) { - sort.Sort(sortBySeq(t.packets)) - - for _, v := range t.packets { - output = append(output, v.Data...) + for _, p := range t.packets { + output = append(output, p.Data...) } return output } +// Size returns total size of message +func (t *TCPMessage) Size() (size int) { + for _, p := range t.packets { + size += len(p.Data) + } + + return +} + // AddPacket to the message and ensure packet uniqueness // TCP allows that packet can be re-send multiple times func (t *TCPMessage) AddPacket(packet *TCPPacket) { + t.mu.Lock() + defer t.mu.Unlock() + packetFound := false for _, pkt := range t.packets { @@ -99,9 +125,88 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { if packetFound { log.Println("Received packet with same sequence") } else { - t.packets = append(t.packets, packet) + // Packets not always captured in same Seq order, and sometimes we need to prepend + if len(t.packets) == 0 || packet.Seq > t.packets[len(t.packets)-1].Seq { + t.packets = append(t.packets, packet) + } else { + t.packets = append([]*TCPPacket{packet}, t.packets...) + } + + t.End = time.Now().UnixNano() } - // Reset message timeout timer - t.timer.Reset(MsgExpire) + if !t.isMultipart() { + t.Timeout() + } else { + // If more then 1 packet, wait for more, and set expiration + if len(t.packets) == 1 { + // Every time we receive packet we reset this timer + t.timer = time.AfterFunc(*t.expire, t.Timeout) + } else { + // Reset message timeout timer + t.timer.Reset(*t.expire) + } + } +} + +// isMultipart returns true if message contains from multiple tcp packets +func (t *TCPMessage) isMultipart() bool { + if len(t.packets) > 1 { + return true + } + + payload := t.packets[0].Data + m := payload[:4] + + if t.IsIncoming { + // If one GET, OPTIONS, or HEAD request + if bytes.Equal(m, []byte("GET ")) || bytes.Equal(m, []byte("OPTI")) || bytes.Equal(m, []byte("HEAD")) { + return false + } else { + // Sometimes header comes after the body :( + if bytes.Equal(m, []byte("POST")) || bytes.Equal(m, []byte("PUT ")) || bytes.Equal(m, []byte("PATC")) { + if length := proto.Header(payload, []byte("Content-Length")); len(length) > 0 { + l, _ := strconv.Atoi(string(length)) + + // If content-length equal current body length + if l > 0 && l == len(proto.Body(payload)) { + return false + } + } + } + } + } else { + if length := proto.Header(payload, []byte("Content-Length")); len(length) > 0 { + if length[0] == '0' { + return false + } + + l, _ := strconv.Atoi(string(length)) + + // If content-length equal current body length + if l > 0 && l == len(proto.Body(payload)) { + return false + } + } + } + + return true +} + +func (t *TCPMessage) UUID() []byte { + var key []byte + + if t.IsIncoming { + key = strconv.AppendInt(key, t.Start, 10) + key = strconv.AppendUint(key, uint64(t.Ack), 10) + } else { + key = strconv.AppendInt(key, t.RequestStart, 10) + key = strconv.AppendUint(key, uint64(t.RequestAck), 10) + } + + uuid := make([]byte, 40) + sha := sha1.Sum(key) + hex.Encode(uuid, sha[:20]) + + return uuid } diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index 3ebcf88..e1ef358 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/raw_socket_listener/tcp_packet.go @@ -50,7 +50,6 @@ func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) { // Parse TCP Packet, inspired by: https://github.com/miekg/pcap/blob/master/packet.go func (t *TCPPacket) Parse() { t.ParseBasic() - t.DestPort = binary.BigEndian.Uint16(t.Data[2:4]) t.Flags = binary.BigEndian.Uint16(t.Data[12:14]) & 0x1FF t.Window = binary.BigEndian.Uint16(t.Data[14:16]) t.Checksum = binary.BigEndian.Uint16(t.Data[16:18]) @@ -59,6 +58,7 @@ func (t *TCPPacket) Parse() { // ParseBasic set of fields func (t *TCPPacket) ParseBasic() { + t.DestPort = binary.BigEndian.Uint16(t.Data[2:4]) t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2]) t.Seq = binary.BigEndian.Uint32(t.Data[4:8]) t.Ack = binary.BigEndian.Uint32(t.Data[8:12]) @@ -69,6 +69,11 @@ func (t *TCPPacket) ParseBasic() { // String output for a TCP Packet func (t *TCPPacket) String() string { + maxLen := len(t.Data) + if maxLen > 500 { + maxLen = 500 + } + return strings.Join([]string{ "Source port: " + strconv.Itoa(int(t.SrcPort)), "Dest port:" + strconv.Itoa(int(t.DestPort)), @@ -90,7 +95,7 @@ func (t *TCPPacket) String() string { "Checksum:" + strconv.Itoa(int(t.Checksum)), "Data size:" + strconv.Itoa(len(t.Data)), - "Data:" + string(t.Data), + "Data:" + string(t.Data[:maxLen]), }, "\n") } diff --git a/settings.go b/settings.go index d74a4f6..37ac211 100644 --- a/settings.go +++ b/settings.go @@ -3,8 +3,9 @@ package main import ( "flag" "fmt" - "log" "os" + "sync" + "time" ) var VERSION string @@ -42,6 +43,8 @@ type AppSettings struct { inputRAW MultiOption + middleware string + inputHTTP MultiOption outputHTTP MultiOption @@ -79,6 +82,8 @@ func init() { flag.Var(&Settings.inputRAW, "input-raw", "Capture traffic from given port (use RAW sockets and require *sudo* access):\n\t# Capture traffic from 8080 port\n\tgor --input-raw :8080 --output-http staging.com") + flag.StringVar(&Settings.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") flag.Var(&Settings.outputHTTP, "output-http", "Forwards incoming requests to given http address.\n\t# Redirect all incoming requests to staging.com address \n\tgor --input-raw :80 --output-http http://staging.com") @@ -118,10 +123,19 @@ func init() { 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%") } +var previousDebugTime int64 +var debugMutex sync.Mutex + // Debug gets called only if --verbose flag specified func Debug(args ...interface{}) { if Settings.verbose { - fmt.Printf("[DEBUG][PID %d] ", os.Getpid()) - log.Println(args...) + debugMutex.Lock() + now := time.Now() + diff := float64(now.UnixNano()-previousDebugTime) / 1000000 + previousDebugTime = now.UnixNano() + debugMutex.Unlock() + + fmt.Printf("[DEBUG][PID %d][%d][%fms] ", os.Getpid(), now.UnixNano(), diff) + fmt.Println(args...) } } diff --git a/test_input.go b/test_input.go index a9a359a..c699dff 100644 --- a/test_input.go +++ b/test_input.go @@ -3,6 +3,7 @@ package main import ( "crypto/rand" "encoding/base64" + "time" ) // TestInput used for testing purpose, it allows emitting requests on demand @@ -20,9 +21,12 @@ func NewTestInput() (i *TestInput) { func (i *TestInput) Read(data []byte) (int, error) { buf := <-i.data - copy(data, buf) - return len(buf), nil + header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano()) + copy(data[0:len(header)], header) + copy(data[len(header):], buf) + + return len(buf) + len(header), nil } // EmitGET emits GET request without headers