Make the distinction of outbound and inbound ports

tap/api/api.go

@@ -6,11 +6,12 @@ import (
 )
 
 type Extension struct {
-	Name      string
-	Path      string
-	Plug      *plugin.Plugin
-	Ports     []string
-	Dissector Dissector
+	Name          string
+	Path          string
+	Plug          *plugin.Plugin
+	InboundPorts  []string
+	OutboundPorts []string
+	Dissector     Dissector
 }
 
 type Dissector interface {

tap/extensions/amqp/main.go

@@ -15,7 +15,8 @@ type dissecting string
 
 func (g dissecting) Register(extension *api.Extension) {
 	extension.Name = "amqp"
-	extension.Ports = []string{"5671", "5672"}
+	extension.OutboundPorts = []string{"5671", "5672"}
+	extension.InboundPorts = []string{}
 }
 
 func (g dissecting) Ping() {

tap/extensions/http/main.go

@@ -19,7 +19,8 @@ type dissecting string
 
 func (g dissecting) Register(extension *api.Extension) {
 	extension.Name = "http"
-	extension.Ports = []string{"80", "8080", "443"}
+	extension.OutboundPorts = []string{"80", "8080", "443"}
+	extension.InboundPorts = []string{}
 }
 
 func (g dissecting) Ping() {

tap/extensions/kafka/main.go

@@ -15,7 +15,8 @@ type dissecting string
 
 func (g dissecting) Register(extension *api.Extension) {
 	extension.Name = "kafka"
-	extension.Ports = []string{"9092"}
+	extension.OutboundPorts = []string{"9092"}
+	extension.InboundPorts = []string{}
 }
 
 func (g dissecting) Ping() {

tap/passive_tapper.go

@@ -125,6 +125,8 @@ var nErrors uint
 var ownIps []string             // global
 var hostMode bool               // global
 var extensions []*api.Extension // global
+var allOutboundPorts []string   // global
+var allInboundPorts []string    // global
 
 type OutputChannelItem struct {
 }
@@ -240,6 +242,21 @@ func startMemoryProfiler() {
 	}()
 }
 
+func MergeUnique(slice []string, merge []string) []string {
+	for _, i := range merge {
+		add := true
+		for _, ele := range slice {
+			if ele == i {
+				add = false
+			}
+		}
+		if add {
+			slice = append(slice, i)
+		}
+	}
+	return slice
+}
+
 func loadExtensions() {
 	dir, _ := filepath.Abs(filepath.Dir(os.Args[0]))
 	extensionsDir := path.Join(dir, "./extensions/")
@@ -265,7 +282,11 @@ func loadExtensions() {
 		extension.Dissector = dissector
 		log.Printf("Extension Properties: %+v\n", extension)
 		extensions[i] = extension
+		allOutboundPorts = MergeUnique(allOutboundPorts, extension.OutboundPorts)
+		allInboundPorts = MergeUnique(allInboundPorts, extension.InboundPorts)
 	}
+	log.Printf("allOutboundPorts: %v\n", allOutboundPorts)
+	log.Printf("allInboundPorts: %v\n", allInboundPorts)
 }
 
 func startPassiveTapper(outboundLinkWriter *OutboundLinkWriter) {

tap/tcp_stream_factory.go

@@ -32,7 +32,7 @@ func containsPort(ports []string, port string) bool {
 func (h *tcpStream) run() {
 	b := bufio.NewReader(&h.r)
 	for _, extension := range extensions {
-		if containsPort(extension.Ports, h.transport.Dst().String()) {
+		if containsPort(extension.OutboundPorts, h.transport.Dst().String()) {
 			extension.Dissector.Ping()
 			extension.Dissector.Dissect(b)
 		}
@@ -46,7 +46,7 @@ func (h *tcpStreamFactory) New(net, transport gopacket.Flow) tcpassembly.Stream
 		transport: transport,
 		r:         tcpreader.NewReaderStream(),
 	}
-	if transport.Dst().String() == "80" {
+	if containsPort(allOutboundPorts, transport.Dst().String()) {
 		go stream.run()
 	}
 	return &stream.r