Merge branch 'dev' of github.com:FireTail-io/firetail-kubernetes-sensor into dev

Author: rileyfiretail

README.md

@@ -14,7 +14,7 @@
 | ----------------------------------------------- | ---------   | ------------------------------------------------------------ | ------------------------------------------------------------   |
 | `FIRETAIL_API_TOKEN`                            | ✅         | `PS-02-XXXXXXXX`                                             | The API token the sensor will use to report logs to FireTail  |
 | `BPF_EXPRESSION`                                | ❌         | `tcp and (port 80 or port 443)`                              | The BPF filter used by the sensor. See docs for syntax info: https://www.tcpdump.org/manpages/pcap-filter.7.html |
-| `MAX_CONTENT_LENGTH`                            | ❌         | `1048576`                                                    | The sensor will only read request or response bodies if their length is less than `MAX_CONTENT_LENGTH` bytes. |
+| `MAX_CONTENT_LENGTH`                            | ❌         | `1048576`                                                    | The sensor will only read requests or responses if their length is less than `MAX_CONTENT_LENGTH` bytes. |
 | `ENABLE_ONLY_LOG_JSON`                          | ❌         | `true`                                                       | Enables only logging requests where the content-type implies the payload should be JSON, or the payload is valid JSON regardless of the content-type. |
 | `DISABLE_SERVICE_IP_FILTERING`                  | ❌         | `true`                                                       | Disables polling Kubernetes for the IP addresses of services & subsequently ignoring all requests captured that aren't made to one of those IPs. |
 | `FIRETAIL_API_URL`                              | ❌         | `https://api.logging.eu-west-1.prod.firetail.app/logs/bulk`  | The API url the sensor will send logs to. Defaults to the EU region production environment. |

src/bidirectional_stream.go

@@ -28,8 +28,22 @@ func (f *bidirectionalStreamFactory) New(netFlow, tcpFlow gopacket.Flow) tcpasse
 
 	// The second time we see the same connection, it will be from the server to the client
 	if conn, ok := f.conns.LoadAndDelete(fmt.Sprint(key)); ok {
+		slog.Debug(
+			"Found existing connection, assuming this is a server to client connection",
+			"Src", netFlow.Src().String(),
+			"Dst", netFlow.Dst().String(),
+			"SrcPort", tcpFlow.Src().String(),
+			"DstPort", tcpFlow.Dst().String(),
+		)
 		return &conn.(*bidirectionalStream).serverToClient
 	}
+	slog.Debug(
+		"Found new connection, assuming this is a client to server connection",
+		"Src", netFlow.Src().String(),
+		"Dst", netFlow.Dst().String(),
+		"SrcPort", tcpFlow.Src().String(),
+		"DstPort", tcpFlow.Dst().String(),
+	)
 
 	s := &bidirectionalStream{
 		net:                       netFlow,
@@ -87,7 +101,7 @@ func (s *bidirectionalStream) run() {
 			slog.Debug("Failed to read request bytes from stream:", "Err", err.Error(), "BytesRead", bytesRead)
 			return
 		}
-		request, err := http.ReadRequest(bufio.NewReader(bytes.NewReader(requestBytes)))
+		request, err := http.ReadRequest(bufio.NewReader(bytes.NewReader(requestBytes[:bytesRead])))
 		if err != nil {
 			slog.Debug("Failed to read request bytes:", "Err", err.Error())
 			return
@@ -116,7 +130,7 @@ func (s *bidirectionalStream) run() {
 			slog.Debug("Failed to read response bytes from stream:", "Err", err.Error(), "BytesRead", bytesRead)
 			return
 		}
-		response, err := http.ReadResponse(bufio.NewReader(bytes.NewReader(responseBytes)), nil)
+		response, err := http.ReadResponse(bufio.NewReader(bytes.NewReader(responseBytes[:bytesRead])), nil)
 		if err != nil {
 			slog.Debug("Failed to read response bytes:", "Err", err.Error())
 			return

src/request_and_response.go

@@ -53,9 +53,18 @@ func (s *httpRequestAndResponseStreamer) start() {
 		),
 	)
 
-	handler, packetsChannel := s.getHandleAndPacketsChannel()
+	go func() {
+		ticker := time.Tick(time.Minute)
+		for {
+			select {
+			case <-ticker:
+				slog.Debug("Flushing old conns...")
+				assembler.FlushOlderThan(time.Now().Add(-2 * time.Minute))
+			}
+		}
+	}()
 
-	ticker := time.Tick(time.Minute)
+	handler, packetsChannel := s.getHandleAndPacketsChannel()
 	for {
 		select {
 		case packet, ok := <-packetsChannel:
@@ -87,10 +96,14 @@ func (s *httpRequestAndResponseStreamer) start() {
 				)
 				continue
 			}
+			slog.Debug(
+				"Captured packet:",
+				"Src", src,
+				"Dst", dst,
+				"SrcPort", tcp.SrcPort.String(),
+				"DstPort", tcp.DstPort.String(),
+			)
 			assembler.AssembleWithTimestamp(packet.NetworkLayer().NetworkFlow(), tcp, packet.Metadata().Timestamp)
-		case <-ticker:
-			assembler.FlushOlderThan(time.Now().Add(-2 * time.Minute))
-		default:
 		}
 	}
 }