Dissect HTTP response from inside the HTTP extension

tap/api/api.go

@@ -17,5 +17,5 @@ type Extension struct {
 type Dissector interface {
 	Register(*Extension)
 	Ping()
-	Dissect(b *bufio.Reader) interface{}
+	Dissect(b *bufio.Reader, isClient bool) interface{}
 }

tap/extensions/amqp/main.go

@@ -23,7 +23,7 @@ func (g dissecting) Ping() {
 	log.Printf("pong AMQP\n")
 }
 
-func (g dissecting) Dissect(b *bufio.Reader) interface{} {
+func (g dissecting) Dissect(b *bufio.Reader, isClient bool) interface{} {
 	// TODO: Implement
 	return nil
 }

tap/extensions/http/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bufio"
 	"io"
+	"io/ioutil"
 	"log"
 	"net/http"
 
@@ -27,38 +28,32 @@ func (g dissecting) Ping() {
 	log.Printf("pong HTTP\n")
 }
 
-func DiscardBytesToFirstError(r io.Reader) (discarded int, err error) {
+func (g dissecting) Dissect(b *bufio.Reader, isClient bool) interface{} {
 	for {
-		n, e := r.Read(discardBuffer)
+		if isClient {
-		discarded += n
+			req, err := http.ReadRequest(b)
-		if e != nil {
+			if err == io.EOF || err == io.ErrUnexpectedEOF {
-			return discarded, e
+				// We must read until we see an EOF... very important!
-		}
+				return nil
-	}
+			} else if err != nil {
-}
+				log.Println("Error reading stream:", err)
-
+			} else {
-func DiscardBytesToEOF(r io.Reader) (discarded int) {
+				body, _ := ioutil.ReadAll(req.Body)
-	for {
+				req.Body.Close()
-		n, e := DiscardBytesToFirstError(r)
+				log.Printf("Received request: %+v with body: %+v\n", req, body)
-		discarded += n
+			}
-		if e == io.EOF {
-			return
-		}
-	}
-}
-
-func (g dissecting) Dissect(b *bufio.Reader) interface{} {
-	for {
-		req, err := http.ReadRequest(b)
-		if err == io.EOF {
-			// We must read until we see an EOF... very important!
-			return nil
-		} else if err != nil {
-			log.Println("Error reading stream:", err)
 		} else {
-			bodyBytes := DiscardBytesToEOF(req.Body)
+			res, err := http.ReadResponse(b, nil)
-			req.Body.Close()
+			if err == io.EOF || err == io.ErrUnexpectedEOF {
-			log.Println("Received request from stream:", req, "with", bodyBytes, "bytes in request body")
+				// We must read until we see an EOF... very important!
+				return nil
+			} else if err != nil {
+				log.Println("Error reading stream:", err)
+			} else {
+				body, _ := ioutil.ReadAll(res.Body)
+				res.Body.Close()
+				log.Printf("Received response: %+v with body: %+v\n", res, body)
+			}
 		}
 	}
 }

tap/extensions/kafka/main.go

@@ -23,7 +23,7 @@ func (g dissecting) Ping() {
 	log.Printf("pong Kafka\n")
 }
 
-func (g dissecting) Dissect(b *bufio.Reader) interface{} {
+func (g dissecting) Dissect(b *bufio.Reader, isClient bool) interface{} {
 	// TODO: Implement
 	return nil
 }

tap/passive_tapper.go

@@ -91,7 +91,7 @@ var dumpToHar = flag.Bool("hardump", false, "Dump traffic to har files")
 var HarOutputDir = flag.String("hardir", "", "Directory in which to store output har files")
 var harEntriesPerFile = flag.Int("harentriesperfile", 200, "Number of max number of har entries to store in each file")
 
-var filter = flag.String("f", "tcp and dst port 80", "BPF filter for pcap")
+var filter = flag.String("f", "tcp and (src port 80 or dst port 80)", "BPF filter for pcap")
 
 var statsTracker = StatsTracker{}
 

tap/tcp_stream_factory.go

@@ -29,12 +29,22 @@ func containsPort(ports []string, port string) bool {
 	return false
 }
 
-func (h *tcpStream) run() {
+func (h *tcpStream) clientRun() {
 	b := bufio.NewReader(&h.r)
 	for _, extension := range extensions {
 		if containsPort(extension.OutboundPorts, h.transport.Dst().String()) {
 			extension.Dissector.Ping()
-			extension.Dissector.Dissect(b)
+			extension.Dissector.Dissect(b, true)
+		}
+	}
+}
+
+func (h *tcpStream) serverRun() {
+	b := bufio.NewReader(&h.r)
+	for _, extension := range extensions {
+		if containsPort(extension.OutboundPorts, h.transport.Src().String()) {
+			extension.Dissector.Ping()
+			extension.Dissector.Dissect(b, false)
 		}
 	}
 }
@@ -47,7 +57,9 @@ func (h *tcpStreamFactory) New(net, transport gopacket.Flow) tcpassembly.Stream
 		r:         tcpreader.NewReaderStream(),
 	}
 	if containsPort(allOutboundPorts, transport.Dst().String()) {
-		go stream.run()
+		go stream.clientRun()
+	} else if containsPort(allOutboundPorts, transport.Src().String()) {
+		go stream.serverRun()
 	}
 	return &stream.r
 }