github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2026-07-10 00:51:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement the run() function for the TCP stream

Browse Source
This commit is contained in:
M. Mert Yildiran
2021-08-17 08:43:16 +03:00
parent bfa9d9960f
commit 3410a6067c
10 changed files with 101 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
agent/go.sum
View File

@@ -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=

10
tap/api/api.go
View File

@@ -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()
}

6
tap/extensions/amqp/main.go
View File

@@ -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"

6
tap/extensions/http/main.go
View File

@@ -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"

6
tap/extensions/kafka/main.go
View File

@@ -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"

1
tap/go.mod
View File

@@ -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

2
tap/go.sum
View File

@@ -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=

5
tap/passive_tapper.go
View File

@@ -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
}
}

22
tap/tcp_stream.go
View File

@@ -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}
// }
}
}

92
tap/tcp_stream_factory.go
View File

@@ -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
}