Unify all input/output plugins

README.md

@@ -240,7 +240,7 @@ Next goes request id: unique among all requests (sha1 of time and Ack), but rema
 
 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 optional, and does not affect anything at the moment.
+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.

emitter.go

@@ -1,17 +1,11 @@
 package main
 
 import (
-	"crypto/rand"
 	"io"
 	"time"
+	"bytes"
 )
 
-func uuid() []byte {
-	b := make([]byte, 16)
-	rand.Read(b)
-	return b
-}
-
 // Start initialize loop for sending data from inputs to outputs
 func Start(stop chan int) {
 	if Settings.middleware != "" {
@@ -65,14 +59,21 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 				Debug("[EMITTER] input:", string(payload[0:_maxN]))
 			}
 
-			if modifier != nil {
-				payload = modifier.Rewrite(payload)
+			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(payload) == 0 {
+				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]))
 				}

examples/middleware/echo_modifier.sh

@@ -22,7 +22,6 @@ while read line; do
     case ${header:0:1} in
     "1")
         log "Request type: Request"
-        echo "$encoded"
         ;;
     "2")
         log "Request type: Original Response"
@@ -33,6 +32,8 @@ while read line; do
     *)
         log "Unknown request type $header"
     esac
+    echo "$encoded"
+
     log "==================================="
 
     log "Original data: $line"

examples/middleware/token_modifier.go

@@ -80,21 +80,13 @@ func process(buf []byte) {
 					payload = proto.SetPathParam(payload, []byte("token"), alias)
 
 					// Copy modified payload to our buffer
-					copy(buf[headerSize:], payload)
+					buf = append(buf[:headerSize], payload...)
 				}
 			}
 		}
 
-		// Re-compute length in case if payload was modified
-		bufLen := headerSize + len(payload)
-		// Encoding request and sending it back
-		dst := make([]byte, bufLen*2+1)
-		hex.Encode(dst, buf[:bufLen])
-		dst[len(dst)-1] = '\n'
-
-		os.Stdout.Write(dst)
-
-		return
+		// Emitting data back
+		os.Stdout.Write(encode(buf))
 	case '2': // Original response
 		if _, ok := originalTokens[reqID]; ok {
 			// Token is inside response body
@@ -113,6 +105,14 @@ func process(buf []byte) {
 	}
 }
 
+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...)

input_raw.go

@@ -13,16 +13,14 @@ type RAWInput struct {
 	data            chan *raw.TCPMessage
 	address         string
 	expire          time.Duration
-	captureResponse bool
 }
 
 // NewRAWInput constructor for RAWInput. Accepts address with port as argument.
-func NewRAWInput(address string, expire time.Duration, captureResponse bool) (i *RAWInput) {
+func NewRAWInput(address string, expire time.Duration) (i *RAWInput) {
 	i = new(RAWInput)
 	i.data = make(chan *raw.TCPMessage)
 	i.address = address
 	i.expire = expire
-	i.captureResponse = captureResponse
 
 	go i.listen(address)
 
@@ -33,23 +31,18 @@ func (i *RAWInput) Read(data []byte) (int, error) {
 	msg := <-i.data
 	buf := msg.Bytes()
 
-	if i.captureResponse {
-		var header []byte
+	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
+	if msg.IsIncoming {
+		header = payloadHeader(RequestPayload, msg.UUID(), msg.Start)
 	} else {
-		copy(data, buf)
-		return len(buf), nil
+		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) {
@@ -63,7 +56,7 @@ func (i *RAWInput) listen(address string) {
 		log.Fatal("input-raw: error while parsing address", err)
 	}
 
-	listener := raw.NewListener(host, port, i.expire, i.captureResponse)
+	listener := raw.NewListener(host, port, i.expire, true)
 
 	for {
 		// Receiving TCPMessage object

input_raw_test.go

@@ -24,7 +24,7 @@ func TestRAWInput(t *testing.T) {
 
 	listener := startHTTP(func(w http.ResponseWriter, req *http.Request) {})
 
-	input := NewRAWInput(listener.Addr().String(), testRawExpire, false)
+	input := NewRAWInput(listener.Addr().String(), testRawExpire)
 	output := NewTestOutput(func(data []byte) {
 		wg.Done()
 	})
@@ -41,7 +41,8 @@ func TestRAWInput(t *testing.T) {
 	go Start(quit)
 
 	for i := 0; i < 100; i++ {
-		wg.Add(1)
+		// request + response
+		wg.Add(2)
 		client.Get("/")
 	}
 
@@ -66,14 +67,19 @@ func TestInputRAW100Expect(t *testing.T) {
 
 	originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr, testRawExpire, false)
+	input := NewRAWInput(originAddr, testRawExpire)
 
 	// 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 '1':
+			if strings.Contains(string(data), "Expect: 100-continue") {
+				t.Error("Should not contain 100-continue header")
+			}
+			wg.Done()
+		case '2':
+			wg.Done()
 		}
-		wg.Done()
 	})
 
 	listener := startHTTP(func(w http.ResponseWriter, req *http.Request) {
@@ -96,7 +102,8 @@ func TestInputRAW100Expect(t *testing.T) {
 
 	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 {
@@ -123,7 +130,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 
 	originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr, testRawExpire, false)
+	input := NewRAWInput(originAddr, testRawExpire)
 
 	listener := startHTTP(func(w http.ResponseWriter, req *http.Request) {
 		defer req.Body.Close()
@@ -138,7 +145,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 	})
 	replayAddr := listener.Addr().String()
 
-	httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: false})
+	httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: true})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{httpOutput}
@@ -181,7 +188,7 @@ func TestInputRAWLargePayload(t *testing.T) {
 	}))
 	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr, testRawExpire, false)
+	input := NewRAWInput(originAddr, testRawExpire)
 
 	replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024)
@@ -214,34 +221,3 @@ func TestInputRAWLargePayload(t *testing.T) {
 	wg.Wait()
 	close(quit)
 }
-
-func TestInputRAWResponse(t *testing.T) {
-	wg := new(sync.WaitGroup)
-	quit := make(chan int)
-
-	listener := startHTTP(func(w http.ResponseWriter, req *http.Request) {})
-
-	input := NewRAWInput(listener.Addr().String(), testRawExpire, true)
-	output := NewTestOutput(func(data []byte) {
-		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{})
-
-	time.Sleep(time.Millisecond)
-	go Start(quit)
-
-	for i := 0; i < 100; i++ {
-		// 2 because we track both request and response
-		wg.Add(2)
-		client.Get("/")
-	}
-
-	wg.Wait()
-	close(quit)
-}

middleware.go

@@ -94,10 +94,7 @@ func (m *Middleware) read(from io.Reader) {
 			Debug("[MIDDLEWARE-MASTER] Received:", string(buf))
 		}
 
-		// We should accept only request payloads
-		if buf[0] == '1' {
-			m.data <- buf
-		}
+		m.data <- buf
 	}
 
 	if err := scanner.Err(); err != nil {

middleware_test.go

@@ -110,7 +110,7 @@ func TestEchoMiddleware(t *testing.T) {
 	Settings.middleware = "./examples/middleware/echo_modifier.sh"
 
 	// Catch traffic from one service
-	input := NewRAWInput(from.Listener.Addr().String(), testRawExpire, true)
+	input := NewRAWInput(from.Listener.Addr().String(), testRawExpire)
 
 	// And redirect to another
 	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: false})
@@ -165,7 +165,7 @@ func TestTokenMiddleware(t *testing.T) {
 	quit := make(chan int)
 
 	// Catch traffic from one service
-	input := NewRAWInput(from, testRawExpire, true)
+	input := NewRAWInput(from, testRawExpire)
 
 	// And redirect to another
 	output := NewHTTPOutput(to, &HTTPOutputConfig{Debug: true})

output_file.go

@@ -43,6 +43,10 @@ func (o *FileOutput) init(path string) {
 }
 
 func (o *FileOutput) Write(data []byte) (n int, err error) {
+	if !isOriginPayload(data) {
+		return len(data), nil
+	}
+
 	raw := RawRequest{time.Now().UnixNano(), data}
 
 	o.encoder.Encode(raw)

output_http.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"bytes"
 	"io"
 	"log"
 	"sync/atomic"
@@ -140,6 +139,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)
 
@@ -173,17 +176,11 @@ func (o *HTTPOutput) Read(data []byte) (int, error) {
 }
 
 func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) {
-	var uuid []byte
-
-	if len(Settings.middleware) > 0 {
-		headerSize := bytes.IndexByte(request, '\n')
-		meta := bytes.Split(request[:headerSize], []byte{' '})
-		uuid = meta[1]
-		request = request[headerSize+1:]
-	}
+	meta := payloadMeta(request)
+	uuid := meta[1]
 
 	start := time.Now()
-	resp, err := client.Send(request)
+	resp, err := client.Send(payloadBody(request))
 	stop := time.Now()
 
 	if err != nil {

output_http_test.go

@@ -55,7 +55,7 @@ func TestHTTPOutput(t *testing.T) {
 	methods := HTTPMethods{[]byte("GET"), []byte("PUT"), []byte("POST")}
 	Settings.modifierConfig = HTTPModifierConfig{headers: headers, methods: methods}
 
-	output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{})
+	output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{Debug: true})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}

output_tcp.go

@@ -57,6 +57,10 @@ func (o *TCPOutput) worker() {
 }
 
 func (o *TCPOutput) Write(data []byte) (n int, err error) {
+	if !isOriginPayload(data) {
+		return len(data), nil
+	}
+
 	// Hex encoding always 2x number of bytes
 	encoded := make([]byte, len(data)*2+1)
 	hex.Encode(encoded, data)

protocol.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	"strconv"
+	"bytes"
+	"encoding/hex"
+	"crypto/rand"
 )
 
 const (
@@ -10,6 +13,17 @@ const (
 	ReplayedResponsePayload
 )
 
+func uuid() []byte {
+	b := make([]byte, 20)
+	rand.Read(b)
+
+	uuid := make([]byte, 40)
+	hex.Encode(uuid, b)
+
+	return uuid
+}
+
+
 // Timing is request start or round-trip time, depending on payloadType
 func payloadHeader(payloadType int, uuid []byte, timing int64) (header []byte) {
 	sTime := strconv.FormatInt(timing, 10)
@@ -36,3 +50,26 @@ func payloadHeader(payloadType int, uuid []byte, timing int64) (header []byte) {
 
 	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 '1', '2':
+		return true
+	default:
+		return false
+	}
+}
+
+func isRequestPayload(payload []byte) bool {
+	return payload[0] == '1'
+}

raw_socket_listener/listener.go

@@ -91,10 +91,15 @@ func (t *Listener) listen() {
 		select {
 		// 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)
@@ -216,11 +221,12 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 	if t.captureResponse && isIncoming {
 		// If message have multiple packets, delete previous alias
 		if len(message.packets) > 0 {
-			delete(t.respAliases, uint32(message.Size()))
+			delete(t.respAliases, message.ResponseAck)
 		}
 
-		responseAck := packet.Seq + uint32(message.Size()+len(packet.Data))
+		responseAck := packet.Seq + uint32(len(packet.Data))
 		t.respAliases[responseAck] = &request{message.Start, message.Ack}
+		message.ResponseAck = responseAck
 	}
 
 	// Adding packet to message

raw_socket_listener/tcp_message.go

@@ -20,6 +20,7 @@ import (
 type TCPMessage struct {
 	ID           string // Message ID
 	Ack          uint32
+	ResponseAck  uint32
 	RequestStart int64
 	RequestAck   uint32
 	Start        int64
@@ -80,7 +81,11 @@ func (t *TCPMessage) Timeout() {
 		}
 	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
+		}
 	}
 }
 

raw_socket_listener/tcp_packet.go

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

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