Fixed issues:

    - #75 List endpoints by group / status in /endpoints
    - #74 Implement API endpoint to update endpoints fields
    - #73 List of ever loaded modules in report
    - #72 Track list of loaded modules
    - #61 Integrate with ETW

Author: qjerome

api/adminapi_test.go

@@ -2,15 +2,19 @@ package api
 
 import (
 	"bytes"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"math/rand"
 	"net/http"
 	"net/url"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/0xrawsec/golang-evtx/evtx"
+	"github.com/gorilla/websocket"
 )
 
 func doRequest(method, url string) (r AdminAPIResponse) {
@@ -497,3 +501,99 @@ func TestAdminAPIGetEndpointAlerts(t *testing.T) {
 		t.FailNow()
 	}
 }
+
+func TestEventStream(t *testing.T) {
+	// cleanup previous data
+	clean(&mconf, &fconf)
+
+	m, mc := prepareTest()
+	defer func() {
+		m.Shutdown()
+		m.Wait()
+	}()
+
+	expctd := float64(20000)
+	total := float64(0)
+	sumEps := float64(0)
+	nclients := float64(4)
+	slowClients := float64(0)
+	wg := sync.WaitGroup{}
+
+	for i := float64(0); i < nclients; i++ {
+		u := url.URL{Scheme: "wss", Host: format("localhost:%d", 8001), Path: AdmAPIStreamEvents}
+		key := mconf.AdminAPI.Users[0].Key
+		dialer := *websocket.DefaultDialer
+		dialer.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+		t.Logf("connecting to %s", u.String())
+		c, resp, err := dialer.Dial(u.String(), http.Header{"Api-Key": {key}})
+		if err != nil {
+			if err == websocket.ErrBadHandshake {
+				t.Logf("handshake failed with status %d", resp.StatusCode)
+			}
+			t.Errorf("failed to dial: %s", err)
+			t.FailNow()
+		}
+		defer c.Close()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			recvd := float64(0)
+			start := time.Now()
+			slow := false
+
+			if rand.Int()%2 == 0 {
+				slow = true
+				slowClients++
+			}
+
+			for {
+				_, _, err := c.ReadMessage()
+				if err != nil {
+					break
+				}
+				recvd++
+				if recvd == expctd {
+					break
+				}
+				// simulates a slow client
+				if slow {
+					time.Sleep(35 * time.Microsecond)
+				}
+			}
+			eps := recvd / float64(time.Since(start).Seconds())
+			total += recvd
+			// we take into account only normal clients
+			if !slow {
+				sumEps += eps
+				t.Logf("Normal client received %.1f EPS", eps)
+			} else {
+				t.Logf("Slow client received %.1f EPS", eps)
+			}
+		}()
+	}
+
+	mc.PostLogs(readerFromEvents(int(expctd)))
+	tick := time.NewTicker(60 * time.Second)
+loop:
+	for {
+		select {
+		case <-tick.C:
+			break loop
+		default:
+		}
+
+		if total == expctd*nclients {
+			wg.Wait()
+			break
+		}
+	}
+
+	if total != expctd*nclients {
+		t.Errorf("Received less events than expected received=%.0f VS expected=%.0f", total, expctd*nclients)
+		t.FailNow()
+	}
+
+	t.Logf("Average %.1f EPS/client", sumEps/(nclients-slowClients))
+
+}

api/api_client.go

@@ -50,7 +50,6 @@ func (cc *ClientConfig) ManagerIP() net.IP {
 }
 
 func (cc *ClientConfig) DialContext(ctx context.Context, network, addr string) (con net.Conn, err error) {
-	log.Infof("Dial")
 	dialer := net.Dialer{
 		Timeout:   30 * time.Second,
 		KeepAlive: 30 * time.Second,
@@ -68,8 +67,6 @@ func (cc *ClientConfig) DialContext(ctx context.Context, network, addr string) (
 }
 
 func (cc *ClientConfig) DialTLSContext(ctx context.Context, network, addr string) (net.Conn, error) {
-
-	log.Infof("Dial TLS")
 	c, err := tls.Dial(network, addr, &tls.Config{InsecureSkipVerify: cc.Unsafe})
 
 	if err != nil {

api/endpoint.go

@@ -22,7 +22,7 @@ type Endpoint struct {
 	Group          string    `json:"group"`
 	Key            string    `json:"key,omitempty"`
 	Command        *Command  `json:"command,omitempty"`
-	Score          int       `json:"score"`
+	Score          float64   `json:"score"`
 	Status         string    `json:"status"`
 	LastDetection  time.Time `json:"last-detection"`
 	LastConnection time.Time `json:"last-connection"`

api/forwarder.go

@@ -10,7 +10,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/0xrawsec/golang-evtx/evtx"
 	"github.com/0xrawsec/golang-utils/fileutils"
 	"github.com/0xrawsec/golang-utils/fsutil"
 	"github.com/0xrawsec/golang-utils/fsutil/fswalker"
@@ -65,9 +64,10 @@ func NewForwarder(c *ForwarderConfig) (*Forwarder, error) {
 	// Initialize the Forwarder
 	// TODO: better organize forwarder configuration
 	co := Forwarder{
-		fwdConfig:  c,
-		TimeTresh:  time.Second * 10,
-		EventTresh: 50,
+		fwdConfig: c,
+		TimeTresh: time.Second * 10,
+		// Writing events too quickly has a perf impact
+		EventTresh: 500,
 		Pipe:       new(bytes.Buffer),
 		stop:       make(chan bool),
 		done:       make(chan bool),
@@ -122,10 +122,10 @@ func (f *Forwarder) ArchiveLogs() {
 }
 
 // PipeEvent pipes an event to be sent through the forwarder
-func (f *Forwarder) PipeEvent(e *evtx.GoEvtxMap) {
+func (f *Forwarder) PipeEvent(event interface{}) {
 	f.Lock()
 	defer f.Unlock()
-	f.Pipe.Write(evtx.ToJSON(e))
+	f.Pipe.Write(utils.Json(event))
 	f.Pipe.WriteByte('\n')
 	f.EventsPiped++
 }

api/forwarder_test.go

@@ -1,20 +1,25 @@
 package api
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math/rand"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
-	"github.com/0xrawsec/golang-evtx/evtx"
 	"github.com/0xrawsec/golang-utils/log"
+	"github.com/0xrawsec/golang-utils/readers"
 	"github.com/0xrawsec/golang-utils/scanner"
 	"github.com/0xrawsec/golang-utils/sync/semaphore"
+	"github.com/0xrawsec/whids/event"
+	"github.com/0xrawsec/whids/utils"
 )
 
 var (
@@ -50,35 +55,44 @@ var (
 		},
 	}
 
-	events = []string{
-		// regular log
-		`{"Event":{"EventData":{"EventType":"CreateKey","Image":"C:\\Windows\\servicing\\TrustedInstaller.exe","ProcessGuid":"{49F1AF32-38C1-5AC7-0000-00105E5D0B00}","ProcessId":"2544","TargetObject":"HKLM\\SOFTWARE\\Microsoft\\EnterpriseCertificates\\Disallowed","UtcTime":"2018-04-06 20:07:14.423"},"System":{"Channel":"Microsoft-Windows-Sysmon/Operational","Computer":"CALDERA02.caldera.loc","Correlation":{},"EventID":"12","EventRecordID":"886970","Execution":{"ProcessID":"1456","ThreadID":"1712"},"Keywords":"0x8000000000000000","Level":"4","Opcode":"0","Provider":{"Guid":"{5770385F-C22A-43E0-BF4C-06F5698FFBD9}","Name":"Microsoft-Windows-Sysmon"},"Security":{"UserID":"S-1-5-18"},"Task":"12","TimeCreated":{"SystemTime":"2018-04-06T09:07:14.424360200Z"},"Version":"2"}}}`,
-		// alert log
-		`{"Event":{"EventData":{"CreationUtcTime":"2018-02-26 16:28:13.169","Image":"C:\\Program Files\\cagent\\cagent.exe","ProcessGuid":"{49F1AF32-11B0-5A90-0000-0010594E0100}","ProcessId":"1216","TargetFilename":"C:\\commander.exe","UtcTime":"2018-02-26 16:28:13.169"},"GeneInfo":{"Criticality":10,"Signature":["ExecutableFileCreated","NewExeCreatedInRoot"]},"System":{"Channel":"Microsoft-Windows-Sysmon/Operational","Computer":"CALDERA01.caldera.loc","Correlation":{},"EventID":"11","EventRecordID":"1274413","Execution":{"ProcessID":"1408","ThreadID":"1652"},"Keywords":"0x8000000000000000","Level":"4","Opcode":"0","Provider":{"Guid":"{5770385F-C22A-43E0-BF4C-06F5698FFBD9}","Name":"Microsoft-Windows-Sysmon"},"Security":{"UserID":"S-1-5-18"},"Task":"11","TimeCreated":{"SystemTime":"2018-02-26T16:28:13.185436300Z"},"Version":"2"}}}`,
-		`{"Event":{"EventData":{"CommandLine":"\"powershell\" -command -","Company":"Microsoft Corporation","CurrentDirectory":"C:\\Windows\\system32\\","Description":"Windows PowerShell","FileVersion":"6.1.7600.16385 (win7_rtm.090713-1255)","Hashes":"SHA1=5330FEDAD485E0E4C23B2ABE1075A1F984FDE9FC,MD5=852D67A27E454BD389FA7F02A8CBE23F,SHA256=A8FDBA9DF15E41B6F5C69C79F66A26A9D48E174F9E7018A371600B866867DAB8,IMPHASH=F2C0E8A5BD10DBC167455484050CD683","Image":"C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe","IntegrityLevel":"System","LogonGuid":"{49F1AF32-11AE-5A90-0000-0020E7030000}","LogonId":"0x3e7","ParentCommandLine":"C:\\commander.exe -f","ParentImage":"C:\\commander.exe","ParentProcessGuid":"{49F1AF32-359D-5A94-0000-0010A9530C00}","ParentProcessId":"3068","ProcessGuid":"{49F1AF32-35A0-5A94-0000-0010FE5E0C00}","ProcessId":"1244","Product":"Microsoft® Windows® Operating System","TerminalSessionId":"0","User":"NT AUTHORITY\\SYSTEM","UtcTime":"2018-02-26 16:28:16.514"},"GeneInfo":{"Criticality":10,"Signature":["HeurSpawnShell","PowershellStdin"]},"System":{"Channel":"Microsoft-Windows-Sysmon/Operational","Computer":"CALDERA01.caldera.loc","Correlation":{},"EventID":"1","EventRecordID":"1274784","Execution":{"ProcessID":"1408","ThreadID":"1652"},"Keywords":"0x8000000000000000","Level":"4","Opcode":"0","Provider":{"Guid":"{5770385F-C22A-43E0-BF4C-06F5698FFBD9}","Name":"Microsoft-Windows-Sysmon"},"Security":{"UserID":"S-1-5-18"},"Task":"1","TimeCreated":{"SystemTime":"2018-04-06T16:28:16.530122800Z"},"Version":"5"}}}`,
-		`{"Event":{"EventData":{"CallTrace":"C:\\Windows\\SYSTEM32\\ntdll.dll+4d61a|C:\\Windows\\system32\\KERNELBASE.dll+19577|UNKNOWN(000000001ABD2A68)","GrantedAccess":"0x143a","SourceImage":"C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe","SourceProcessGUID":"{49F1AF32-3922-5A94-0000-0010E3581900}","SourceProcessId":"1916","SourceThreadId":"2068","TargetImage":"C:\\Windows\\system32\\lsass.exe","TargetProcessGUID":"{49F1AF32-11AD-5A90-0000-00102F6F0000}","TargetProcessId":"472","UtcTime":"2018-02-26 16:43:26.380"},"GeneInfo":{"Criticality":10,"Signature":["HeurMaliciousAccess","MaliciousLsassAccess","SuspWriteAccess","SuspiciousLsassAccess"]},"System":{"Channel":"Microsoft-Windows-Sysmon/Operational","Computer":"CALDERA01.caldera.loc","Correlation":{},"EventID":"10","EventRecordID":"1293693","Execution":{"ProcessID":"1408","ThreadID":"1652"},"Keywords":"0x8000000000000000","Level":"4","Opcode":"0","Provider":{"Guid":"{5770385F-C22A-43E0-BF4C-06F5698FFBD9}","Name":"Microsoft-Windows-Sysmon"},"Security":{"UserID":"S-1-5-18"},"Task":"10","TimeCreated":{"SystemTime":"2018-02-26T16:43:26.447894800Z"},"Version":"3"}}}`,
-	}
+	eventFile = "./data/events.json"
+	events    = make([]event.EdrEvent, 0)
 )
 
-func emitEvents(count int) (ce chan *evtx.GoEvtxMap) {
-	timecreatedPath := evtx.Path("/Event/System/TimeCreated/SystemTime")
-	ce = make(chan *evtx.GoEvtxMap)
+func init() {
+	data, err := ioutil.ReadFile(eventFile)
+	if err != nil {
+		panic(err)
+	}
+	for line := range readers.Readlines(bytes.NewBuffer(data)) {
+		event := event.EdrEvent{}
+		json.Unmarshal(line, &event)
+		events = append(events, event)
+	}
+}
+
+func emitEvents(count int) (ce chan *event.EdrEvent) {
+	ce = make(chan *event.EdrEvent)
 	go func() {
 		defer close(ce)
 		for i := 0; i < count; i++ {
-			e := new(evtx.GoEvtxMap)
 			i := rand.Int() % len(events)
-			err := json.Unmarshal([]byte(events[i]), e)
-			e.Set(&timecreatedPath, time.Now().Format(time.RFC3339Nano))
-			if err != nil {
-				log.Errorf("Cannot unmarshall event")
-			}
-			ce <- e
+			e := events[i]
+			e.Event.System.TimeCreated.SystemTime = time.Now()
+			ce <- &e
 		}
 	}()
 	return
 }
 
+func readerFromEvents(count int) io.Reader {
+	tmp := make([]string, 0, count)
+	for event := range emitEvents(count) {
+		tmp = append(tmp, string(utils.Json(event)))
+	}
+	return bytes.NewBufferString(strings.Join(tmp, "\n"))
+}
+
 func countLinesInGzFile(filepath string) int {
 	var line int
 	fd, err := os.Open(filepath)

api/log_streamer.go

@@ -5,28 +5,36 @@ import (
 	"sync"
 	"time"
 
-	"github.com/0xrawsec/golang-evtx/evtx"
 	"github.com/0xrawsec/golang-utils/datastructs"
+	"github.com/0xrawsec/whids/event"
 )
 
 type LogStream struct {
 	closed bool
-	S      chan evtx.GoEvtxMap
+	queue  datastructs.Fifo
+	S      chan *event.EdrEvent
 }
 
-func (s *LogStream) Stream(e evtx.GoEvtxMap) bool {
-	for {
-		if s.closed {
-			close(s.S)
-			return false
-		}
-		select {
-		case s.S <- e:
-			return true
-		default:
-			time.Sleep(time.Millisecond * 10)
-		}
+func (s *LogStream) Queue(e *event.EdrEvent) bool {
+	if s.closed {
+		return false
 	}
+	s.queue.Push(e)
+	return true
+}
+
+func (s *LogStream) Stream() {
+	go func() {
+		defer close(s.S)
+		for !s.closed {
+			if i := s.queue.Pop(); i != nil {
+				e := i.Value.(*event.EdrEvent)
+				s.S <- e
+			} else {
+				time.Sleep(time.Millisecond * 50)
+			}
+		}
+	}()
 }
 
 func (s *LogStream) Close() {
@@ -35,21 +43,19 @@ func (s *LogStream) Close() {
 
 type EventStreamer struct {
 	sync.RWMutex
-	queue   datastructs.Fifo
 	streams map[int]*LogStream
 }
 
 func NewEventStreamer() *EventStreamer {
 	return &EventStreamer{
-		queue:   datastructs.Fifo{},
 		streams: map[int]*LogStream{},
 	}
 }
 
 func (s *EventStreamer) NewStream() *LogStream {
 	s.Lock()
 	defer s.Unlock()
-	ls := &LogStream{S: make(chan evtx.GoEvtxMap)}
+	ls := &LogStream{S: make(chan *event.EdrEvent), queue: datastructs.Fifo{}}
 	s.streams[s.newId()] = ls
 	return ls
 }
@@ -64,40 +70,15 @@ func (s *EventStreamer) newId() int {
 	}
 }
 
-func (s *EventStreamer) Queue(e evtx.GoEvtxMap) {
+func (s *EventStreamer) Queue(e *event.EdrEvent) {
 	s.Lock()
 	defer s.Unlock()
 	// we queue only if there is at least a stream open
 	if len(s.streams) > 0 {
-		s.queue.Push(e)
-	}
-}
-
-func (s *EventStreamer) Stream() {
-	go func() {
-		for {
-			if i := s.queue.Pop(); i != nil {
-				e := i.Value.(evtx.GoEvtxMap)
-				for id, stream := range s.streams {
-					if ok := stream.Stream(e); !ok {
-						s.delStream(id)
-					}
-				}
-			} else {
-				// we sleep only if there is nothing to stream
-				// to minimize delay
-				time.Sleep(time.Millisecond * 50)
+		for id, stream := range s.streams {
+			if ok := stream.Queue(e); !ok {
+				delete(s.streams, id)
 			}
 		}
-	}()
-}
-
-func (s *EventStreamer) delStream(id int) {
-	s.Lock()
-	defer s.Unlock()
-	delete(s.streams, id)
-}
-
-func (s *EventStreamer) Close() {
-
+	}
 }