From 3410a6067c5758bcb09450637ed3c818b75442ee Mon Sep 17 00:00:00 2001 From: "M. Mert Yildiran" Date: Tue, 17 Aug 2021 08:43:16 +0300 Subject: [PATCH] Implement the `run()` function for the TCP stream --- agent/go.sum | 2 + tap/api/api.go | 10 ++-- tap/extensions/amqp/main.go | 6 ++- tap/extensions/http/main.go | 6 ++- tap/extensions/kafka/main.go | 6 ++- tap/go.mod | 1 + tap/go.sum | 2 + tap/passive_tapper.go | 5 +- tap/tcp_stream.go | 22 ++++++++- tap/tcp_stream_factory.go | 92 ++++++++++++++++++++---------------- 10 files changed, 101 insertions(+), 51 deletions(-) diff --git a/agent/go.sum b/agent/go.sum index a496ceb4c..4d92b6ed2 100644 --- a/agent/go.sum +++ b/agent/go.sum @@ -42,6 +42,8 @@ github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb0 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0= github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE= github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY= +github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4 h1:NJOOlc6ZJjix0A1rAU+nxruZtR8KboG1848yqpIUo4M= +github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4/go.mod h1:DQPxZS994Ld1Y8uwnJT+dRL04XPD0cElP/pHH/zEBHM= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= diff --git a/tap/api/api.go b/tap/api/api.go index a12d89e18..d83c05f96 100644 --- a/tap/api/api.go +++ b/tap/api/api.go @@ -3,12 +3,14 @@ package api import "plugin" type Extension struct { - Name string - Path string - Plug *plugin.Plugin - Port int + Name string + Path string + Plug *plugin.Plugin + Port string + Dissector Dissector } type Dissector interface { Register(*Extension) + Ping() } diff --git a/tap/extensions/amqp/main.go b/tap/extensions/amqp/main.go index 194044f2a..9cb9c3797 100644 --- a/tap/extensions/amqp/main.go +++ b/tap/extensions/amqp/main.go @@ -15,7 +15,11 @@ type dissecting string func (g dissecting) Register(extension *api.Extension) { fmt.Printf("extension: %v\n", extension) - extension.Port = 5672 + extension.Port = "5672" +} + +func (g dissecting) Ping() { + fmt.Printf("pong\n") } // exported as symbol named "Greeter" diff --git a/tap/extensions/http/main.go b/tap/extensions/http/main.go index 23c096c86..92cb1d27c 100644 --- a/tap/extensions/http/main.go +++ b/tap/extensions/http/main.go @@ -15,7 +15,11 @@ type dissecting string func (g dissecting) Register(extension *api.Extension) { fmt.Printf("extension: %v\n", extension) - extension.Port = 80 + extension.Port = "80" +} + +func (g dissecting) Ping() { + fmt.Printf("pong\n") } // exported as symbol named "Greeter" diff --git a/tap/extensions/kafka/main.go b/tap/extensions/kafka/main.go index b2084d6d1..c61f63056 100644 --- a/tap/extensions/kafka/main.go +++ b/tap/extensions/kafka/main.go @@ -15,7 +15,11 @@ type dissecting string func (g dissecting) Register(extension *api.Extension) { fmt.Printf("extension: %v\n", extension) - extension.Port = 9092 + extension.Port = "9092" +} + +func (g dissecting) Ping() { + fmt.Printf("pong\n") } // exported as symbol named "Greeter" diff --git a/tap/go.mod b/tap/go.mod index a4bb09f0a..4b7dbd388 100644 --- a/tap/go.mod +++ b/tap/go.mod @@ -3,6 +3,7 @@ module github.com/up9inc/mizu/tap go 1.16 require ( + github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4 // indirect github.com/google/gopacket v1.1.19 github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 github.com/up9inc/mizu/tap/api v0.0.0 diff --git a/tap/go.sum b/tap/go.sum index 1e29b2d08..6cce30cd2 100644 --- a/tap/go.sum +++ b/tap/go.sum @@ -1,3 +1,5 @@ +github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4 h1:NJOOlc6ZJjix0A1rAU+nxruZtR8KboG1848yqpIUo4M= +github.com/bradleyfalzon/tlsx v0.0.0-20170624122154-28fd0e59bac4/go.mod h1:DQPxZS994Ld1Y8uwnJT+dRL04XPD0cElP/pHH/zEBHM= github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8= github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo= github.com/romana/rlog v0.0.0-20171115192701-f018bc92e7d7 h1:jkvpcEatpwuMF5O5LVxTnehj6YZ/aEZN4NWD/Xml4pI= diff --git a/tap/passive_tapper.go b/tap/passive_tapper.go index 7fd1efa4d..ae9715464 100644 --- a/tap/passive_tapper.go +++ b/tap/passive_tapper.go @@ -250,7 +250,7 @@ func loadExtensions() { log.Fatal(err) } extensions = make([]*api.Extension, len(files)) - for _, file := range files { + for i, file := range files { filename := file.Name() log.Printf("Loading extension: %s\n", filename) extension := &api.Extension{ @@ -264,8 +264,9 @@ func loadExtensions() { var dissector api.Dissector dissector, _ = symDissector.(api.Dissector) dissector.Register(extension) + extension.Dissector = dissector fmt.Printf("returned extension: %v\n", extension) - _ = append(extensions, extension) + extensions[i] = extension } } diff --git a/tap/tcp_stream.go b/tap/tcp_stream.go index f8bf3461c..b424bde98 100644 --- a/tap/tcp_stream.go +++ b/tap/tcp_stream.go @@ -5,12 +5,25 @@ import ( "encoding/hex" "fmt" "sync" + "time" "github.com/google/gopacket" "github.com/google/gopacket/layers" // pulls in all layers decoders "github.com/google/gopacket/reassembly" ) +type tcpID struct { + srcIP string + dstIP string + srcPort string + dstPort string +} + +type tcpReaderDataMsg struct { + bytes []byte + timestamp time.Time +} + /* It's a connection (bidirectional) * Implements gopacket.reassembly.Stream interface (Accept, ReassembledSG, ReassemblyComplete) * ReassembledSG gets called when new reassembled data is ready (i.e. bytes in order, no duplicates, complete) @@ -26,6 +39,11 @@ type tcpStream struct { reversed bool urls []string ident string + data []byte + msgQueue chan tcpReaderDataMsg + captureTime time.Time + packetsSeen uint + tcpID tcpID sync.Mutex } @@ -147,9 +165,9 @@ func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass // This is where we pass the reassembled information onwards // This channel is read by an httpReader object // if dir == reassembly.TCPDirClientToServer && !t.reversed { - // t.client.msgQueue <- httpReaderDataMsg{data, ac.GetCaptureInfo().Timestamp} + // t.client.msgQueue <- tcpReaderDataMsg{data, ac.GetCaptureInfo().Timestamp} // } else { - // t.server.msgQueue <- httpReaderDataMsg{data, ac.GetCaptureInfo().Timestamp} + // t.server.msgQueue <- tcpReaderDataMsg{data, ac.GetCaptureInfo().Timestamp} // } } } diff --git a/tap/tcp_stream_factory.go b/tap/tcp_stream_factory.go index d4567e9f7..8e62696dd 100644 --- a/tap/tcp_stream_factory.go +++ b/tap/tcp_stream_factory.go @@ -2,10 +2,12 @@ package tap import ( "fmt" + "io" "sync" "github.com/romana/rlog" + "github.com/bradleyfalzon/tlsx" "github.com/google/gopacket" "github.com/google/gopacket/layers" // pulls in all layers decoders "github.com/google/gopacket/reassembly" @@ -22,6 +24,51 @@ type tcpStreamFactory struct { outbountLinkWriter *OutboundLinkWriter } +const checkTLSPacketAmount = 100 + +func (h *tcpStream) run(wg *sync.WaitGroup) { + defer wg.Done() + for _, extension := range extensions { + if extension.Port == h.transport.Dst().String() { + extension.Dissector.Ping() + } + } + // b := bufio.NewReader(h) + // fmt.Printf("b: %v\n", b) +} + +func (h *tcpStream) Read(p []byte) (int, error) { + var msg tcpReaderDataMsg + + ok := true + for ok && len(h.data) == 0 { + msg, ok = <-h.msgQueue + h.data = msg.bytes + + h.captureTime = msg.timestamp + if len(h.data) > 0 { + h.packetsSeen += 1 + } + if h.packetsSeen < checkTLSPacketAmount && len(msg.bytes) > 5 { // packets with less than 5 bytes cause tlsx to panic + clientHello := tlsx.ClientHello{} + err := clientHello.Unmarshall(msg.bytes) + if err == nil { + // statsTracker.incTlsConnectionsCount() + fmt.Printf("Detected TLS client hello with SNI %s\n", clientHello.SNI) + // numericPort, _ := strconv.Atoi(h.tcpID.dstPort) + // h.outboundLinkWriter.WriteOutboundLink(h.tcpID.srcIP, h.tcpID.dstIP, numericPort, clientHello.SNI, TLSProtocol) + } + } + } + if !ok || len(h.data) == 0 { + return 0, io.EOF + } + + l := copy(p, h.data) + h.data = h.data[l:] + return l, nil +} + func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.TCP, ac reassembly.AssemblerContext) reassembly.Stream { rlog.Debugf("* NEW: %s %s", net, transport) fsmOptions := reassembly.TCPSimpleFSMOptions{ @@ -32,9 +79,9 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T dstIp := net.Dst().String() dstPort := int(tcp.DstPort) - if factory.shouldNotifyOnOutboundLink(dstIp, dstPort) { - factory.outbountLinkWriter.WriteOutboundLink(net.Src().String(), dstIp, dstPort, "", "") - } + // if factory.shouldNotifyOnOutboundLink(dstIp, dstPort) { + // factory.outbountLinkWriter.WriteOutboundLink(net.Src().String(), dstIp, dstPort, "", "") + // } props := factory.getStreamProps(srcIp, dstIp, dstPort) isHTTP := props.isTapTarget stream := &tcpStream{ @@ -47,43 +94,8 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T ident: fmt.Sprintf("%s:%s", net, transport), optchecker: reassembly.NewTCPOptionCheck(), } - if stream.isHTTP { - // stream.client = httpReader{ - // msgQueue: make(chan httpReaderDataMsg), - // ident: fmt.Sprintf("%s %s", net, transport), - // tcpID: tcpID{ - // srcIP: net.Src().String(), - // dstIP: net.Dst().String(), - // srcPort: transport.Src().String(), - // dstPort: transport.Dst().String(), - // }, - // hexdump: *hexdump, - // parent: stream, - // isClient: true, - // isOutgoing: props.isOutgoing, - // harWriter: factory.harWriter, - // outboundLinkWriter: factory.outbountLinkWriter, - // } - // stream.server = httpReader{ - // msgQueue: make(chan httpReaderDataMsg), - // ident: fmt.Sprintf("%s %s", net.Reverse(), transport.Reverse()), - // tcpID: tcpID{ - // srcIP: net.Dst().String(), - // dstIP: net.Src().String(), - // srcPort: transport.Dst().String(), - // dstPort: transport.Src().String(), - // }, - // hexdump: *hexdump, - // parent: stream, - // isOutgoing: props.isOutgoing, - // harWriter: factory.harWriter, - // outboundLinkWriter: factory.outbountLinkWriter, - // } - // factory.wg.Add(2) - // // Start reading from channels stream.client.bytes and stream.server.bytes - // go stream.client.run(&factory.wg) - // go stream.server.run(&factory.wg) - } + factory.wg.Add(1) + go stream.run(&factory.wg) return stream }