pi-car-companion/server/src/network-observer.ts

import { execFile } from "node:child_process";
import { chmod, readFile, rename, writeFile } from "node:fs/promises";
import { promisify } from "node:util";
import { z } from "zod";
import { networkActivitySchema } from "./network-activity.js";

const execFileAsync = promisify(execFile);
const sleep = (milliseconds: number) => new Promise((resolve) => setTimeout(resolve, milliseconds));
const config = z
  .object({
    NETWORK_ACTIVITY_PATH: z.string().default("/var/lib/pi-car-companion/network-activity.json"),
    NETWORK_HOTSPOT_INTERFACE: z.string().regex(/^[a-zA-Z0-9_.-]+$/).default("wlan0"),
    NETWORK_OBSERVER_INTERVAL_MS: z.coerce.number().int().min(1_000).max(30_000).default(2_000)
  })
  .parse(process.env);

type Activity = z.infer<typeof networkActivitySchema>;
type Client = Activity["clients"][number];
type DnsQuery = Activity["dnsQueries"][number];
type Flow = Activity["flows"][number];
let stopped = false;

async function execute(file: string, args: string[], maxBuffer = 256 * 1024): Promise<string> {
  const { stdout } = await execFileAsync(file, args, { timeout: 10_000, maxBuffer });
  return stdout;
}

async function clients(): Promise<Client[]> {
  const leasePath = `/var/lib/NetworkManager/dnsmasq-${config.NETWORK_HOTSPOT_INTERFACE}.leases`;
  const [leases, neighbors] = await Promise.all([
    readFile(leasePath, "utf8").catch(() => ""),
    execute("/usr/sbin/ip", ["neigh", "show", "dev", config.NETWORK_HOTSPOT_INTERFACE]).catch(() => "")
  ]);
  const states = new Map<string, { mac: string | null; state: string }>();
  for (const line of neighbors.trim().split("\n")) {
    const match = line.match(/^(\S+)(?:\s+lladdr\s+(\S+))?\s+(\S+)$/);
    if (match?.[1] && match[3]) states.set(match[1], { mac: match[2] ?? null, state: match[3] });
  }
  const result = new Map<string, Client>();
  for (const line of leases.trim().split("\n")) {
    const [expiry, mac, ip, hostname] = line.trim().split(/\s+/);
    if (!expiry || !mac || !ip) continue;
    const neighbor = states.get(ip);
    result.set(ip, {
      ipAddress: ip,
      macAddress: mac,
      hostname: hostname && hostname !== "*" ? hostname.slice(0, 128) : null,
      leaseExpiresAt: Number.isFinite(Number(expiry)) ? new Date(Number(expiry) * 1000).toISOString() : null,
      state: neighbor?.state ?? "LEASED",
      connected: Boolean(neighbor && !["FAILED", "INCOMPLETE"].includes(neighbor.state))
    });
  }
  for (const [ip, neighbor] of states) {
    if (!result.has(ip)) {
      result.set(ip, {
        ipAddress: ip,
        macAddress: neighbor.mac ?? "unknown",
        hostname: null,
        leaseExpiresAt: null,
        state: neighbor.state,
        connected: !["FAILED", "INCOMPLETE"].includes(neighbor.state)
      });
    }
  }
  return [...result.values()].sort((left, right) => Number(right.connected) - Number(left.connected) || left.ipAddress.localeCompare(right.ipAddress));
}

async function dnsQueries(): Promise<{ available: boolean; queries: DnsQuery[] }> {
  try {
    const output = await execute(
      "/usr/bin/journalctl",
      ["--unit", "NetworkManager.service", "--since", "-15 minutes", "--output", "json", "--no-pager"],
      1024 * 1024
    );
    const queries: DnsQuery[] = [];
    for (const line of output.trim().split("\n")) {
      try {
        const entry = JSON.parse(line) as { MESSAGE?: unknown; __REALTIME_TIMESTAMP?: unknown };
        if (typeof entry.MESSAGE !== "string") continue;
        const match = entry.MESSAGE.match(/query\[([^\]]+)]\s+(\S+)\s+from\s+(10\.42\.0\.\d+)/);
        if (!match?.[1] || !match[2] || !match[3]) continue;
        const microseconds = Number(entry.__REALTIME_TIMESTAMP);
        queries.push({
          timestamp: Number.isFinite(microseconds) ? new Date(Math.floor(microseconds / 1000)).toISOString() : new Date().toISOString(),
          clientAddress: match[3],
          type: match[1].slice(0, 16),
          name: match[2].slice(0, 253)
        });
      } catch {
        // Ignore unrelated or malformed journal records.
      }
    }
    return { available: true, queries: queries.slice(-250).reverse() };
  } catch {
    return { available: false, queries: [] };
  }
}

function value(line: string, name: string): string | null {
  return line.match(new RegExp(`(?:^|\\s)${name}=([^\\s]+)`))?.[1] ?? null;
}

function numericValue(line: string, name: string): number | null {
  const raw = value(line, name);
  if (raw === null) return null;
  const parsed = Number(raw);
  return Number.isInteger(parsed) && parsed >= 0 ? parsed : null;
}

async function flows(): Promise<{ available: boolean; flows: Flow[] }> {
  try {
    const output = await execute("/usr/sbin/conntrack", ["-L"], 1024 * 1024);
    const parsed: Flow[] = [];
    for (const line of output.trim().split("\n")) {
      const protocol = line.match(/^(tcp|udp)\s/)?.[1] as "tcp" | "udp" | undefined;
      const sourceAddress = value(line, "src");
      const destinationAddress = value(line, "dst");
      const sourcePort = numericValue(line, "sport");
      const destinationPort = numericValue(line, "dport");
      if (!protocol || !sourceAddress?.startsWith("10.42.0.") || !destinationAddress || sourcePort === null || destinationPort === null) continue;
      const state = protocol === "tcp" ? line.match(/^tcp\s+\d+\s+\d+\s+(\S+)/)?.[1] ?? "UNKNOWN" : "ACTIVE";
      const packets = numericValue(line, "packets");
      const bytes = numericValue(line, "bytes");
      parsed.push({
        protocol,
        state,
        sourceAddress,
        sourcePort,
        destinationAddress,
        destinationPort,
        packets,
        bytes
      });
    }
    return { available: true, flows: parsed.slice(0, 150) };
  } catch {
    return { available: false, flows: [] };
  }
}

async function collect(): Promise<Activity> {
  const [clientList, dns, activeFlows] = await Promise.all([clients(), dnsQueries(), flows()]);
  const activeClients = new Set(activeFlows.flows.map((flow) => flow.sourceAddress));
  const clientsWithFlowState = clientList.map((client) =>
    !client.connected && activeClients.has(client.ipAddress) ? { ...client, connected: true, state: "ACTIVE" } : client
  );
  const messages: string[] = [];
  if (!dns.available) messages.push("DNS query metadata is unavailable; restart the hotspot after installing its dnsmasq configuration.");
  if (!activeFlows.available) messages.push("Active flows are unavailable because conntrack is not installed or accessible.");
  return networkActivitySchema.parse({
    collectedAt: new Date().toISOString(),
    clients: clientsWithFlowState,
    dnsQueries: dns.queries,
    flows: activeFlows.flows,
    dnsAvailable: dns.available,
    flowsAvailable: activeFlows.available,
    messages
  });
}

async function publish(activity: Activity): Promise<void> {
  const temporary = `${config.NETWORK_ACTIVITY_PATH}.tmp`;
  await writeFile(temporary, `${JSON.stringify(activity)}\n`, { mode: 0o644 });
  await chmod(temporary, 0o644);
  await rename(temporary, config.NETWORK_ACTIVITY_PATH);
}

process.on("SIGINT", () => { stopped = true; });
process.on("SIGTERM", () => { stopped = true; });
while (!stopped) {
  try {
    await publish(await collect());
  } catch (error) {
    console.error(error instanceof Error ? error.message : "Network activity collection failed");
  }
  await sleep(config.NETWORK_OBSERVER_INTERVAL_MS);
}