generate-keymap.py

#!/usr/bin/env python
# Usage: with both the terminal and /dev/input keylogger running, press every
# key on your keyboard. This program will correlate the two transcripts to
# generate a mapping from Linux kernel keycodes to terminal chars/escape codes.
#
# In other words, run:
#
#     ./keylogger > kb-keylogger.csv & ./terminal > kb-terminal.csv
#     killall -SIGINT keylogger
#     ./generate-keymap.py
#
from bisect import bisect_left, bisect_right
from collections import namedtuple
import csv

WINDOW = 5_000_000  # 5 ms in ns

Keystroke = namedtuple("Keystroke", ("key", "time"))


def read_csv(path):
    with open(path, newline="") as f:
        last_time = None
        for row in csv.DictReader(f):
            if t := row.get("TimeNS"):
                time = int(t)
            elif t := row.get("TimeMicroseconds"):
                time = int(t) * 1000
            else:
                raise KeyError

            last_time = last_time or time
            yield Keystroke(int(row["Key"]), time - last_time)


def terminal_csv():
    with open("whole-keyboard-terminal.csv", newline="") as f:
        last_time = None
        for row in csv.DictReader(f):
            key, time = int(row["Character"]), int(row["TimeNS"])
            last_time = last_time or time
            yield Keystroke(key, time - last_time)


def range_query(l, min=None, max=None, key=None):
    lo = min and bisect_left(l, min, key=key)
    hi = max and bisect_right(l, max, key=key)
    return l[lo:hi]


def keylogged_to_escaped(keylogged, escaped):
    def time(event):
        return event.time

    terminal_events = sorted(read_csv(escaped), key=time)

    for event in read_csv(keylogged):
        lo, hi = event.time - WINDOW, event.time + WINDOW
        window = range_query(terminal_events, lo, hi, key=time)
        if window:
            yield event.key, bytes(e.key for e in window)


escaped_map = []
for k, escaped in keylogged_to_escaped("kb-keylogger.csv", "kb-terminal.csv"):
    escaped_map.extend((None,) * (k - len(escaped_map) + 1))
    escaped_map[k] = escaped

print(escaped_map)

keylogger.c

// compile: cc -O3 keylogger.c -o keylogger
// run: ./keylogger >> log.csv
// license: CC0-1.0 (dedicated to public domain where possible)

#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <unistd.h>

static const unsigned short EV_KEY = 0x01;

// from linux/Documentation/input/input.rst
struct input_event {
    struct timeval time;
    unsigned short type;
    unsigned short code;
    unsigned int value;
};

int main(int argc, char **argv) {
    if (argc != 2) {
        puts("usage: keylogger /dev/input/eventN");
        return EXIT_FAILURE;
    }

    int f = open(argv[1], O_RDONLY | O_NOCTTY);
    if (f < 0) {
        perror("open");
        return EXIT_FAILURE;
    }

    uint64_t last_time_us = 0;

    struct input_event ev;
    while (read(f, &ev, sizeof(ev)) == sizeof(ev)) {
        if (ev.type == EV_KEY && ev.value != 0) {
            last_time_us = ev.time.tv_sec * 1000000 + ev.time.tv_usec;
            break;
        }
    }

    if (!last_time_us) {
        perror("read");
        return EXIT_FAILURE;
    }

    puts("Key,TimeMicroseconds");
    while (read(f, &ev, sizeof(ev)) == sizeof(ev)) {
        if (ev.type == EV_KEY && ev.value != 0) {
            uint64_t time_us = ev.time.tv_sec * 1000000 + ev.time.tv_usec;
            printf("%hu,%" PRIu64 "\n", (char)ev.code, time_us - last_time_us);
            fflush(stdout);
            last_time_us = time_us;
        }
    }

    perror("read");
    return EXIT_FAILURE;
}

replay.py

#!/usr/bin/env python
from collections import namedtuple
import time
import csv
import sys

# generated via generate-keymap.py
# fmt: off
KEYMAP = [None, None, b"1", b"2", b"3", b"4", b"5", b"6", b"7", b"8", b"9", b"0", b"-", b"=", b"\x7f", b"\t", b"q", b"w", b"e", b"r", b"t", b"y", b"u", b"i", b"o", b"p", b"[", b"]", b"\n", None, b"a", b"s", b"d", b"f", b"g", b"h", b"j", b"k", b"l", b";", b"'", b"`", None, b"\\", b"z", b"x", b"c", b"v", b"b", b"n", b"m", b",", b".", b"/", None, b"*", None, b" ", None, b"\x1bOP", b"\x1bOQ", b"\x1bOR", b"\x1bOS", b"\x1b[15~", b"\x1b[17~", b"\x1b[18~", b"\x1b[19~", b"\x1b[20~", b"\x1b[21~", None, None, b"\x1b[1~", b"\x1b[A", b"\x1b[5~", b"-", b"\x1b[D", None, b"\x1b[C", b"+", b"\x1b[4~", b"\x1b[B", b"\x1b[6~", None, None, None, None, None, b"\x1b[23~", b"\x1b[24~", None, None, None, None, None, None, None, b"\n", None, b"/", None, None, None, b"\x1b[1~", b"\x1b[A", b"\x1b[5~", b"\x1b[D", b"\x1b[C", b"\x1b[4~", b"\x1b[B", b"\x1b[6~", b"\x1b[2~", b"\x1b[3~"]

Keystroke = namedtuple("Keystroke", ("key", "time"))


def read_csv(path):
    with open(path, newline="") as f:
        last_time = None
        for row in csv.DictReader(f):
            if (key := row["Key"]) == "Key":
                # multiple sessions concatenated into one file
                last_time = None
                continue

            if t := row.get("TimeNS"):
                time = int(t)
            elif t := row.get("TimeMicroseconds"):
                time = int(t) * 1000
            else:
                raise KeyError

            last_time = last_time or time
            yield Keystroke(int(key), time - last_time)


if len(sys.argv) != 2:
    print("usage: ./replay.py <keylogged.csv>")
else:
    for event in read_csv(sys.argv[1]):
        if event.key < len(KEYMAP) and KEYMAP[event.key]:
            sys.stdout.buffer.write(KEYMAP[event.key])
            sys.stdout.buffer.flush()
            time.sleep(0.01)

License: CC0-1.0