data.js

/* dialout data layer — turn raw conn_event records from the kernel into
 * normalized dials and aggregate them by process. The eBPF side
 * (dialout.bpf.c) fires once per outbound TCP connect via fexit/tcp_connect;
 * this module classifies each remote endpoint (private vs public, named
 * service, odd port) and groups dials under the process that made them.
 * Nothing here knows about colors, columns, or terminals — main.js and
 * dump.js share it. */

import { RingBuf } from "yeet:bpf";
import bpf from "./dialout.bpf.o";

const AF_INET = 2;
const AF_INET6 = 10;

/* Ports we consider ordinary for a box to dial out on. A *public* dial to
 * anything outside this set is flagged "odd" — not proof of trouble, but
 * the kind of egress worth a second look (a 4444, a 6667, a random high
 * port to the internet). Loopback/LAN dials are never flagged on port. */
const COMMON_EGRESS = new Set([
  22, 25, 53, 80, 110, 123, 143, 443, 465, 587, 853, 873, 990, 993, 995, 8080,
  8443,
]);

const SERVICES = {
  20: "ftp-data",
  21: "ftp",
  22: "ssh",
  23: "telnet",
  25: "smtp",
  53: "dns",
  67: "dhcp",
  80: "http",
  110: "pop3",
  111: "rpcbind",
  123: "ntp",
  135: "msrpc",
  143: "imap",
  179: "bgp",
  389: "ldap",
  443: "https",
  445: "smb",
  465: "smtps",
  514: "syslog",
  587: "submission",
  636: "ldaps",
  853: "dns-tls",
  873: "rsync",
  990: "ftps",
  993: "imaps",
  995: "pop3s",
  1080: "socks",
  1433: "mssql",
  1521: "oracle",
  1883: "mqtt",
  2049: "nfs",
  2181: "zookeeper",
  2375: "docker",
  2376: "docker-tls",
  2379: "etcd",
  2380: "etcd-peer",
  3000: "grafana",
  3128: "proxy",
  3306: "mysql",
  3389: "rdp",
  4222: "nats",
  5432: "postgres",
  5672: "amqp",
  5900: "vnc",
  6379: "redis",
  6443: "kube-api",
  6667: "irc",
  8080: "http-alt",
  8443: "https-alt",
  8883: "mqtt-tls",
  9000: "minio",
  9042: "cassandra",
  9092: "kafka",
  9200: "elastic",
  9300: "elastic",
  11211: "memcached",
  15672: "rabbitmq",
  27017: "mongodb",
};

function bytesOf(raw, n) {
  let u8;
  if (raw == null) u8 = new Uint8Array(n);
  else if (raw instanceof Uint8Array) u8 = raw;
  else if (Array.isArray(raw)) u8 = Uint8Array.from(raw);
  else if (typeof raw === "string") {
    u8 = new Uint8Array(raw.length);
    for (let i = 0; i < raw.length; i++) u8[i] = raw.charCodeAt(i) & 0xff;
  } else u8 = Uint8Array.from(Object.values(raw));
  return u8;
}

/* Collapse the longest run of all-zero groups into "::" (RFC 5952-ish). */
function compress6(parts) {
  let best = -1,
    bestLen = 0,
    run = -1,
    runLen = 0;
  for (let i = 0; i < parts.length; i++) {
    if (parts[i] === "0") {
      if (run < 0) run = i;
      runLen = i - run + 1;
      if (runLen > bestLen) {
        bestLen = runLen;
        best = run;
      }
    } else {
      run = -1;
      runLen = 0;
    }
  }
  if (bestLen < 2) return parts.join(":");
  return `${parts.slice(0, best).join(":")}::${parts.slice(best + bestLen).join(":")}`;
}

export function formatAddr(family, raw) {
  const b = bytesOf(raw, 16);
  if (family === AF_INET) return `${b[0]}.${b[1]}.${b[2]}.${b[3]}`;
  /* An IPv4-mapped v6 address (::ffff:a.b.c.d) is really a v4 dial. */
  if (v4MappedOffset(b) === 12) return `${b[12]}.${b[13]}.${b[14]}.${b[15]}`;
  const parts = [];
  for (let i = 0; i < 16; i += 2) parts.push(((b[i] << 8) | b[i + 1]).toString(16));
  return compress6(parts);
}

function v4MappedOffset(b) {
  for (let i = 0; i < 10; i++) if (b[i] !== 0) return -1;
  return b[10] === 0xff && b[11] === 0xff ? 12 : -1;
}

function scopeV4(o0, o1) {
  if (o0 === 127) return "loopback";
  if (o0 === 10) return "private";
  if (o0 === 172 && o1 >= 16 && o1 <= 31) return "private";
  if (o0 === 192 && o1 === 168) return "private";
  if (o0 === 169 && o1 === 254) return "linklocal";
  if (o0 === 100 && o1 >= 64 && o1 <= 127) return "cgnat";
  if (o0 >= 224 && o0 <= 239) return "multicast";
  if (o0 === 0) return "local";
  if (o0 === 255) return "broadcast";
  return "public";
}

/* Where the dial is headed, by destination address only. "public" is the
 * one that matters — it means the box reached past its own networks. */
export function classify(family, raw) {
  const b = bytesOf(raw, 16);
  if (family === AF_INET) return scopeV4(b[0], b[1]);
  if (v4MappedOffset(b) === 12) return scopeV4(b[12], b[13]);

  let allZero = true;
  for (let i = 0; i < 16; i++) if (b[i] !== 0) allZero = false;
  if (allZero) return "local";
  if (b[0] === 0 && b[1] === 0 && b[15] === 1) {
    let mid = true;
    for (let i = 2; i < 15; i++) if (b[i] !== 0) mid = false;
    if (mid) return "loopback";
  }
  if (b[0] === 0xfe && (b[1] & 0xc0) === 0x80) return "linklocal";
  if ((b[0] & 0xfe) === 0xfc) return "private"; /* fc00::/7 unique-local */
  if (b[0] === 0xff) return "multicast";
  return "public";
}

export function serviceName(port) {
  return SERVICES[port] ?? null;
}

export function normalize(e) {
  const family = e.family;
  const scope = classify(family, e.daddr);
  const isPublic = scope === "public";
  return {
    wall: Date.now(),
    pid: e.pid >>> 0,
    ppid: e.ppid >>> 0,
    uid: e.uid >>> 0,
    comm: e.comm || "?",
    pcomm: e.pcomm || "?",
    family,
    saddr: formatAddr(family, e.saddr),
    sport: e.sport & 0xffff,
    daddr: formatAddr(family, e.daddr),
    dport: e.dport & 0xffff,
    ret: e.ret | 0,
    ok: (e.ret | 0) === 0,
    scope,
    public: isPublic,
    unusual: isPublic && !COMMON_EGRESS.has(e.dport & 0xffff),
    service: serviceName(e.dport & 0xffff),
  };
}

const RECENT_CAP = 200;

/* Rolling aggregate of dials, grouped process -> remote endpoint. The
 * renderer reads snapshot()s of this; dump.js bypasses it and streams the
 * raw normalized dials instead. */
export class Tracker {
  constructor() {
    this.procs = new Map();
    this.hosts = new Set();
    this.recent = [];
    this.totals = { dials: 0, public: 0, unusual: 0, failed: 0 };
  }

  add(c) {
    this.totals.dials++;
    if (c.public) this.totals.public++;
    if (c.unusual) this.totals.unusual++;
    if (!c.ok) this.totals.failed++;
    if (c.scope === "public") this.hosts.add(c.daddr);

    this.recent.push(c);
    if (this.recent.length > RECENT_CAP) this.recent.shift();

    const pkey = `${c.pid}/${c.comm}`;
    let p = this.procs.get(pkey);
    if (!p) {
      p = {
        pid: c.pid,
        comm: c.comm,
        ppid: c.ppid,
        pcomm: c.pcomm,
        uid: c.uid,
        first: c.wall,
        last: c.wall,
        total: 0,
        public: 0,
        unusual: 0,
        failed: 0,
        endpoints: new Map(),
      };
      this.procs.set(pkey, p);
    }
    p.last = c.wall;
    p.total++;
    if (c.public) p.public++;
    if (c.unusual) p.unusual++;
    if (!c.ok) p.failed++;

    const ekey = `${c.daddr}:${c.dport}`;
    let ep = p.endpoints.get(ekey);
    if (!ep) {
      ep = {
        daddr: c.daddr,
        dport: c.dport,
        family: c.family,
        scope: c.scope,
        public: c.public,
        unusual: c.unusual,
        service: c.service,
        first: c.wall,
        last: c.wall,
        count: 0,
        lastRet: c.ret,
      };
      p.endpoints.set(ekey, ep);
    }
    ep.last = c.wall;
    ep.count++;
    ep.lastRet = c.ret;
    return c;
  }

  snapshot() {
    return {
      procs: [...this.procs.values()],
      recent: this.recent,
      hostCount: this.hosts.size,
      procCount: this.procs.size,
      totals: this.totals,
    };
  }
}

/* Load the probe, attach fexit/tcp_connect (auto-attached on start by its
 * SEC name — only network hooks need an explicit .attach), and stream
 * normalized dials to `onEvent`. Rejects with a plain Error carrying a
 * privilege hint if the daemon can't load or verify the program. */
export async function capture(onEvent, onError) {
  let control;
  try {
    control = await bpf
      .bind("events", { kind: "ringbuf", btf_struct: "conn_event" })
      .start();
  } catch (e) {
    const detail = e && (e.message || e.code) ? `${e.code ? `${e.code}: ` : ""}${e.message ?? ""}`.trim() : String(e);
    throw new Error(
      `Could not load the dialout eBPF probe${detail ? ` (${detail})` : ""}. ` +
        "It needs a kernel with BTF and the yeet daemon running with CAP_BPF " +
        "(it loads an fexit probe on tcp_connect).",
    );
  }

  const events = new RingBuf(control, "events");
  const sub = await events.subscribe(
    (rec) => onEvent(normalize(rec.conn_event ?? rec)),
    (err) => onError && onError(err),
  );

  return {
    async stop() {
      try {
        await sub.unsubscribe();
      } catch {}
      try {
        await control.stop();
      } catch {}
    },
  };
}