From 19bfb8211cc53330ab8a14335b53707459ce0e54 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Fri, 13 May 2016 14:33:20 +0500 Subject: [PATCH] Use latest libpcap and ensure to close sockets --- .dockerignore | 2 ++ Dockerfile | 3 ++- Makefile | 15 +++++++------ emitter.go | 9 +++++++- gor.go | 17 +++++++++++++++ http_client.go | 6 +++++- input_raw.go | 30 ++++++++++++-------------- plugins.go | 9 ++++++++ raw_socket_listener/listener.go | 14 +++++++++++- raw_socket_listener/listener_test.go | 32 ++++++++++++++-------------- raw_socket_listener/tcp_packet.go | 12 ++++++----- 11 files changed, 101 insertions(+), 48 deletions(-) diff --git a/.dockerignore b/.dockerignore index 335ec95..b93f1d9 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1 +1,3 @@ *.tar.gz +gor +gor.test diff --git a/Dockerfile b/Dockerfile index da6eebd..98fb7a4 100644 --- a/Dockerfile +++ b/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 diff --git a/Makefile b/Makefile index adb6f9b..32fe0ba 100644 --- a/Makefile +++ b/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 diff --git a/emitter.go b/emitter.go index d39320c..9575484 100644 --- a/emitter.go +++ b/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): } } } diff --git a/gor.go b/gor.go index 61abfb3..9af6e93 100644 --- a/gor.go +++ b/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) { diff --git a/http_client.go b/http_client.go index 074968d..6755ae0 100644 --- a/http_client.go +++ b/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)) diff --git a/input_raw.go b/input_raw.go index a88a203..9ff00ab 100644 --- a/input_raw.go +++ b/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) - - for i.listener == nil { - time.Sleep(time.Millisecond) - } + i.listen(address) i.listener.IsReady() return @@ -73,18 +69,20 @@ func (i *RAWInput) listen(address string) { ch := i.listener.Receiver() - for { - select { - case <-i.quit: - return - default: + go func(){ + for { + select { + case <-i.quit: + return + default: + } + + // Receiving TCPMessage object + m := <-ch + + i.data <- m } - - // Receiving TCPMessage object - m := <-ch - - i.data <- m - } + }() } func (i *RAWInput) String() string { diff --git a/plugins.go b/plugins.go index 8379cf5..754e5ec 100644 --- a/plugins.go +++ b/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) } diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index e3db57f..4e700df 100644 --- a/raw_socket_listener/listener.go +++ b/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 } diff --git a/raw_socket_listener/listener_test.go b/raw_socket_listener/listener_test.go index 80c0977..5882f94 100644 --- a/raw_socket_listener/listener_test.go +++ b/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.processTCPPacket(respPacket) + listener.packetsChan <- reqPacket.Dump() + 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.processTCPPacket(reqPacket) + listener.packetsChan <- respPacket.Dump() + 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.processTCPPacket(reqPacket2) - listener.processTCPPacket(reqPacket3) + listener.packetsChan <- reqPacket1.Dump() + listener.packetsChan <- reqPacket2.Dump() + listener.packetsChan <- reqPacket3.Dump() - listener.processTCPPacket(respPacket1) - listener.processTCPPacket(respPacket2) + listener.packetsChan <- respPacket1.Dump() + 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.processTCPPacket(respPacket2) + listener.packetsChan <- respPacket1.Dump() + listener.packetsChan <- respPacket2.Dump() - listener.processTCPPacket(reqPacket1) - listener.processTCPPacket(reqPacket2) - listener.processTCPPacket(reqPacket3) + listener.packetsChan <- reqPacket1.Dump() + listener.packetsChan <- reqPacket2.Dump() + 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() diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index fdfe515..3631dc6 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/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(buf[8:12], t.Ack) + binary.BigEndian.PutUint32(tcpBuf[4:8], t.Seq) + binary.BigEndian.PutUint32(tcpBuf[8:12], t.Ack) - tcpBuf[16] = 64 - copy(tcpBuf[20:], t.Data) + tcpBuf[12] = 64 + copy(tcpBuf[16:], t.Data) return buf }