Use latest libpcap and ensure to close sockets

.dockerignore

@@ -1 +1,3 @@
 *.tar.gz
+gor
+gor.test

Dockerfile

@@ -10,7 +10,8 @@ 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 apt-get install libpcap-dev -y
+RUN apt-get install flex bison -y
+RUN wget http://www.tcpdump.org/release/libpcap-1.7.4.tar.gz && tar xzf libpcap-1.7.4.tar.gz && cd libpcap-1.7.4 && ./configure && make install
 RUN go get github.com/google/gopacket
 RUN go get -u github.com/golang/lint/golint
 

Makefile

@@ -4,14 +4,15 @@ RUN = docker run -v `pwd`:$(SOURCE_PATH) -p 0.0.0.0:8000:8000 -t -i gor
 BENCHMARK = BenchmarkRAWInput
 TEST = TestRawListenerBench
 VERSION = DEV-$(shell date +%s)
+LDFLAGS = -ldflags "-X main.VERSION=$(VERSION) -extldflags \"-static\""
 
 release: release-x64
 
 release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i gor go build -ldflags "-X main.VERSION=$(VERSION) -extldflags \"-static\"" && tar -czf gor_$(VERSION)_x64.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i gor go build $(LDFLAGS) && 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 -i gor go build -ldflags "-X main.VERSION=$(VERSION)" && tar -czf gor_$(VERSION)_x86.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i gor go build $(LDFLAGS) && tar -czf gor_$(VERSION)_x86.tar.gz gor && rm gor
 
 build:
 	docker build -t gor .
@@ -27,13 +28,13 @@ race:
 	$(RUN) go test ./... $(ARGS) -v -race -timeout 15s
 
 test:
-	$(RUN) go test ./. -timeout 30s $(ARGS) -v
+	$(RUN) go test ./. -race -timeout 30s $(LDFLAGS) $(ARGS)  -v
 
 test_all:
-	$(RUN) go test ./... -timeout 60s $(ARGS) -v
+	$(RUN) go test ./... -timeout 60s $(LDFLAGS) $(ARGS) -v
 
 testone:
-	$(RUN) go test ./... -timeout 4s -run $(TEST) $(ARGS) -v
+	$(RUN) go test ./... -timeout 4s $(LDFLAGS) -run $(TEST) $(ARGS) -v
 
 cover:
 	$(RUN) go test $(ARGS) -race -v -timeout 15s -coverprofile=coverage.out
@@ -46,7 +47,7 @@ vet:
 	$(RUN) go vet
 
 bench:
-	$(RUN) go test -v -run NOT_EXISTING -bench $(BENCHMARK) -benchtime 5s
+	$(RUN) go test $(LDFLAGS) -v -run NOT_EXISTING -bench $(BENCHMARK) -benchtime 5s
 
 profile_test:
 	$(RUN) go test $(LDFLAGS) -run $(TEST) ./raw_socket_listener/. $(ARGS) -memprofile mem.mprof -cpuprofile cpu.out
@@ -54,7 +55,7 @@ profile_test:
 
 # Used mainly for debugging, because docker container do not have access to parent machine ports
 run:
-	$(RUN) go run $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw 127.0.0.1:9000 --input-http 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh"
+	$(RUN) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw 127.0.0.1:9000 --input-http 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh"
 
 run-2:
 	$(RUN) go run $(SOURCE) --input-file ./fixtures/requests.gor --output-dummy=0

emitter.go

@@ -32,8 +32,15 @@ func Start(stop chan int) {
 	for {
 		select {
 		case <-stop:
+			pluginMu.Lock()
+			for _, p := range Plugins.All {
+				if cp, ok := p.(io.Closer); ok {
+					cp.Close()
+				}
+			}
+			pluginMu.Unlock()
 			return
-		case <-time.After(time.Second):
+		case <-time.After(100 * time.Millisecond):
 		}
 	}
 }

gor.go

@@ -7,6 +7,9 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"os/signal"
+	"syscall"
+	"io"
 	"runtime"
 	_ "runtime/debug"
 	"runtime/pprof"
@@ -50,6 +53,20 @@ func main() {
 	}
 
 	Start(nil)
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+	go func(){
+	    <- c
+
+	    for _, p := range Plugins.All {
+			if cp, ok := p.(io.Closer); ok {
+				cp.Close()
+			}
+		}
+
+	    os.Exit(1)
+	}()
 }
 
 func profileCPU(cpuprofile string) {

http_client.go

@@ -12,8 +12,11 @@ import (
 	"strings"
 	"syscall"
 	"time"
+	"sync"
 )
 
+var httpMu sync.Mutex
+
 var defaultPorts = map[string]string{
 	"http":  "80",
 	"https": "443",
@@ -176,7 +179,8 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) {
 		return
 	}
 
-	payload := c.respBuf[:n]
+	payload := make([]byte, n)
+	copy(payload, c.respBuf[:n])
 
 	if c.config.Debug {
 		Debug("[HTTPClient] Received:", string(payload))

input_raw.go

@@ -32,11 +32,7 @@ func NewRAWInput(address string, engine int, expire time.Duration) (i *RAWInput)
 	i.engine = engine
 	i.quit = make(chan bool)
 
-	go i.listen(address)
+	i.listen(address)
-
-	for i.listener == nil {
-		time.Sleep(time.Millisecond)
-	}
 	i.listener.IsReady()
 
 	return
@@ -73,18 +69,20 @@ func (i *RAWInput) listen(address string) {
 
 	ch := i.listener.Receiver()
 
-	for {
+	go func(){
-		select {
+		for {
-		case <-i.quit:
+			select {
-			return
+			case <-i.quit:
-		default:
+				return
+			default:
+			}
+
+			// Receiving TCPMessage object
+			m := <-ch
+
+			i.data <- m
 		}
-
+	}()
-		// Receiving TCPMessage object
-		m := <-ch
-
-		i.data <- m
-	}
 }
 
 func (i *RAWInput) String() string {

plugins.go

@@ -5,14 +5,18 @@ import (
 	"reflect"
 	"strings"
 	"time"
+	"sync"
 )
 
 // InOutPlugins struct for holding references to plugins
 type InOutPlugins struct {
 	Inputs  []io.Reader
 	Outputs []io.Writer
+	All []interface{}
 }
 
+
+var pluginMu sync.Mutex
 // Plugins holds all the plugin objects
 var Plugins *InOutPlugins = new(InOutPlugins)
 
@@ -67,10 +71,15 @@ func registerPlugin(constructor interface{}, options ...interface{}) {
 	if isW {
 		Plugins.Outputs = append(Plugins.Outputs, pluginWrapper.(io.Writer))
 	}
+
+	Plugins.All = append(Plugins.All, plugin)
 }
 
 // InitPlugins specify and initialize all available plugins
 func InitPlugins() {
+	pluginMu.Lock()
+	defer pluginMu.Unlock()
+
 	for _, options := range Settings.inputDummy {
 		registerPlugin(NewDummyInput, options)
 	}

raw_socket_listener/listener.go

@@ -62,6 +62,8 @@ type Listener struct {
 	messageExpire time.Duration
 
 	conn    net.PacketConn
+	pcapHandles []*pcap.Handle
+
 	quit    chan bool
 	readyCh chan bool
 }
@@ -279,7 +281,11 @@ func (t *Listener) readPcap() {
 			}
 			defer handle.Close()
 
-			bpf := "port " + strconv.Itoa(int(t.port))
+			t.mu.Lock()
+			t.pcapHandles = append(t.pcapHandles, handle)
+			t.mu.Unlock()
+
+			bpf := "tcp port " + strconv.Itoa(int(t.port))
 
 			// log.Println("Applying bpf programm:", bpf, " Device:", device.Name)
 			if err := handle.SetBPFFilter(bpf); err != nil {
@@ -287,6 +293,7 @@ func (t *Listener) readPcap() {
 				wg.Done()
 				return
 			}
+			// log.Println("BPF appplied", device.Name)
 
 			source := gopacket.NewPacketSource(handle, handle.LinkType())
 			source.Lazy = true
@@ -574,5 +581,10 @@ func (t *Listener) Close() {
 	if t.conn != nil {
 		t.conn.Close()
 	}
+
+	for _, h := range t.pcapHandles {
+		h.Close()
+	}
+
 	return
 }

raw_socket_listener/listener_test.go

@@ -20,8 +20,8 @@ func TestRawListenerInput(t *testing.T) {
 	respAck := reqPacket.Seq + uint32(len(reqPacket.Data))
 	respPacket := buildPacket(false, respAck, reqPacket.Seq+1, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
-	listener.processTCPPacket(reqPacket)
+	listener.packetsChan <- reqPacket.Dump()
-	listener.processTCPPacket(respPacket)
+	listener.packetsChan <- respPacket.Dump()
 
 	select {
 	case req = <-listener.messagesChan:
@@ -56,8 +56,8 @@ func TestRawListenerResponse(t *testing.T) {
 	respPacket := buildPacket(false, 1+uint32(len(reqPacket.Data)), 2, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
 	// If response packet comes before request
-	listener.processTCPPacket(respPacket)
+	listener.packetsChan <- respPacket.Dump()
-	listener.processTCPPacket(reqPacket)
+	listener.packetsChan <- reqPacket.Dump()
 
 	select {
 	case req = <-listener.messagesChan:
@@ -102,12 +102,12 @@ func TestRawListener100Continue(t *testing.T) {
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
 	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
 
-	listener.processTCPPacket(reqPacket1)
+	listener.packetsChan <- reqPacket1.Dump()
-	listener.processTCPPacket(reqPacket2)
+	listener.packetsChan <- reqPacket2.Dump()
-	listener.processTCPPacket(reqPacket3)
+	listener.packetsChan <- reqPacket3.Dump()
 
-	listener.processTCPPacket(respPacket1)
+	listener.packetsChan <- respPacket1.Dump()
-	listener.processTCPPacket(respPacket2)
+	listener.packetsChan <- respPacket2.Dump()
 
 	select {
 	case req = <-listener.messagesChan:
@@ -159,12 +159,12 @@ func TestRawListener100ContinueWrongOrder(t *testing.T) {
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
 	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
 
-	listener.processTCPPacket(respPacket1)
+	listener.packetsChan <- respPacket1.Dump()
-	listener.processTCPPacket(respPacket2)
+	listener.packetsChan <- respPacket2.Dump()
 
-	listener.processTCPPacket(reqPacket1)
+	listener.packetsChan <- reqPacket1.Dump()
-	listener.processTCPPacket(reqPacket2)
+	listener.packetsChan <- reqPacket2.Dump()
-	listener.processTCPPacket(reqPacket3)
+	listener.packetsChan <- reqPacket3.Dump()
 
 	select {
 	case req = <-listener.messagesChan:
@@ -386,7 +386,7 @@ func TestRawListenerBench(t *testing.T) {
 
 	// Should re-construct message from all possible combinations
 	for i := 0; i < 1000; i++ {
-		go func() {
+		go func(i int) {
 			for j := 0; j < 100; j++ {
 				var packets []*TCPPacket
 
@@ -412,7 +412,7 @@ func TestRawListenerBench(t *testing.T) {
 
 				time.Sleep(5 * time.Millisecond)
 			}
-		}()
+		}(i)
 	}
 
 	ch := l.Receiver()

raw_socket_listener/tcp_packet.go

@@ -74,16 +74,18 @@ func (t *TCPPacket) ParseBasic() {
 
 func (t *TCPPacket) Dump() []byte {
 	buf := make([]byte, len(t.Data)+16+16)
+	copy(buf[:16], t.Addr)
+
 	tcpBuf := buf[16:]
 
 	binary.BigEndian.PutUint16(tcpBuf[2:4], t.DestPort)
-	binary.BigEndian.PutUint16(buf[0:2], t.SrcPort)
+	binary.BigEndian.PutUint16(tcpBuf[0:2], t.SrcPort)
 
-	binary.BigEndian.PutUint32(buf[4:8], t.Seq)
+	binary.BigEndian.PutUint32(tcpBuf[4:8], t.Seq)
-	binary.BigEndian.PutUint32(buf[8:12], t.Ack)
+	binary.BigEndian.PutUint32(tcpBuf[8:12], t.Ack)
 
-	tcpBuf[16] = 64
+	tcpBuf[12] = 64
-	copy(tcpBuf[20:], t.Data)
+	copy(tcpBuf[16:], t.Data)
 
 	return buf
 }