Working modifier implementation with examples

Dockerfile

@@ -2,7 +2,7 @@ FROM google/golang
 
 RUN cd /goroot/src/ && GOOS=linux GOARCH=386 ./make.bash --no-clean
 
-RUN apt-get install ruby -y
+RUN apt-get update && apt-get install ruby vim-common -y
 
 WORKDIR /gopath/src/gor
 

Makefile

@@ -22,7 +22,7 @@ dbench:
 
 # Used mainly for debugging, because docker container do not have access to parent machine ports
 drun:
-	docker run -v `pwd`:/gopath/src/gor -t -i gor go run $(SOURCE) --input-modifier="./pong_modifier.rb" --input-dummy=0 --input-http=:9000 --output-http="http://localhost:9000"  --verbose
+	docker run -v `pwd`:/gopath/src/gor -t -i gor go run $(SOURCE) --input-modifier="./examples/echo_modifier.sh" --input-dummy=0 --output-http="http://localhost:9000"  --verbose
 
 dbash:
 	docker run -v `pwd`:/gopath/src/gor -t -i gor /bin/bash

emitter.go

@@ -26,6 +26,7 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 
 	for {
 		nr, er := src.Read(buf)
+
 		if nr > 0 && len(buf) > nr {
 			Debug("Sending", src, ": ", string(buf[0:nr]))
 

examples/echo_modifier.rb

@@ -0,0 +1,16 @@
+#!/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] Original data: #{data}"
+    STDERR.puts "[DEBUG] Decoded request: #{decoded}"
+    STDERR.puts "[DEBUG] Encoded data: #{encoded}"
+end

examples/echo_modifier.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+while read line; do
+    decoded=$(echo "$line" | xxd -r -p)
+    encoded=$(echo "$decoded" | xxd -p | tr -d "\\n")
+    echo "$encoded"
+
+    >&2 echo "[DEBUG] Original data: $line"
+    >&2 echo "[DEBUG] Decoded request: $decoded"
+    >&2 echo "[DEBUG] Encoded data: $encoded"
+done;

examples/echo_modifier/echo_modifier.go

@@ -1,44 +0,0 @@
-package main
-
-import (
-    "os"
-    "bufio"
-)
-
-func main() {
-    reader := bufio.NewReader(os.Stdin)
-    data := make(chan []byte)
-
-    go ReadStdin(data)
-
-    for {
-        os.Stdout.Print(<- data, '¶')
-    }
-}
-
-func ReadStdin(data chan []byte){
-    for {
-        buf, err := reader.ReadBytes('¶')
-        buf_len := len(buf)
-        if buf_len > 0 {
-            new_buf_len := len(buf) - 2
-            if new_buf_len > 0 {
-                new_buf := make([]byte, new_buf_len)
-                copy(new_buf, buf[:new_buf_len])
-                data <- new_buf
-                if err != nil {
-                    if err != io.EOF {
-                        log.Printf("error: %s\n", err)
-                    }
-                }
-            }
-        }
-    }
-}
-
-while data = STDIN.gets(separator)
-  STDERR.puts "==== Start ===="
-  STDERR.puts data
-  puts data
-  STDERR.puts "==== End ===="
-end

input_dummy.go

@@ -30,7 +30,7 @@ func (i *DummyInput) emit() {
 	for {
 		select {
 		case <-ticker.C:
-			i.data <- []byte("GET / HTTP/1.1\r\n\r\n")
+			i.data <- []byte("POST /pub/WWW/å HTTP/1.1\nHost: www.w3.org\r\n\r\na=1&b=2")
 		}
 	}
 }

plugins.go

@@ -62,6 +62,7 @@ func registerPlugin(constructor interface{}, options ...interface{}) {
 		Plugins.Inputs = append(Plugins.Inputs, plugin_wrapper.(io.Reader))
 	}
 
+
 	if _, ok := plugin.(io.Writer); ok {
 		Plugins.Outputs = append(Plugins.Outputs, plugin_wrapper.(io.Writer))
 	}

pong_modifier.rb

@@ -1,13 +0,0 @@
-#!/usr/bin/env ruby
-# encoding: utf-8
-require "base64"
-
-STDERR.puts "Starting modifier"
-puts "Starting modifier"
-
-while data = STDIN.gets.chomp
-  STDERR.puts "==== Start ===="
-  STDERR.puts Base64.encode64(data)
-  puts data
-  STDERR.puts "==== End ===="
-end

traffic_modifier.go

@@ -6,7 +6,8 @@ import (
     "io"
     "os/exec"
     "os"
-    "encoding/base64"
+    "bufio"
+    "encoding/hex"
 )
 
 type TrafficModifier struct {
@@ -23,43 +24,74 @@ func NewTrafficModifier(plugin interface{}, command string) io.Reader {
     m := new(TrafficModifier)
     m.plugin = plugin
     m.command = command
+    m.data = make(chan []byte)
 
-    cmd := exec.Command("bash", "-c", command)
-    cmd.Stderr = os.Stderr
+    cmd := exec.Command(command)
 
     m.Stdout, _ = cmd.StdoutPipe()
     m.Stdin, _ = cmd.StdinPipe()
-
-    m.Stdout = base64.NewDecoder(base64.StdEncoding, m.Stdout)
+    cmd.Stderr = os.Stderr
 
     go m.copy(m.Stdin, m.plugin.(io.Reader))
+    go m.read(m.Stdout)
 
-    err := cmd.Run()
+    go func(){
+        err := cmd.Start()
 
-    if (err != nil) {
-        log.Fatal(err)
-    }
+        if (err != nil) {
+            log.Fatal(err)
+        }
+    }()
+
+    defer cmd.Wait()
 
     return m
 }
 
 func (m *TrafficModifier) copy(to io.Writer, from io.Reader) {
     buf := make([]byte, 5*1024*1024)
+    dst := make([]byte, len(buf)*2)
 
     for {
-        nr, er := from.Read(buf)
+        nr, _ := from.Read(buf)
         if nr > 0 && len(buf) > nr {
-            to.Write(base64.StdEncoding.Encode(buf))
+            hex.Encode(dst, buf[0:nr])
+            to.Write(dst[0:nr*2])
+            to.Write([]byte("\r\n"))
         }
     }
 }
 
-func (m *TrafficModifier) Read(data []byte) (n int, err error) {
-    n, err = m.Stdout.Read(data)
+func (m *TrafficModifier) read(from io.Reader) {
+    buf := make([]byte, 5*1024*1024)
+
+    scanner := bufio.NewScanner(from)
+
+    for scanner.Scan() {
+        bytes := scanner.Bytes()
+        hex.Decode(buf, bytes)
+
+        Debug("Received:", buf[0:len(bytes)/2])
+
+        m.data <- buf[0:len(bytes)/2]
+    }
+
+    if err := scanner.Err(); err != nil {
+        fmt.Fprintln(os.Stderr, "Traffic modifier command failed:", err)
+    }
 
     return
 }
 
+func (m *TrafficModifier) Read(data []byte) (int, error) {
+    Debug("Trying to read channel!")
+    buf := <- m.data
+    copy(data, buf)
+
+    return len(buf), nil
+}
+
+
 func (m *TrafficModifier) String() string {
     return fmt.Sprintf("Modifying traffic for %s using '%s' command", m.plugin, m.command)
 }