Merge pull request #162 from buger/input-modifier

Middleware for custom request rewrite logic

.gitignore

@@ -4,3 +4,7 @@
 *.out
 
 *.bin
+
+*.gz
+
+*.class

.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"
+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"

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

Makefile

@@ -1,14 +1,14 @@
-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/
 
 release: release-x86 release-x64
 
 release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 --env CGO_ENABLED=0 -i gor go build && tar -czf gor_x64.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 --env CGO_ENABLED=0 -i gor go build -ldflags "-X main.VERSION $(VERSION)"&& tar -czf gor_$(VERSION)_x64.tar.gz gor && rm gor
 
 release-x86:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 --env CGO_ENABLED=0 -i gor go build && tar -czf gor_x86.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 --env CGO_ENABLED=0 -i gor go build -ldflags "-X main.VERSION $(VERSION)" && tar -czf gor_$(VERSION)_x86.tar.gz gor && rm gor
 
 dbuild:
 	docker build -t 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
+	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor /bin/bash

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:
 ```

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 {

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);
+        }
+    }
+}

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

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;

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...)
+}

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()

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 {
@@ -42,7 +43,9 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient {
 
 	u, _ := url.Parse(baseURL)
 	if !strings.Contains(u.Host, ":") {
-		u.Host += ":" + defaultPorts[u.Scheme]
+		if u.Scheme != "http" {
+			u.Host += ":" + defaultPorts[u.Scheme]
+		}
 	}
 
 	if config.Timeout.Nanoseconds() == 0 {
@@ -50,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)
@@ -66,7 +69,11 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient {
 func (c *HTTPClient) Connect() (err error) {
 	c.Disconnect()
 
-	c.conn, err = net.Dial("tcp", c.host)
+	if !strings.Contains(c.host, ":") {
+		c.conn, err = net.DialTimeout("tcp", c.host+":80", c.config.ConnectionTimeout)
+	} else {
+		c.conn, err = net.DialTimeout("tcp", c.host, c.config.ConnectionTimeout)
+	}
 
 	if c.scheme == "https" {
 		tlsConn := tls.Client(c.conn, &tls.Config{InsecureSkipVerify: true})
@@ -85,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)
 	}
 }
 
@@ -118,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
 		}
 	}
@@ -136,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
 	}
 
@@ -154,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))
 	}
@@ -193,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
+}

http_client_test.go

@@ -3,20 +3,24 @@ 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"
 )
 
 func TestHTTPClientURLPort(t *testing.T) {
 	c1 := NewHTTPClient("http://example.com", &HTTPClientConfig{})
-	if c1.baseURL != "http://example.com:80" {
-		t.Error("Sould add 80 port for http:", c1.baseURL)
+	if c1.baseURL != "http://example.com" {
+		t.Error("Sould not add 80 port for http:", c1.baseURL)
 	}
 
 	c2 := NewHTTPClient("https://example.com", &HTTPClientConfig{})
@@ -30,8 +34,8 @@ func TestHTTPClientURLPort(t *testing.T) {
 	}
 
 	c4 := NewHTTPClient("example.com", &HTTPClientConfig{})
-	if c4.baseURL != "http://example.com:80" {
-		t.Error("Sould add default protocol:", c4.baseURL)
+	if c4.baseURL != "http://example.com" {
+		t.Error("Sould not add default protocol:", c4.baseURL)
 	}
 }
 
@@ -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")
+	}
+}

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)
 

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")...)
 		}
 	}
 }

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")
 }

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) {

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)
+}

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)

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 {

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+1)
-		hex.Encode(encoded, msg)
-		conn.Write(append(encoded, '\n'))
+		conn.Write(msg)
+		conn.Write([]byte(payloadSeparator))
 	}
 
 	wg.Wait()

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)
+}

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 = ""
+}

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
 }

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)
 	}

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}

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))

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())
 				}
 			}()
 		}

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 {

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
+}

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
+}

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
+}

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
 }

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")
 }
 

settings.go

@@ -3,14 +3,12 @@ package main
 import (
 	"flag"
 	"fmt"
-	"log"
 	"os"
+	"sync"
+	"time"
 )
 
-const (
-	// VERSION specifies Gor current version
-	VERSION = "0.9.8"
-)
+var VERSION string
 
 // MultiOption allows to specify multiple flags with same name and collects all values into array
 type MultiOption []string
@@ -45,6 +43,8 @@ type AppSettings struct {
 
 	inputRAW MultiOption
 
+	middleware string
+
 	inputHTTP  MultiOption
 	outputHTTP MultiOption
 
@@ -82,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")
@@ -121,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...)
 	}
 }

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