Initial PR to implement openSSL userspace tracker

Signed-off-by: Mohamed S. Mahmoud <[email protected]>
(cherry picked from commit 49444b1)

Author: msherif1234

.mk/bc.mk

@@ -22,7 +22,9 @@ 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",
+	"probe_exit_SSL_write":"uretprobe"
 }
 endef
 
@@ -38,7 +40,9 @@ define MAPS
 	"filter_map":"lpm_trie",
 	"peer_filter_map":"lpm_trie",
 	"ipsec_ingress_map":"hash",
-	"ipsec_egress_map":"hash"
+	"ipsec_egress_map":"hash",
+	"active_ssl_write_map":"hash",
+	"ssl_data_event_map":"ringbuf"
 }
 endef
 

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,20 @@ struct {
     __uint(pinning, LIBBPF_PIN_BY_NAME);
 } ipsec_egress_map SEC(".maps");
 
+// HashMap to store active SSL write args
+struct {
+    __uint(type, BPF_MAP_TYPE_HASH);
+    __type(key, u64);
+    __type(value, struct active_ssl_buf_t);
+    __uint(map_flags, BPF_F_NO_PREALLOC);
+    __uint(pinning, LIBBPF_PIN_BY_NAME);
+} active_ssl_write_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,86 @@
+/*
+ * 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 = (int)PT_REGS_RC(ctx);
+    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 (do_sampling == 0 || 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);
+
+    BPF_PRINTK("openssl uprobe/SSL_write type: %d, buf: %p\n", ssl_type, buf);
+
+    struct active_ssl_buf_t active_ssl_buf_t;
+    __builtin_memset(&active_ssl_buf_t, 0, sizeof(active_ssl_buf_t));
+    active_ssl_buf_t.ssl_type = ssl_type;
+    bpf_probe_read_user(&active_ssl_buf_t.buf, sizeof(active_ssl_buf_t.buf), buf);
+
+    ret = bpf_map_update_elem(&active_ssl_write_map, &pid_tgid, &active_ssl_buf_t, BPF_NOEXIST);
+    if (ret != 0) {
+        if (trace_messages) {
+            BPF_PRINTK("error creating new active_ssl_write_map dir: %d err: %d\n", pid_tgid, ret);
+        }
+    }
+    return 0;
+}
+
+SEC("uretprobe/SSL_write")
+int probe_exit_SSL_write(struct pt_regs *ctx) {
+    if (do_sampling == 0 || enable_ssl == 0) {
+        return 0;
+    }
+    u64 pid_tgid = bpf_get_current_pid_tgid();
+
+    struct active_ssl_buf_t *active_ssl_buf = bpf_map_lookup_elem(&active_ssl_write_map, &pid_tgid);
+    if (active_ssl_buf != NULL) {
+        u8 ssl_type = active_ssl_buf->ssl_type;
+        generate_SSL_data_event(ctx, pid_tgid, ssl_type, (const char *)active_ssl_buf->buf);
+        bpf_map_delete_elem(&active_ssl_write_map, &pid_tgid);
+    }
+    return 0;
+}
+
+#endif /* __OPENSSL_TRACKER_H__  */

bpf/types.h

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

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
 }
 

pkg/config/config.go

@@ -252,6 +252,10 @@ type Agent struct {
 	// This setting is only used when the interface name could not be found for a given index and MAC.
 	// E.g. "0a:58=eth0" (used for ovn-kubernetes)
 	PreferredInterfaceForMACPrefix string `env:"PREFERRED_INTERFACE_FOR_MAC_PREFIX"`
+	// EnableSSL enable tracking SSL flows encryption
+	EnableSSL bool `env:"ENABLE_SSL" envDefault:"false"`
+	// OpenSSLPath path to the openssl binary
+	OpenSSLPath string `env:"OPENSSL_PATH" envDefault:"/usr/bin/openssl"`
 
 	/* Deprecated configs are listed below this line
 	 * See manageDeprecatedConfigs function for details