fix: fix for request body parsing

common/ip_resolver.go

@@ -152,6 +152,15 @@ func (resolver *K8sIPResolver) ResolveIP(ip string) Workload {
 	}
 }
 
+func (resolver *K8sIPResolver) CacheDNS(ip string, dns string) Workload {
+	resolver.dnsResolvedIps.Add(ip, dns)
+	return Workload{
+		Name:      dns,
+		Namespace: "external",
+		Kind:      "external",
+	}
+}
+
 func (resolver *K8sIPResolver) StartWatching() error {
 	// register watchers
 	podsWatcher, err := resolver.clientset.CoreV1().Pods("").Watch(context.Background(), metav1.ListOptions{})

containers/container.go

@@ -1,11 +1,15 @@
 package containers
 
 import (
+	"io"
+	"log"
 	"os"
 	"strings"
 	"sync"
 	"time"
 
+	"encoding/base64"
+
 	"github.com/coroot/coroot-node-agent/cgroup"
 	"github.com/coroot/coroot-node-agent/common"
 	"github.com/coroot/coroot-node-agent/ebpftracer"
@@ -26,7 +30,7 @@ import (
 var (
 	gcInterval       = 10 * time.Minute
 	pingTimeout      = 300 * time.Millisecond
-	payloadThreshold = 1024 * 2
+	payloadThreshold = 1024 * 1024
 )
 
 type ContainerID string
@@ -138,6 +142,7 @@ type Container struct {
 
 	done        chan struct{}
 	ip_resolver IPResolver
+	hostIpsMap  sync.Map
 }
 
 func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, hostConntrack *Conntrack, pid uint32, ip_resolver IPResolver) (*Container, error) {
@@ -174,6 +179,7 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, host
 
 		done:        make(chan struct{}),
 		ip_resolver: ip_resolver,
+		hostIpsMap:  sync.Map{},
 	}
 
 	for _, n := range md.networks {
@@ -302,6 +308,12 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
 			workload_src = c.ip_resolver.ResolveIP(d.src.IP().String())
 			workload_dest = c.ip_resolver.ResolveIP(d.dst.IP().String())
 		}
+
+		if d.dstWorkload.Kind == "external" {
+			if host, ok := c.hostIpsMap.Load(d.dst); ok {
+				workload_dest = common.Workload{Name: host.(string), Namespace: "external", Kind: "external"}
+			}
+		}
 		ch <- counter(metrics.NetConnectsSuccessful, float64(count), d.src.String(), d.dst.String(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind)
 	}
 	for dst, count := range c.connectsFailed {
@@ -316,6 +328,11 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
 			workload_src = c.ip_resolver.ResolveIP(d.src.IP().String())
 			workload_dest = c.ip_resolver.ResolveIP(d.dst.IP().String())
 		}
+		if d.dstWorkload.Kind == "external" {
+			if host, ok := c.hostIpsMap.Load(d.dst); ok {
+				workload_dest = common.Workload{Name: host.(string), Namespace: "external", Kind: "external"}
+			}
+		}
 		ch <- counter(metrics.NetRetransmits, float64(count), d.src.String(), d.dst.String(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind)
 	}
 
@@ -327,6 +344,11 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
 		workload_src := c.ip_resolver.ResolveIP(addrPair.dst.IP().String())
 		workload_dest := c.ip_resolver.ResolveIP(conn.ActualDest.IP().String())
 		actualDestWorkload := c.ip_resolver.ResolveActualIP(conn.ActualDest.IP().String())
+		if workload_dest.Kind == "external" {
+			if host, ok := c.hostIpsMap.Load(conn.ActualDest.IP().String()); ok {
+				workload_dest = common.Workload{Name: host.(string), Namespace: "external", Kind: "external"}
+			}
+		}
 		connections[AddrPair{src: addrPair.dst, dst: conn.ActualDest, srcWorkload: workload_src, dstWorkload: workload_dest, actualDestWorkload: actualDestWorkload}]++
 	}
 	for d, count := range connections {
@@ -503,6 +525,20 @@ func (c *Container) onListenClose(pid uint32, addr netaddr.IPPort) {
 	}
 }
 
+func ignoreControlPlane(name string) bool {
+	keywords := strings.Split(*flags.IgnoreControlPlane, ",")
+	if len(keywords) == 0 {
+		return false
+	}
+	for _, keyword := range keywords {
+		if strings.Contains(strings.ToLower(name), keyword) {
+			klog.Warningln("Ignoring control plane ", name)
+			return true
+		}
+	}
+	return false
+}
+
 func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPPort, timestamp uint64, failed bool) {
 	if common.PortFilter.ShouldBeSkipped(dst.Port()) {
 		return
@@ -514,6 +550,11 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP
 	if dst.IP().IsLoopback() && !p.isHostNs() {
 		return
 	}
+	srcWorkload := c.ip_resolver.ResolveIP(src.IP().String())
+	if ignoreControlPlane(srcWorkload.Name) {
+		klog.Warningf("Ignoring src workload %s, %s \n", src.IP().String(), srcWorkload.Name)
+		return
+	}
 	actualDst, err := c.getActualDestination(p, src, dst)
 	if err != nil {
 		if !common.IsNotExist(err) {
@@ -613,19 +654,43 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R
 		host, error := l7.ParseHostFromHttpRequest(string(r.Payload))
 		if error == nil {
 			conn.dstWorkload.Name = host
+			c.hostIpsMap.Store(conn.ActualDest.IP().String(), host)
 		}
 	}
 	stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest, r, conn.srcWorkload, conn.dstWorkload, conn.actualDestWorkload)
 	trace := tracing.NewTrace(string(c.id), conn.ActualDest, conn.srcWorkload, conn.dstWorkload, conn.actualDestWorkload)
 	switch r.Protocol {
 	case l7.ProtocolHTTP:
 		stats.observe(r.Status.Http(), "", r.Duration)
-		method, path := l7.ParseHttp(r.Payload)
 		payload := ""
-		if int(r.PayloadSize) < payloadThreshold {
-			payload = string(r.Payload)
+		headers := ""
+		method := ""
+		uri := ""
+		response := ""
+		host := ""
+		req, err := l7.ParseHTTPRequest(r.Payload)
+		if err != nil {
+			log.Printf("Failed to parse payload %s, %q", err, string(r.Payload))
+			method, uri = l7.ParseHttp(r.Payload)
+			host, _ = l7.ParseHostFromHttpRequest(string(r.Payload))
+		} else {
+			if req != nil && req.Body != nil {
+				body, _ := io.ReadAll(req.Body)
+				base64String := base64.StdEncoding.EncodeToString(body)
+				payload = string(base64String)
+			}
+			if req != nil && req.Header != nil {
+				headersStr := l7.ConvertHeadersToString(req.Header)
+				headers = base64.StdEncoding.EncodeToString([]byte(headersStr))
+			}
+			method = req.Method
+			uri = req.URL.Path
+			host = req.Host
+		}
+		if r.Response != nil {
+			response = base64.StdEncoding.EncodeToString(r.Response)
 		}
-		trace.HttpRequest(method, path, r.Status, r.Duration, r.PayloadSize, payload)
+		trace.HttpRequest(method, uri, r.Status, r.Duration, r.PayloadSize, payload, headers, response, host)
 	case l7.ProtocolHTTP2:
 		if conn.http2Parser == nil {
 			conn.http2Parser = l7.NewHttp2Parser()

containers/registry.go

@@ -34,6 +34,7 @@ type ProcessInfo struct {
 type IPResolver interface {
 	ResolveIP(string) common.Workload
 	ResolveActualIP(string) common.Workload
+	CacheDNS(string, string) common.Workload
 	StartWatching() error
 	StopWatching()
 }

ebpftracer/ebpf/l7/l7.c

@@ -24,7 +24,7 @@
 #define METHOD_HTTP2_CLIENT_FRAMES  5
 #define METHOD_HTTP2_SERVER_FRAMES  6
 
-#define MAX_PAYLOAD_SIZE 1024 // must be power of 2
+#define MAX_PAYLOAD_SIZE 1024 * 5 // must be power of 2
 #define TRUNCATE_PAYLOAD_SIZE(size) ({                                  \
     size = MIN(size, MAX_PAYLOAD_SIZE-1);                               \
     asm volatile ("%0 &= %1" : "+r"(size) : "i"(MAX_PAYLOAD_SIZE-1));   \
@@ -63,7 +63,9 @@ struct l7_event {
     __u16 padding;
     __u32 statement_id;
     __u64 payload_size;
+    __u64 response_size;
     char payload[MAX_PAYLOAD_SIZE];
+    char response[MAX_PAYLOAD_SIZE];
 };
 
 struct {
@@ -388,7 +390,8 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
     e->method = METHOD_UNKNOWN;
     e->statement_id = 0;
     e->payload_size = 0;
-
+    e->response_size = ret;
+    COPY_PAYLOAD(e->response, ret, payload);
     if (is_rabbitmq_consume(payload, ret)) {
         e->protocol = PROTOCOL_RABBITMQ;
         e->method = METHOD_CONSUME;
@@ -427,7 +430,6 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
     e->protocol = req->protocol;
     e->payload_size = req->payload_size;
     COPY_PAYLOAD(e->payload, req->payload_size, req->payload);
-
     bpf_map_delete_elem(&active_l7_requests, &k);
     if (e->protocol == PROTOCOL_HTTP) {
         response = is_http_response(payload, &e->status);

ebpftracer/l7/http.go

@@ -3,9 +3,16 @@ package l7
 import (
 	"bytes"
 	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
 	"log"
+	"net/http"
+	"net/url"
 	"regexp"
 	"strings"
+
+	"github.com/coroot/coroot-node-agent/flags"
 )
 
 func ParseHttp(payload []byte) (string, string) {
@@ -23,30 +30,74 @@ func ParseHttp(payload []byte) (string, string) {
 	return string(method), string(uri)
 }
 
-func ParseHttpAndRest(payload []byte) (string, string, string, string) {
-	data := string(payload)
-	data = strings.ReplaceAll(data, "\\n\\n", "\n\n")
-	split := strings.Split(data, "\n\n")
-	rest := []byte(split[0])
-	d := split[1]
-	method, rest, ok := bytes.Cut(rest, space)
+func ParseHTTPRequest(data []byte) (*http.Request, error) {
+	method, rest, ok := bytes.Cut(data, space)
 	if !ok {
-		return "", "", "", ""
+		log.Printf("invalid metheod")
+		return nil, errors.New("invalid payload")
 	}
 	if !isHttpMethod(string(method)) {
-		return "", "", "", ""
+		log.Printf("invalid method")
+		return nil, errors.New("invalid payload")
 	}
 	uri, rest, ok := bytes.Cut(rest, space)
 	if !ok {
 		uri = append(uri, []byte("...")...)
 	}
 
-	_, headers, ok := bytes.Cut(rest, []byte{'\n'})
+	httpVersion, rest, _ := bytes.Cut(rest, []byte{'\n'})
+	remaining := bytes.Split(rest, []byte{'\r', '\n', '\r', '\n'})
+	headers := remaining[0]
+	var body []byte
+	if len(remaining) > 1 {
+		body = remaining[1]
+	}
 	if !ok {
-		uri = append(uri, []byte("...")...)
+		return nil, errors.New("invalid headers")
+	}
+	parsedURL, err := url.ParseRequestURI(string(uri))
+
+	if err != nil {
+		return nil, errors.New("invalid uri")
+	}
+
+	// Parse headers (skipping first line)
+	u := &url.URL{
+		Scheme:   parsedURL.Scheme,
+		Path:     parsedURL.Path,
+		RawQuery: parsedURL.RawQuery,
+	}
+	var header = http.Header{}
+	var host = ""
+	headerLines := bytes.Split(headers, []byte("\n"))
+
+	for _, line := range headerLines {
+		part1, part2, ok := bytes.Cut(line, []byte(":"))
+		if !ok {
+			continue
+		}
+		if strings.HasPrefix(string(part1), "Host") {
+			host = strings.TrimSpace(string(part2))
+		}
+		key := strings.TrimSpace(string(part1))
+		val := strings.TrimSpace(string(part2))
+		header.Add(key, val)
+	}
+	req := &http.Request{
+		Method:     string(method),
+		URL:        u,
+		Proto:      string(httpVersion),
+		ProtoMajor: 1,
+		ProtoMinor: 1,
+		Host:       host,
+		Header:     header,
+	}
+	if len(body) > 0 || body != nil {
+		p := body
+		req.Body = io.NopCloser(bytes.NewReader(p))
+		defer req.Body.Close()
 	}
-	log.Printf("data %s", headers)
-	return string(method), string(uri), string(headers), string(d)
+	return req, nil
 }
 
 func ParseHostFromHttpRequest(input string) (string, error) {
@@ -72,36 +123,40 @@ func ParseHostFromHttpRequest(input string) (string, error) {
 	return host, nil
 }
 
-func SanitizeString(input string) string {
-	// Regular expression patterns to match various sensitive data formats
-	sensitivePatterns := []*regexp.Regexp{
-		// Authorization header (Bearer or Basic)
-		// Example: Authorization: Basic c3FhXzdhODNiZTRjY2Y0M2E2NzFhMTI0ODViYmMyY2I4ZGU4MDk0MDQyMzE6
-		// Reason: Matches common formats for authorization tokens.
-		regexp.MustCompile(`(?i)Authorization: (Bearer|Basic)\s+[a-zA-Z0-9\-_\.=]+`),
+func ConvertHeadersToString(headers http.Header) string {
+	var headerStrings []string
 
-		// API key
-		// Example: ApiKey abcdef1234567890
-		// Reason: Matches common formats for API keys.
-		regexp.MustCompile(`(?i)ApiKey\s+[a-zA-Z0-9\-_\.=]+`),
+	for key, values := range headers {
+		for _, value := range values {
+			headerStrings = append(headerStrings, fmt.Sprintf("%s: %s", key, SanitizeString(value)))
+		}
+	}
 
-		// JWT token
-		// Example: JWT eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
-		// Reason: Matches common formats for JWT tokens.
-		regexp.MustCompile(`(?i)JWT\s+[a-zA-Z0-9\-_\.=]+`),
+	return strings.Join(headerStrings, ", ")
+}
 
-		// OAuth token
-		// Example: OAuth token1234567890
-		// Reason: Matches common formats for OAuth tokens.
-		regexp.MustCompile(`(?i)OAuth\s+[a-zA-Z0-9\-_\.=]+`),
+func SanitizeString(input string) string {
+	if !*flags.SanitizeHeaders {
+		return input
+	}
+	patternList := strings.Split(*flags.SensitiveHeaderPattern, ",")
+	// Compile regex patterns
+	var sensitivePatterns []*regexp.Regexp
+	for _, pattern := range patternList {
+		// Trim leading and trailing whitespace from the pattern
+		pattern = strings.TrimSpace(pattern)
+		if len(pattern) > 0 {
+			// Compile each pattern and append to the list
+			sensitivePatterns = append(sensitivePatterns, regexp.MustCompile(pattern))
+		}
 	}
 
 	// Replace sensitive data with placeholder '*'
 	sanitized := input
 	for _, pattern := range sensitivePatterns {
 		sanitized = pattern.ReplaceAllStringFunc(sanitized, func(match string) string {
 			// Only replace the sensitive part, keeping the structure intact
-			sanitized_string := strings.Repeat("*", len(match))
+			sanitized_string := strings.Repeat("*", min(len(match), 5))
 			log.Printf("Replacing %s with %s , using pattern %s", match, sanitized_string, pattern)
 			return sanitized_string
 		})