Initial PR to implement openSSL userspace tracker

Signed-off-by: Mohamed S. Mahmoud <[email protected]>

Author: msherif1234

.mk/bc.mk

@@ -22,7 +22,8 @@ define PROGRAMS
 	"xfrm_input_kprobe": "kprobe",
 	"xfrm_input_kretprobe": "kretprobe",
 	"xfrm_output_kprobe": "kprobe",
-	"xfrm_output_kretprobe": "kretprobe"
+	"xfrm_output_kretprobe": "kretprobe",
+	"probe_entry_SSL_write":"uprobe",
 }
 endef
 
@@ -38,7 +39,8 @@ define MAPS
 	"filter_map":"lpm_trie",
 	"peer_filter_map":"lpm_trie",
 	"ipsec_ingress_map":"hash",
-	"ipsec_egress_map":"hash"
+	"ipsec_egress_map":"hash",
+	"ssl_data_event_map":"ringbuf"
 }
 endef
 

Dockerfile

@@ -3,7 +3,6 @@ ARG TARGETARCH
 # Build the manager binary
 FROM docker.io/library/golang:1.24 as builder
 
-ARG TARGETARCH
 ARG LDFLAGS
 
 WORKDIR /opt/app-root

Makefile

@@ -205,7 +205,7 @@ create-and-deploy-kind-cluster: prereqs ## Create a kind cluster and deploy the
 
 .PHONY: destroy-kind-cluster
 destroy-kind-cluster: ## Destroy the kind cluster.
-	oc delete -f scripts/agent.yml
+	kubectl delete -f scripts/agent.yml
 	kind delete cluster
 
 ##@ Images

bpf/configs.h

@@ -15,4 +15,5 @@ volatile const u8 enable_network_events_monitoring = 0;
 volatile const u8 network_events_monitoring_groupid = 0;
 volatile const u8 enable_pkt_translation_tracking = 0;
 volatile const u8 enable_ipsec = 0;
+volatile const u8 enable_ssl = 0;
 #endif //__CONFIGS_H__

bpf/flows.c

@@ -57,6 +57,11 @@
  */
 #include "ipsec.h"
 
+/*
+ * Defines ssl tracker
+ */
+#include "openssl_tracker.h"
+
 // return 0 on success, 1 if capacity reached
 static __always_inline int add_observed_intf(flow_metrics *value, pkt_info *pkt, u32 if_index,
                                              u8 direction) {

bpf/maps_definition.h

@@ -97,4 +97,11 @@ struct {
     __uint(pinning, LIBBPF_PIN_BY_NAME);
 } ipsec_egress_map SEC(".maps");
 
+// Ringbuf for SSL data events
+struct {
+    __uint(type, BPF_MAP_TYPE_RINGBUF);
+    __uint(max_entries, 1 << 27); // 16KB * 1000 events/sec * 5sec "eviction time" = ~128MB
+    __uint(pinning, LIBBPF_PIN_BY_NAME);
+} ssl_data_event_map SEC(".maps");
+
 #endif //__MAPS_DEFINITION_H__

bpf/openssl_tracker.h

@@ -0,0 +1,65 @@
+/*
+ * OpenSSL monitoring uprobe/uretprobe eBPF hook.
+ */
+
+#ifndef __OPENSSL_TRACKER_H__
+#define __OPENSSL_TRACKER_H__
+
+#include "utils.h"
+
+static inline void generate_SSL_data_event(struct pt_regs *ctx, u64 pid_tgid, u8 ssl_type,
+                                           const char *buf, int len) {
+    if (len <= 0) {
+        return;
+    }
+
+    struct ssl_data_event_t *event;
+    event = bpf_ringbuf_reserve(&ssl_data_event_map, sizeof(*event), 0);
+    if (!event) {
+        return;
+    }
+    event->timestamp_ns = bpf_ktime_get_ns();
+    event->pid_tgid = pid_tgid;
+    event->ssl_type = ssl_type;
+    event->data_len = len < MAX_DATA_SIZE_OPENSSL ? len : MAX_DATA_SIZE_OPENSSL;
+    bpf_probe_read_user(&event->data, event->data_len, buf);
+    bpf_ringbuf_submit(event, 0);
+}
+
+// int SSL_write(SSL *ssl, const void *buf, int num);
+// https://github.com/openssl/openssl/blob/master/ssl/ssl_lib.c#L2666
+SEC("uprobe/SSL_write")
+int probe_entry_SSL_write(struct pt_regs *ctx) {
+    if (enable_ssl == 0) {
+        return 0;
+    }
+
+    u64 pid_tgid = bpf_get_current_pid_tgid();
+
+    BPF_PRINTK("openssl uprobe/SSL_write pid: %d\n", pid_tgid);
+    // https://github.com/openssl/openssl/blob/master/ssl/ssl_local.h#L1233
+    void *ssl = (void *)PT_REGS_PARM1(ctx);
+
+    u32 ssl_type;
+    int ret;
+
+    ret = bpf_probe_read_user(&ssl_type, sizeof(ssl_type), (u32 *)ssl);
+    if (ret) {
+        BPF_PRINTK("(OPENSSL) bpf_probe_read ssl_type_ptr failed, ret: %d\n", ret);
+        return 0;
+    }
+    const char *buf = (const char *)PT_REGS_PARM2(ctx);
+    int num = (int)PT_REGS_PARM3(ctx); // Third parameter: number of bytes to write
+
+    BPF_PRINTK("openssl uprobe/SSL_write type: %d, buf: %p, num: %d\n", ssl_type, buf, num);
+
+    // Read the data immediately in the uprobe (before SSL_write processes it)
+    // This captures the plaintext before encryption
+    if (num > 0) {
+        generate_SSL_data_event(ctx, pid_tgid, ssl_type, buf, num);
+    }
+
+    return 0;
+}
+
+#endif /* __OPENSSL_TRACKER_H__  */

bpf/types.h

@@ -277,4 +277,17 @@ struct filter_value_t {
 // Force emitting enums/structs into the ELF
 const static struct filter_value_t *unused12 __attribute__((unused));
 
+#define MAX_DATA_SIZE_OPENSSL 1024 * 16
+// SSL data event
+struct ssl_data_event_t {
+    u64 timestamp_ns;
+    u64 pid_tgid;
+    s32 data_len;
+    u8 ssl_type;
+    char data[MAX_DATA_SIZE_OPENSSL];
+} ssl_data_event;
+
+// Force emitting enums/structs into the ELF
+const static struct ssl_data_event_t *unused14 __attribute__((unused));
+
 #endif /* __TYPES_H__ */

examples/test-ssl-host.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+# Test SSL tracking with HOST processes (not containers)
+
+echo "=== Testing SSL with Host Process ==="
+echo ""
+echo "This will SSH to a cluster node and run curl directly on the host"
+echo "This should trigger the SSL uprobes since the host process uses"
+echo "the same libssl.so that the agent attached to."
+echo ""
+
+# Get node name
+NODE=$(kubectl get nodes -o jsonpath='{.items[0].metadata.name}')
+echo "Target node: $NODE"
+echo ""
+
+# Run curl on the host (via docker exec into the kind node)
+echo "Running curl with HTTP/1.1 (--http1.1) on host..."
+docker exec $NODE curl -s --http1.1 https://httpbin.org/get > /dev/null
+echo "Request completed"
+echo ""
+
+echo "Check agent logs for SSL events:"
+kubectl logs -n netobserv-privileged -l k8s-app=netobserv-ebpf-agent | grep 'SSL EVENT'

pkg/agent/agent.go

@@ -86,7 +86,8 @@ type Flows struct {
 	promoServer   *http.Server
 	sampleDecoder *ovnobserv.SampleDecoder
 
-	metrics *metrics.Metrics
+	metrics     *metrics.Metrics
+	rbSSLTracer *flow.RingBufTracer
 }
 
 // ebpfFlowFetcher abstracts the interface of ebpf.FlowFetcher to allow dependency injection in tests
@@ -97,6 +98,7 @@ type ebpfFlowFetcher interface {
 	LookupAndDeleteMap(*metrics.Metrics) map[ebpf.BpfFlowId]model.BpfFlowContent
 	DeleteMapsStaleEntries(timeOut time.Duration)
 	ReadRingBuf() (ringbuf.Record, error)
+	ReadSSLRingBuf() (ringbuf.Record, error)
 }
 
 // FlowsAgent instantiates a new agent, given a configuration.
@@ -175,6 +177,8 @@ func FlowsAgent(cfg *config.Agent) (*Flows, error) {
 		BpfManBpfFSPath:                cfg.BpfManBpfFSPath,
 		EnableIPsecTracker:             cfg.EnableIPsecTracking,
 		FilterConfig:                   filterRules,
+		EnableSSL:                      cfg.EnableSSL,
+		OpenSSLPath:                    cfg.OpenSSLPath,
 	}
 
 	fetcher, err := tracer.NewFlowFetcher(ebpfConfig, m)
@@ -206,6 +210,10 @@ func flowsAgent(
 
 	mapTracer := flow.NewMapTracer(fetcher, cfg.CacheActiveTimeout, cfg.StaleEntriesEvictTimeout, m, s, cfg.EnableUDNMapping)
 	rbTracer := flow.NewRingBufTracer(fetcher, mapTracer, cfg.CacheActiveTimeout, m)
+	var rbSSLTracer *flow.RingBufTracer
+	if cfg.EnableSSL {
+		rbSSLTracer = flow.NewSSLRingBufTracer(fetcher, mapTracer, cfg.CacheActiveTimeout, m)
+	}
 	accounter := flow.NewAccounter(cfg.CacheMaxFlows, cfg.CacheActiveTimeout, time.Now, monotime.Now, m, s, cfg.EnableUDNMapping)
 	limiter := flow.NewCapacityLimiter(m)
 
@@ -222,6 +230,7 @@ func flowsAgent(
 		informer:    informer,
 		promoServer: promoServer,
 		metrics:     m,
+		rbSSLTracer: rbSSLTracer,
 	}, nil
 }
 
@@ -392,6 +401,10 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*mo
 	alog.Debug("connecting flows processing graph")
 	mapTracer := node.AsStart(f.mapTracer.TraceLoop(ctx, f.cfg.ForceGC))
 	rbTracer := node.AsStart(f.rbTracer.TraceLoop(ctx))
+	var rbSSLTracer *node.Start[*model.RawRecord]
+	if f.cfg.EnableSSL {
+		rbSSLTracer = node.AsStart(f.rbSSLTracer.TraceLoop(ctx))
+	}
 
 	accounter := node.AsMiddle(f.accounter.Account,
 		node.ChannelBufferLen(f.cfg.BuffersLength))
@@ -408,6 +421,9 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*mo
 		node.ChannelBufferLen(ebl))
 
 	rbTracer.SendsTo(accounter)
+	if rbSSLTracer != nil {
+		rbSSLTracer.SendsTo(accounter)
+	}
 
 	mapTracer.SendsTo(limiter)
 	accounter.SendsTo(limiter)
@@ -416,6 +432,9 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*mo
 	alog.Debug("starting graph")
 	mapTracer.Start()
 	rbTracer.Start()
+	if rbSSLTracer != nil {
+		rbSSLTracer.Start()
+	}
 	return export, nil
 }