mirror of
https://github.com/kiddin9/op-packages.git
synced 2026-10-28 06:22:51 +08:00
2546 lines
91 KiB
Plaintext
Executable File
2546 lines
91 KiB
Plaintext
Executable File
#!/usr/bin/ucode
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
// Copyright 2026 VolanticSystems
|
|
//
|
|
// appflowd — aggregate Netify Agent (netifyd) DPI flow events into live
|
|
// per-application / per-device / per-category traffic statistics and serve
|
|
// them over ubus. See docs/DESIGN.md; section references below point there.
|
|
//
|
|
// Data path (§3):
|
|
// netifyd --unix stream, length-framed JSON--> appflowd --ubus--> LuCI
|
|
//
|
|
// Everything that counts a byte goes through account() (§7.1), which is the
|
|
// single seam historical accounting will tap into.
|
|
|
|
import * as fs from "fs";
|
|
import * as uloop from "uloop";
|
|
import * as ubus from "ubus";
|
|
import * as socket from "socket";
|
|
import { cursor } from "uci";
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Constants — the memory bounds of §3.3 live here. Every table in the
|
|
* daemon is capped; nothing grows with uptime or traffic volume.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
const NAME = "appflowd";
|
|
const VERSION = "1.0";
|
|
|
|
const RX_CHUNK = 16384; // bytes per recv()
|
|
const RX_MAX = 262144; // partial-line buffer ceiling; beyond this the
|
|
// peer is not speaking our protocol -> resync
|
|
const RX_DRAIN_MAX = 64; // recv() calls per readable event (fairness)
|
|
|
|
const AGG_MAX = 384; // max aggregates per class (apps/devices/cats)
|
|
const APP_DEV_MAX = 64; // devices tracked per application (LRU)
|
|
const APP_HOST_MAX = 16; // hostnames tracked per application (LRU)
|
|
const NAME_MAX = 8192; // id->tag map ceiling (nDPI catalogue is ~2.5k)
|
|
const TAG_MAX = 64; // max stored length of any detection tag
|
|
const UNKNOWN_TYPE_MAX = 16; // distinct unhandled event types remembered
|
|
const EVENT_TYPE_MAX = 24; // distinct event types counted in status
|
|
const FLOWS_REPLY_MAX = 200; // hard ceiling for the flows[] ubus reply
|
|
const LIST_REPLY_MAX = 200; // hard ceiling for apps[]/devices[] replies
|
|
|
|
// Trailing-hour time series (§7.2 buckets, GL-parity `stats` method):
|
|
// 12 slots x 5 minutes. Slot numbers are absolute (epoch_ms / SERIES_MS), so
|
|
// a slot is self-describing and stale slots need no eager clearing.
|
|
const SERIES_SLOTS = 12;
|
|
const SERIES_MS = 300000;
|
|
// Per (app x device) we only need trailing-hour *totals*, not a series, so a
|
|
// coarser 4 x 15 min ring is used: same code path, quarter the memory. The
|
|
// window it reports is therefore 45-60 minutes rather than exactly 60.
|
|
const DEVSER_SLOTS = 4;
|
|
const DEVSER_MS = 900000;
|
|
|
|
const PRUNE_MS = 15000; // idle-flow / aggregate sweep interval
|
|
const LEASE_TTL_MS = 30000; // DHCP lease file re-read throttle (§4)
|
|
const LOG_REPEAT_MS = 60000; // identical-message log suppression window
|
|
const RECONNECT_MIN_MS = 1000;
|
|
const RECONNECT_MAX_MS = 30000;
|
|
|
|
const CATEGORY_FILE = "/etc/netify.d/netify-categories.json";
|
|
const LEASE_FILE = "/tmp/dhcp.leases";
|
|
const STATUS_FILE = "/var/run/netifyd/status.json";
|
|
const CONNTRACK_FILE = "/proc/net/nf_conntrack";
|
|
|
|
const CT_TTL_MS = 2000; // reuse one conntrack snapshot this long (3.2)
|
|
const CT_RETRY_MS = 60000; // re-probe an unreadable conntrack this often
|
|
const CT_FORCE_MIN_MS = 250; // floor between miss-driven refreshes (3.2)
|
|
const CT_MAX = 8192; // entries kept from one snapshot (memory bound, 3.3)
|
|
|
|
const CFG_DEFAULTS = {
|
|
socket_path: "/var/run/netifyd/netifyd.sock",
|
|
conntrack_path: "/proc/net/nf_conntrack",
|
|
flow_max: 4096,
|
|
rate_window: 30,
|
|
top_n: 10,
|
|
idle_timeout: 300,
|
|
ai_breakout: 1
|
|
};
|
|
|
|
// Conntrack snapshot used to resolve NAT twins (3.2). `map` is keyed on the
|
|
// REPLY tuple, which is the view netifyd's external capture reports, and holds
|
|
// true when that flow was source-translated. `avail` is null until first
|
|
// probed, then true/false; false is re-probed every CT_RETRY_MS so a conntrack
|
|
// module loaded after us is picked up without needing a restart.
|
|
let conntrack = { at: 0, probed: 0, avail: null, map: {}, entries: 0 };
|
|
|
|
// Any event type matching this and carrying flow.digest finalises a flow.
|
|
// netifyd 4.4.7 emits "flow_purge"; the regex keeps a renamed or additional
|
|
// event working without a code change (the dispatcher below is data-driven).
|
|
const PURGE_RE = regexp("(purge|expire|delete|remove|terminat|close)", "i");
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* State
|
|
* ------------------------------------------------------------------ */
|
|
|
|
let cfg = {};
|
|
|
|
let sock = null; // socket handle to netifyd
|
|
let sock_fd = null; // uloop fd handle
|
|
let rxbuf = ""; // partial line carry-over
|
|
let connected = false;
|
|
let ever_connected = false; // distinguishes the first connect from reconnects
|
|
let backoff_ms = RECONNECT_MIN_MS;
|
|
let retry_timer = null;
|
|
|
|
let flows = {}; // digest -> flow record
|
|
let nflows = 0;
|
|
// True once any internal (LAN-bridge) capture has been seen. The single-count
|
|
// shadow rule only trusts "the internal capture already counted this" when
|
|
// internal capture is actually active; on an external-only netifyd setup it
|
|
// must not shadow client flows or the dashboard would read near-zero.
|
|
let saw_internal = false;
|
|
|
|
let apps = {}, devs = {}, cats = {};
|
|
let totals = null; // aggregate covering everything
|
|
|
|
let names = { app: {}, proto: {} }; // id -> netifyd tag (definitions event)
|
|
let catnames = { app: {}, proto: {} }; // id -> category tag (categories file)
|
|
|
|
let leases = { at: 0, mtime: -1, by_mac: {}, by_ip: {} };
|
|
let router = { macs: {}, ips: {} }; // this router's own L2/L3 identities
|
|
let agent = { version: null, uptime: 0, flow_count: 0, ts: 0 };
|
|
|
|
let stats = {
|
|
events: 0, frames: 0, malformed: 0, unknown: 0, unknown_types: {},
|
|
by_type: {}, reconnects: 0, connect_fail: 0, purges: 0,
|
|
flows_seen: 0, flows_late: 0, flows_pruned: 0, flows_dropped: 0,
|
|
// Flows relabelled by the AI hostname table. Observable so a test can
|
|
// assert the feature fired rather than inferring it from a label.
|
|
ai_labeled: 0,
|
|
// Flows evicted because the table was AT ITS CAP, as opposed to evicted for
|
|
// going idle. The two used to share flows_pruned, where cap pressure was
|
|
// invisible underneath ordinary housekeeping.
|
|
//
|
|
// This exists because flows_dropped, which reads like the counter for
|
|
// "flows we could not keep", CANNOT BE MOVED BY ANY INPUT. flow_new()
|
|
// increments it only if the table is still full after prune(now, true), and
|
|
// that prune unconditionally sheds int(flow_max/10)+1 entries, so it always
|
|
// frees space and the fall-through never happens. Measured 2026-08-26 by
|
|
// tests/protocol-suite.sh: 200 distinct digests against flow_max=64 gave
|
|
// tracked=60 pruned=143 dropped=0. An operator reading `dropped: 0`
|
|
// concluded nothing was being lost while flows were being shed constantly.
|
|
// Same shape as aggs_refused below, and found the same way.
|
|
//
|
|
// flows_dropped is kept as the genuine last-resort backstop it always was.
|
|
// flows_shed is the number an operator should actually watch.
|
|
flows_shed: 0,
|
|
shadowed: 0, aggs_pruned: 0, rx_bytes: 0, resyncs: 0, sweeps: 0,
|
|
// Times an aggregate table was full of LIVE entries and a new key was
|
|
// therefore not tracked at all. Distinct from aggs_pruned, which counts
|
|
// evictions of idle entries and is normal housekeeping. A non-zero value
|
|
// here means the dashboard is silently incomplete.
|
|
aggs_refused: 0,
|
|
no_total: 0, last_event_ms: 0, last_reset_ms: 0,
|
|
// NAT twin resolution outcomes (3.2): twins shadowed on conntrack evidence,
|
|
// router-originated flows kept, and flows conntrack could not decide, which
|
|
// are shadowed on the safe default.
|
|
ct_twins: 0, ct_local: 0, ct_unresolved: 0,
|
|
// Purge-only flows rescued (see flow_update). Expected to stay 0 on
|
|
// netifyd 4.4.7; a non-zero value means that assumption has changed.
|
|
purge_only: 0,
|
|
// Byte conservation (§2.5): reported == attributed + shadowed, or
|
|
// something is leaking. Both silent byte-loss defects found in this
|
|
// daemon would have been visible here from the start.
|
|
bytes_reported: 0, bytes_attributed: 0, bytes_shadowed: 0
|
|
};
|
|
|
|
let start_ms = 0;
|
|
let start_mono = 0;
|
|
let log_seen = {};
|
|
// The published ubus object must stay referenced: dropping the last reference
|
|
// lets ucode collect it, which unregisters the object from ubus.
|
|
let ubus_obj = null;
|
|
|
|
// Forward declarations. ucode binds an identifier at compile time: one that
|
|
// is still undeclared resolves to a global, while `function f()` further down
|
|
// declares a local, so the call would find null. connect/schedule_reconnect
|
|
// are mutually recursive and prune() is used by the flow table above its own
|
|
// definition, so those four are declared here and assigned below.
|
|
let prune, connect, disconnect, schedule_reconnect;
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Small helpers
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function now_ms() {
|
|
let c = clock();
|
|
return c[0] * 1000 + int(c[1] / 1000000);
|
|
}
|
|
|
|
function mono_ms() {
|
|
let c = clock(true);
|
|
return c[0] * 1000 + int(c[1] / 1000000);
|
|
}
|
|
|
|
function num(v, def) {
|
|
if (v == null)
|
|
return def;
|
|
let n = int(v);
|
|
return (type(n) == "int") ? n : def;
|
|
}
|
|
|
|
function clampi(v, lo, hi) {
|
|
return (v < lo) ? lo : ((v > hi) ? hi : v);
|
|
}
|
|
|
|
// procd feeds our stdout to logd as daemon.info and our stderr as daemon.err
|
|
// (see the init script), so route by severity and flush explicitly -- stdout
|
|
// is block buffered when it is a pipe.
|
|
function logmsg(level, fmt, ...args) {
|
|
let out = (level == "error") ? fs.stderr : fs.stdout;
|
|
out.write(sprintf("[%s] %s\n", level, sprintf(fmt, ...args)));
|
|
out.flush();
|
|
}
|
|
|
|
// Log at most one identical message per LOG_REPEAT_MS. Reconnect churn while
|
|
// netifyd is down must not turn into log spam.
|
|
function logonce(key, level, fmt, ...args) {
|
|
let now = mono_ms();
|
|
if (log_seen[key] && (now - log_seen[key]) < LOG_REPEAT_MS)
|
|
return;
|
|
if (length(log_seen) > 32)
|
|
log_seen = {};
|
|
log_seen[key] = now;
|
|
logmsg(level, fmt, ...args);
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Configuration (uci: /etc/config/appflow)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function cfg_load() {
|
|
let s = {};
|
|
try {
|
|
s = cursor().get_all("appflow", "main") ?? {};
|
|
} catch (e) {
|
|
logmsg("warn", "uci: %s — using defaults", e);
|
|
}
|
|
|
|
/* User-supplied hostname mappings, from `config hostmap` sections.
|
|
*
|
|
* Each adds or OVERRIDES one entry in the built-in table, so this is both
|
|
* the way to teach appflow a service it does not know and the way to
|
|
* correct one it gets wrong. Consulted before the built-in table.
|
|
*
|
|
* config hostmap
|
|
* option suffix 'torproject.org'
|
|
* option name 'Tor'
|
|
* option category 'Privacy'
|
|
* option strong '0'
|
|
*
|
|
* A malformed section is skipped and named in the log rather than
|
|
* silently ignored: a typo in a suffix produces a rule that matches
|
|
* nothing, which looks identical to the feature not working.
|
|
*/
|
|
let hostmap = {};
|
|
let bad = 0;
|
|
try {
|
|
cursor().foreach("appflow", "hostmap", function(h) {
|
|
let suf = lc(trim(h.suffix ?? ""));
|
|
let nm = trim(h.name ?? "");
|
|
|
|
/* A suffix must contain a dot, or it can never match: the matcher
|
|
* rebuilds the last N labels of a hostname and compares, so a
|
|
* single label matches nothing. */
|
|
if (!length(suf) || !length(nm) || index(suf, ".") < 0) {
|
|
bad++;
|
|
logmsg("warn", "hostmap: skipping entry with suffix %s name %s",
|
|
suf ?? "(empty)", nm ?? "(empty)");
|
|
|
|
return;
|
|
}
|
|
|
|
hostmap[suf] = [ nm,
|
|
length(trim(h.category ?? "")) ? trim(h.category) : "other",
|
|
num(h.strong, 0) != 0 ];
|
|
});
|
|
} catch (e) {
|
|
logmsg("warn", "hostmap: %s", e);
|
|
}
|
|
if (length(hostmap))
|
|
logmsg("info", "hostmap: %d user entr%s loaded%s",
|
|
length(hostmap), length(hostmap) == 1 ? "y" : "ies",
|
|
bad ? sprintf(", %d skipped", bad) : "");
|
|
|
|
cfg = {
|
|
socket_path: length(s.socket_path ?? "") ? s.socket_path : CFG_DEFAULTS.socket_path,
|
|
conntrack_path: length(s.conntrack_path ?? "") ? s.conntrack_path : CFG_DEFAULTS.conntrack_path,
|
|
flow_max: clampi(num(s.flow_max, CFG_DEFAULTS.flow_max), 64, 65536),
|
|
rate_window: clampi(num(s.rate_window, CFG_DEFAULTS.rate_window), 5, 600),
|
|
top_n: clampi(num(s.top_n, CFG_DEFAULTS.top_n), 1, 100),
|
|
idle_timeout: clampi(num(s.idle_timeout, CFG_DEFAULTS.idle_timeout), 15, 86400),
|
|
// Absent means ON. A dashboard feature that defaults to off is a
|
|
// feature nobody sees, and this one only ever relabels flows netifyd
|
|
// had no answer for.
|
|
ai_breakout: num(s.ai_breakout, CFG_DEFAULTS.ai_breakout) != 0,
|
|
hostmap: hostmap
|
|
};
|
|
cfg.rate_ms = cfg.rate_window * 1000;
|
|
cfg.idle_ms = cfg.idle_timeout * 1000;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Time-bucketed accumulators (§7.2)
|
|
*
|
|
* A ring of `n` slots, each tagged with its absolute slot number. Writes are
|
|
* O(1) and self-invalidating: a slot whose tag is not in the trailing window
|
|
* is simply not counted, so no eager rotation is needed for correctness. The
|
|
* periodic sweep below still clears stale slots so their bytes cannot be
|
|
* resurrected by a clock step and so status output stays cheap.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function ring_new(n) {
|
|
let r = { s: [], d: [], u: [] };
|
|
for (let i = 0; i < n; i++) {
|
|
push(r.s, -1);
|
|
push(r.d, 0);
|
|
push(r.u, 0);
|
|
}
|
|
return r;
|
|
}
|
|
|
|
function ring_add(r, slot, n, up, down) {
|
|
let i = slot % n;
|
|
if (r.s[i] != slot) {
|
|
r.s[i] = slot;
|
|
r.d[i] = 0;
|
|
r.u[i] = 0;
|
|
}
|
|
r.d[i] += down;
|
|
r.u[i] += up;
|
|
}
|
|
|
|
function ring_total(r, slot, n) {
|
|
let d = 0, u = 0;
|
|
for (let i = 0; i < n; i++) {
|
|
if (r.s[i] > (slot - n) && r.s[i] <= slot) {
|
|
d += r.d[i];
|
|
u += r.u[i];
|
|
}
|
|
}
|
|
return [ d, u ];
|
|
}
|
|
|
|
function ring_series(r, slot, n, slot_ms) {
|
|
let out = [];
|
|
for (let k = n - 1; k >= 0; k--) {
|
|
let s = slot - k, i = s % n, ok = (r.s[i] == s);
|
|
push(out, {
|
|
ts: s * slot_ms,
|
|
download: ok ? r.d[i] : 0,
|
|
upload: ok ? r.u[i] : 0
|
|
});
|
|
}
|
|
return out;
|
|
}
|
|
|
|
function ring_sweep(r, slot, n) {
|
|
for (let i = 0; i < n; i++)
|
|
if (r.s[i] >= 0 && (r.s[i] <= (slot - n) || r.s[i] > slot)) {
|
|
r.s[i] = -1;
|
|
r.d[i] = 0;
|
|
r.u[i] = 0;
|
|
}
|
|
}
|
|
|
|
function ring_clear(r, n) {
|
|
for (let i = 0; i < n; i++) {
|
|
r.s[i] = -1;
|
|
r.d[i] = 0;
|
|
r.u[i] = 0;
|
|
}
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Aggregates
|
|
*
|
|
* Shape is {cur, buckets} per §7.2: `cur` is the session accumulator plus a
|
|
* two-bucket sliding window used for the live rate, `buckets` is the hour-long
|
|
* time series. History support (§7) extends `buckets`, nothing else.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function agg_new(kind, key, name, now) {
|
|
return {
|
|
kind, key, name,
|
|
cur: { up: 0, down: 0, flows: 0, first: now, last: 0 },
|
|
// Sliding rate window: bytes land in the open bucket; when it ages past
|
|
// rate_window it becomes the closed bucket and a fresh one opens. The
|
|
// reported rate spans closed+open, i.e. 1x-2x the window. Chosen over
|
|
// an EWMA because netifyd delivers flow_stats in ~15 s bursts: an EWMA
|
|
// tuned tight enough to be live would oscillate between spike and zero,
|
|
// while this integrates each burst over a known interval. Cost is O(1)
|
|
// memory and no per-second timer.
|
|
rate: { t0: now, up: 0, down: 0, pdt: 0, pup: 0, pdown: 0 },
|
|
buckets: ring_new(SERIES_SLOTS)
|
|
};
|
|
}
|
|
|
|
function agg_roll(a, now) {
|
|
let span = now - a.rate.t0;
|
|
if (span >= cfg.rate_ms) {
|
|
a.rate.pdt = span;
|
|
a.rate.pup = a.rate.up;
|
|
a.rate.pdown = a.rate.down;
|
|
a.rate.t0 = now;
|
|
a.rate.up = 0;
|
|
a.rate.down = 0;
|
|
}
|
|
}
|
|
|
|
// Pure: computes the rate WITHOUT rolling the window, so a read never mutates
|
|
// state. Two clients polling at different intervals (or a second browser tab)
|
|
// must see identical numbers for identical traffic; the window is advanced only
|
|
// by agg_bump and housekeeping. Rolls are applied virtually here.
|
|
function agg_rate(a, now) {
|
|
let span = now - a.rate.t0;
|
|
let pdt, pup, pdown, cspan, cup, cdown;
|
|
if (span >= cfg.rate_ms) { // window would have rolled
|
|
pdt = span; pup = a.rate.up; pdown = a.rate.down;
|
|
cspan = 0; cup = 0; cdown = 0;
|
|
} else {
|
|
pdt = a.rate.pdt; pup = a.rate.pup; pdown = a.rate.pdown;
|
|
cspan = span; cup = a.rate.up; cdown = a.rate.down;
|
|
}
|
|
// Floor the denominator at one window so a single burst cannot be reported
|
|
// as an impossibly high instantaneous rate.
|
|
let denom = pdt + cspan;
|
|
if (denom < cfg.rate_ms)
|
|
denom = cfg.rate_ms;
|
|
return [
|
|
int((pup + cup) * 1000 / denom),
|
|
int((pdown + cdown) * 1000 / denom)
|
|
];
|
|
}
|
|
|
|
function agg_bump(a, up, down, now, slot) {
|
|
agg_roll(a, now);
|
|
a.cur.up += up;
|
|
a.cur.down += down;
|
|
a.cur.last = now;
|
|
a.rate.up += up;
|
|
a.rate.down += down;
|
|
ring_add(a.buckets, slot, SERIES_SLOTS, up, down);
|
|
}
|
|
|
|
function agg_reset(a, now) {
|
|
a.cur.up = 0;
|
|
a.cur.down = 0;
|
|
a.cur.first = now;
|
|
a.cur.last = 0;
|
|
a.rate = { t0: now, up: 0, down: 0, pdt: 0, pup: 0, pdown: 0 };
|
|
ring_clear(a.buckets, SERIES_SLOTS);
|
|
}
|
|
|
|
// Evict the least recently active zero-flow aggregate when a class is full.
|
|
function agg_make(tbl, kind, key, name, now) {
|
|
let a = tbl[key];
|
|
if (a)
|
|
return a;
|
|
|
|
if (length(tbl) >= AGG_MAX) {
|
|
let victim = null, oldest = now + 1;
|
|
for (let k in tbl) {
|
|
let c = tbl[k];
|
|
if (c.cur.flows > 0)
|
|
continue;
|
|
if (c.cur.last < oldest) {
|
|
oldest = c.cur.last;
|
|
victim = k;
|
|
}
|
|
}
|
|
if (victim == null) {
|
|
// Every slot is backed by a live flow: refuse to grow (§3.3).
|
|
// Bytes still land in the totals, so the summary stays truthful.
|
|
// Counted, not just logged. logonce() fires ONCE for the life of
|
|
// the process, so without this an operator polling status sees
|
|
// `devices: 384` and no sign that anything is being dropped -- the
|
|
// table looks healthy precisely when it has stopped accepting the
|
|
// device the user is looking for. A security panel on 2026-08-25
|
|
// converged on this table being fillable by a LAN attacker
|
|
// (confirmed: six spoofed MACs produced six aggregates on
|
|
// hardware). Evicting live data is already refused above, which is
|
|
// the right call; what was missing was any way to SEE the refusal.
|
|
stats.aggs_refused++;
|
|
logonce("aggfull", "warn", "%s table full (%d), not tracking %s",
|
|
kind, AGG_MAX, key);
|
|
return null;
|
|
}
|
|
delete tbl[victim];
|
|
stats.aggs_pruned++;
|
|
}
|
|
|
|
a = agg_new(kind, key, name, now);
|
|
tbl[key] = a;
|
|
return a;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Per-application breakdowns (bounded): device totals and hostnames.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function detail_dev(a, fr, up, down, now, dslot) {
|
|
if (!a.devs)
|
|
a.devs = {};
|
|
let e = a.devs[fr.dev_key];
|
|
if (!e) {
|
|
if (length(a.devs) >= APP_DEV_MAX) {
|
|
// Seed from the first candidate, not from `now`: a backwards clock
|
|
// step (routine on an RTC-less router after its first NTP sync) can
|
|
// push every stored `last` past now, which would leave victim null
|
|
// and grow the map unbounded.
|
|
let victim = null, oldest = null;
|
|
for (let k in a.devs)
|
|
if (oldest == null || a.devs[k].last < oldest) {
|
|
oldest = a.devs[k].last;
|
|
victim = k;
|
|
}
|
|
if (victim == null)
|
|
return; // cannot make room; do not exceed the cap
|
|
delete a.devs[victim];
|
|
}
|
|
e = { mac: fr.dev_mac, ip: fr.dev_ip, last: now, ring: ring_new(DEVSER_SLOTS) };
|
|
a.devs[fr.dev_key] = e;
|
|
}
|
|
if (fr.dev_ip)
|
|
e.ip = fr.dev_ip;
|
|
e.last = now;
|
|
ring_add(e.ring, dslot, DEVSER_SLOTS, up, down);
|
|
}
|
|
|
|
function detail_host(a, fr, up, down, now) {
|
|
if (!fr.host)
|
|
return;
|
|
if (!a.hosts)
|
|
a.hosts = {};
|
|
let e = a.hosts[fr.host];
|
|
if (!e) {
|
|
if (length(a.hosts) >= APP_HOST_MAX) {
|
|
let victim = null, oldest = null;
|
|
for (let k in a.hosts)
|
|
if (oldest == null || a.hosts[k].last < oldest) {
|
|
oldest = a.hosts[k].last;
|
|
victim = k;
|
|
}
|
|
if (victim == null)
|
|
return;
|
|
delete a.hosts[victim];
|
|
}
|
|
// Same trailing-window ring as the device breakdown, so the two tables
|
|
// in the detail drawer report over the same interval and stay comparable.
|
|
e = { last: now, ring: ring_new(DEVSER_SLOTS) };
|
|
a.hosts[fr.host] = e;
|
|
}
|
|
e.last = now;
|
|
ring_add(e.ring, int(now / DEVSER_MS), DEVSER_SLOTS, up, down);
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Persistence interface (§7.3)
|
|
*
|
|
* v1 ships this dormant on purpose. A history implementation (RAM-rotating
|
|
* file, collectd exec, sqlite on USB) plugs in here without touching the
|
|
* accounting path. Flash-wear rule: coarse buckets only, never per sample.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
const store = {
|
|
flush: function(_aggregates, _slot) {
|
|
// no-op in v1
|
|
}
|
|
};
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* THE accounting choke point (§7.1)
|
|
*
|
|
* Every byte appflowd ever counts passes through here, exactly once.
|
|
* `d_local` / `d_other` are netifyd's per-update directional deltas; the flow
|
|
* record knows which of the two endpoints is the attributed device, so:
|
|
* upload = bytes sent BY the device (device side -> peer)
|
|
* download = bytes sent TO the device (peer -> device side)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function account(fr, d_local, d_other, ts) {
|
|
if (fr.shadow) { // deduplicated NAT twin (§3.2)
|
|
// Deliberately dropped, not lost. Counted so that
|
|
// bytes_reported == bytes_attributed + bytes_shadowed holds, and any
|
|
// residual is a real leak rather than the dedup doing its job.
|
|
let s = d_local + d_other;
|
|
if (s > 0)
|
|
stats.bytes_shadowed += s;
|
|
return;
|
|
}
|
|
|
|
let up = fr.dev_local ? d_local : d_other;
|
|
let down = fr.dev_local ? d_other : d_local;
|
|
if (up < 0) up = 0;
|
|
if (down < 0) down = 0;
|
|
|
|
stats.bytes_attributed += up + down;
|
|
fr.up += up;
|
|
fr.down += down;
|
|
fr.seen = ts;
|
|
|
|
if (up == 0 && down == 0)
|
|
return;
|
|
|
|
let slot = int(ts / SERIES_MS);
|
|
let dslot = int(ts / DEVSER_MS);
|
|
|
|
agg_bump(totals, up, down, ts, slot);
|
|
if (fr.a) {
|
|
agg_bump(fr.a, up, down, ts, slot);
|
|
detail_dev(fr.a, fr, up, down, ts, dslot);
|
|
detail_host(fr.a, fr, up, down, ts);
|
|
}
|
|
if (fr.d)
|
|
agg_bump(fr.d, up, down, ts, slot);
|
|
if (fr.c)
|
|
agg_bump(fr.c, up, down, ts, slot);
|
|
|
|
// History (§7) taps in exactly here: the buckets these bumps fill are what
|
|
// store.flush() drains once a persistent implementation exists.
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Naming (§2.3)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
// Post-capitalization fixes for tokens where naive Title Case reads wrong.
|
|
const BRAND_CAPS = {
|
|
"Openwrt": "OpenWrt", "Dns": "DNS", "Http": "HTTP", "Https": "HTTPS",
|
|
"Ssl": "SSL", "Tls": "TLS", "Ssh": "SSH", "Ntp": "NTP", "Vpn": "VPN",
|
|
"Cdn": "CDN", "Voip": "VoIP", "Tv": "TV", "Api": "API", "Ip": "IP",
|
|
"Iot": "IoT", "Ai": "AI", "Ads": "Ads",
|
|
// Brands whose own capitalization is word-internal, which single-letter
|
|
// Title Case cannot produce.
|
|
"Youtube": "YouTube", "Github": "GitHub", "Gitlab": "GitLab",
|
|
"Whatsapp": "WhatsApp", "Tiktok": "TikTok", "Paypal": "PayPal",
|
|
"Icloud": "iCloud", "Iphone": "iPhone", "Deepl": "DeepL",
|
|
"Openai": "OpenAI", "Openvpn": "OpenVPN", "Wireguard": "WireGuard",
|
|
"Bittorrent": "BitTorrent", "Speedtest": "Speedtest", "Steamos": "SteamOS"
|
|
};
|
|
|
|
function title_case(s) {
|
|
let out = [];
|
|
for (let w in split(s, " ")) {
|
|
let t = length(w) ? (uc(substr(w, 0, 1)) + substr(w, 1)) : w;
|
|
push(out, BRAND_CAPS[t] ?? t);
|
|
}
|
|
return join(" ", out);
|
|
}
|
|
|
|
// Netify application tags look like "netify.google-ads"; nDPI protocol tags
|
|
// are already display-styled ("HTTP/S", "ZOOM", "WireGuard") and are kept
|
|
// verbatim.
|
|
function display_name(tag) {
|
|
if (!length(tag ?? ""))
|
|
return null;
|
|
if (substr(tag, 0, 7) != "netify.")
|
|
return tag;
|
|
let slug = substr(tag, 7);
|
|
slug = replace(slug, /[-_.]/g, " ");
|
|
return title_case(slug);
|
|
}
|
|
|
|
function load_categories() {
|
|
let raw;
|
|
try {
|
|
raw = fs.readfile(CATEGORY_FILE);
|
|
} catch (e) {
|
|
raw = null;
|
|
}
|
|
if (!raw) {
|
|
logonce("cats", "warn", "category names unavailable: %s", CATEGORY_FILE);
|
|
return;
|
|
}
|
|
|
|
let obj;
|
|
try {
|
|
obj = json(raw);
|
|
} catch (e) {
|
|
logmsg("warn", "%s: parse error: %s", CATEGORY_FILE, e);
|
|
return;
|
|
}
|
|
|
|
// The file maps tag -> id; we need id -> tag. The far larger
|
|
// application_index / protocol_index member lists are deliberately not
|
|
// retained: flow events carry the category id directly.
|
|
for (let t in obj.application_tag_index ?? {})
|
|
catnames.app[obj.application_tag_index[t]] = t;
|
|
for (let t in obj.protocol_tag_index ?? {})
|
|
catnames.proto[obj.protocol_tag_index[t]] = t;
|
|
}
|
|
|
|
function load_leases(now) {
|
|
if (leases.at && (now - leases.at) < LEASE_TTL_MS)
|
|
return;
|
|
leases.at = now;
|
|
|
|
let st;
|
|
try {
|
|
st = fs.stat(LEASE_FILE);
|
|
} catch (e) {
|
|
st = null;
|
|
}
|
|
if (!st) {
|
|
leases.mtime = -1;
|
|
return;
|
|
}
|
|
if (st.mtime == leases.mtime)
|
|
return;
|
|
leases.mtime = st.mtime;
|
|
|
|
let raw;
|
|
try {
|
|
raw = fs.readfile(LEASE_FILE);
|
|
} catch (e) {
|
|
return;
|
|
}
|
|
if (!raw)
|
|
return;
|
|
|
|
let by_mac = {}, by_ip = {};
|
|
for (let line in split(raw, "\n")) {
|
|
let f = split(trim(line), /[ \t]+/);
|
|
// <expiry> <mac> <ip> <hostname> <clientid>
|
|
if (length(f) < 4 || f[3] == "*")
|
|
continue;
|
|
by_mac[lc(f[1])] = f[3];
|
|
by_ip[f[2]] = f[3];
|
|
}
|
|
leases.by_mac = by_mac;
|
|
leases.by_ip = by_ip;
|
|
}
|
|
|
|
function device_name(key, mac, ip, now) {
|
|
if (key == "router")
|
|
return "Router";
|
|
if (key == "unknown")
|
|
return "Unknown";
|
|
if (key == "multicast")
|
|
return "Multicast / Broadcast";
|
|
load_leases(now);
|
|
return leases.by_mac[mac ?? ""] ?? leases.by_ip[ip ?? ""] ?? (mac ?? ip ?? key);
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Router identity — needed to tell "this box" from a client (§3.2).
|
|
* ------------------------------------------------------------------ */
|
|
|
|
// Rebuild the router-identity maps wholesale from an interfaces block. An
|
|
// agent_status frame that carries `interfaces` is complete and authoritative,
|
|
// so replacing rather than merging keeps these maps bounded (IPv6 privacy
|
|
// addresses and WAN re-leases would otherwise accumulate stale entries for the
|
|
// daemon's whole lifetime). The reduced agent_status frame omits `interfaces`
|
|
// entirely — guarded by the caller — and must not wipe what we know.
|
|
function router_learn(ifaces) {
|
|
if (type(ifaces) != "object")
|
|
return;
|
|
let macs = {}, ips = {};
|
|
for (let ifname in ifaces) {
|
|
let i = ifaces[ifname];
|
|
if (length(i?.mac ?? ""))
|
|
macs[lc(i.mac)] = ifname;
|
|
for (let a in i?.addr ?? [])
|
|
ips[a] = ifname;
|
|
}
|
|
router.macs = macs;
|
|
router.ips = ips;
|
|
}
|
|
|
|
// netifyd publishes the same interface block in its status file; read it once
|
|
// at start-up so attribution is correct before the first agent_status arrives.
|
|
function router_bootstrap() {
|
|
let raw;
|
|
try {
|
|
raw = fs.readfile(STATUS_FILE);
|
|
} catch (e) {
|
|
return;
|
|
}
|
|
if (!raw)
|
|
return;
|
|
try {
|
|
router_learn(json(raw).interfaces);
|
|
} catch (e) {
|
|
// status.json is rewritten in place by netifyd; a torn read is normal
|
|
}
|
|
}
|
|
|
|
function is_router(mac, ip) {
|
|
return (length(mac ?? "") && router.macs[mac] != null) ||
|
|
(length(ip ?? "") && router.ips[ip] != null);
|
|
}
|
|
|
|
// IEEE 802 group addresses: IPv4 multicast (01:00:5e:*), IPv6 multicast
|
|
// (33:33:*) and broadcast. These are not devices — attributing them per
|
|
// address would let any LAN host mint unbounded device aggregates by looping
|
|
// over multicast groups, so they collapse into one pseudo-device.
|
|
function is_group_mac(mac) {
|
|
if (!length(mac ?? ""))
|
|
return false;
|
|
return substr(mac, 0, 8) == "01:00:5e" ||
|
|
substr(mac, 0, 5) == "33:33" ||
|
|
mac == "ff:ff:ff:ff:ff:ff";
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Flow table (§3.2, §3.3)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function flow_detach(fr) {
|
|
if (fr.shadow)
|
|
return;
|
|
if (fr.a && fr.a.cur.flows > 0) fr.a.cur.flows--;
|
|
if (fr.d && fr.d.cur.flows > 0) fr.d.cur.flows--;
|
|
if (fr.c && fr.c.cur.flows > 0) fr.c.cur.flows--;
|
|
}
|
|
|
|
function flow_attach(fr) {
|
|
if (fr.shadow)
|
|
return;
|
|
if (fr.a) fr.a.cur.flows++;
|
|
if (fr.d) fr.d.cur.flows++;
|
|
if (fr.c) fr.c.cur.flows++;
|
|
}
|
|
|
|
function flow_drop(digest) {
|
|
let fr = flows[digest];
|
|
if (!fr)
|
|
return;
|
|
flow_detach(fr);
|
|
delete flows[digest];
|
|
nflows--;
|
|
}
|
|
|
|
// Identify (or re-identify, on a detection update) an existing flow record.
|
|
/* ------------------------------------------------------------------ *
|
|
* NAT twin resolution (3.2)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
// On the EXTERNAL (WAN) capture of a NAT-ed flow, netifyd 4.4.7 reports the
|
|
// POST-translation identity: local_ip and local_mac are this router's own WAN
|
|
// address and MAC. That is indistinguishable from a flow the router itself
|
|
// originated, so the capture cannot be classified from the event alone.
|
|
//
|
|
// netifyd's own ip_nat field is exactly the question being asked, and it is
|
|
// carried on the record as fr.nat, but 4.4.7 does not set it on this path:
|
|
// measured 0 true out of 24 external captures of genuinely NAT-ed client
|
|
// flows, and 0 out of 24 on the matching internal captures. It is therefore
|
|
// read but never trusted as evidence.
|
|
//
|
|
// conntrack does know, because it holds both halves of the translation. For a
|
|
// masqueraded flow the ORIGINAL source is the LAN client while the REPLY
|
|
// destination is this router's WAN address; for router-originated traffic the
|
|
// two are the same address:
|
|
//
|
|
// src=192.168.1.50 dst=P ... src=P dst=192.168.72.10 <- client, NAT-ed
|
|
// src=192.168.72.10 dst=Q ... src=Q dst=192.168.72.10 <- router's own
|
|
//
|
|
// Reading a snapshot is not free, so it is reused for CT_TTL_MS: a burst of new
|
|
// flows shares one read, and an idle router never reads at all, because loading
|
|
// is demand-driven from ct_nat_twin() below.
|
|
function ct_load(now, force) {
|
|
if (conntrack.avail == false && (now - conntrack.probed) < CT_RETRY_MS)
|
|
return;
|
|
// A forced reload comes from a lookup miss, which is the ordinary case for a
|
|
// flow younger than the current snapshot. Honour it, but never more often
|
|
// than CT_FORCE_MIN_MS, so a burst of new flows cannot turn into a burst of
|
|
// reads of a file that is large on a busy router.
|
|
if (force) {
|
|
if ((now - conntrack.at) < CT_FORCE_MIN_MS)
|
|
return;
|
|
} else if (conntrack.at && (now - conntrack.at) < CT_TTL_MS) {
|
|
// Throttle on "we read recently", NOT on "we read recently and got
|
|
// entries". Gating this on conntrack.entries meant a snapshot that
|
|
// parsed to zero rows while the file was readable disabled the throttle
|
|
// entirely, and the unforced path has no CT_FORCE_MIN_MS floor, so every
|
|
// external flow identification re-read the whole file.
|
|
return;
|
|
}
|
|
conntrack.at = now;
|
|
conntrack.probed = now;
|
|
|
|
// Read INCREMENTALLY, one line at a time. This used to be fs.readfile()
|
|
// followed by split(raw, "\n"), which materialised the entire conntrack
|
|
// table and then the entire line array before CT_MAX was ever consulted --
|
|
// so CT_MAX bounded the retained map but not peak allocation, and the
|
|
// comment on the constant calling it a memory bound was wrong. On a busy
|
|
// router that is megabytes of transient garbage per read, on a device that
|
|
// may have 64 MB. Reading a line at a time makes the peak one line, and
|
|
// the CT_MAX break now stops the work as well as the retention.
|
|
let fd;
|
|
try {
|
|
fd = fs.open(cfg.conntrack_path ?? CONNTRACK_FILE, "r");
|
|
} catch (e) {
|
|
fd = null;
|
|
}
|
|
if (!fd) {
|
|
if (conntrack.avail != false) {
|
|
conntrack.avail = false;
|
|
conntrack.map = {};
|
|
conntrack.entries = 0;
|
|
logonce("ct-missing", "warn",
|
|
"%s unreadable: cannot distinguish a NAT twin from router-originated traffic, so external captures are shadowed while an internal capture runs. Client accounting stays correct; the router's own traffic is not attributed.",
|
|
cfg.conntrack_path ?? CONNTRACK_FILE);
|
|
}
|
|
return;
|
|
}
|
|
conntrack.avail = true;
|
|
|
|
let map = {}, n = 0;
|
|
while (true) {
|
|
if (n >= CT_MAX)
|
|
break;
|
|
let line = fd.read("line");
|
|
if (line == null || !length(line))
|
|
break;
|
|
// Each line carries two tuples in order, original then reply. Collect the
|
|
// fields positionally rather than matching the whole line: the optional
|
|
// state word (TIME_WAIT, [UNREPLIED], ...) shifts the columns and differs
|
|
// between protocols.
|
|
let src = [], dst = [], sport = [], dport = [];
|
|
for (let tok in split(line, " ")) {
|
|
if (substr(tok, 0, 4) == "src=") push(src, substr(tok, 4));
|
|
else if (substr(tok, 0, 4) == "dst=") push(dst, substr(tok, 4));
|
|
else if (substr(tok, 0, 6) == "sport=") push(sport, substr(tok, 6));
|
|
else if (substr(tok, 0, 6) == "dport=") push(dport, substr(tok, 6));
|
|
}
|
|
if (length(src) < 2 || length(dst) < 2 || length(sport) < 2 || length(dport) < 2)
|
|
continue;
|
|
// Key on the reply tuple; value is whether the source was translated.
|
|
map[src[1] + "|" + dst[1] + "|" + sport[1] + "|" + dport[1]] = (src[0] != dst[1]);
|
|
n++;
|
|
}
|
|
try { fd.close(); } catch (e) {}
|
|
conntrack.map = map;
|
|
conntrack.entries = n;
|
|
}
|
|
|
|
// true = this external capture is the WAN-side twin of a NAT-ed LAN flow
|
|
// false = conntrack positively says the router originated it
|
|
// null = conntrack cannot say (absent, unreadable, or the entry has expired)
|
|
function ct_nat_twin(f, now, fr) {
|
|
// THE PORTS COME FROM THE FLOW RECORD, NOT THE EVENT, AND THAT IS THE WHOLE
|
|
// FIX. netifyd puts local_port/other_port on the initial `flow` event and
|
|
// on nothing afterwards: a `flow_stats` payload carries addresses and byte
|
|
// counters only. This key was built straight off the event, so every call
|
|
// made from an update produced "...|93.184.216.34|192.168.72.10||" with two
|
|
// empty fields, matched nothing, and returned null.
|
|
//
|
|
// Nothing detected that. reshadow() treats null as grounds to discard, so
|
|
// EVERY router-originated flow was shadowed on its first stats update, and
|
|
// the "Router" device counted zero bytes on every install since 1.0.x. The
|
|
// dashboard then correctly hides a zero-byte device, so the feature was
|
|
// absent rather than visibly broken. Production had logged 107,476
|
|
// positive identifications against a row reading zero for 2.6 days.
|
|
//
|
|
// Found 2026-08-31 by instrumenting the daemon after the row was noticed
|
|
// missing on the bench. See tests/protocol-suite.sh `router`.
|
|
let lp = (fr != null && fr.local_port != null) ? fr.local_port : f.local_port;
|
|
let op = (fr != null && fr.other_port != null) ? fr.other_port : f.other_port;
|
|
let key = (f.other_ip ?? "") + "|" + (f.local_ip ?? "") + "|" +
|
|
(op ?? "") + "|" + (lp ?? "");
|
|
ct_load(now, false);
|
|
let v = conntrack.entries ? conntrack.map[key] : null;
|
|
// A miss usually means the flow is newer than the snapshot, because netifyd
|
|
// identifies a flow within moments of it starting. Refresh once and retry
|
|
// before giving up: without this the common case resolves to "unknown" and
|
|
// every router-originated flow is shadowed, which silently degrades to the
|
|
// no-conntrack behaviour even though conntrack is available.
|
|
if (v == null && conntrack.avail != false) {
|
|
ct_load(now, true);
|
|
v = conntrack.entries ? conntrack.map[key] : null;
|
|
}
|
|
return (v == null) ? null : v;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* AI service breakout (§10)
|
|
*
|
|
* WHY THIS EXISTS, measured rather than assumed.
|
|
*
|
|
* netifyd cannot classify AI traffic and will not soon. Its signature set,
|
|
* /etc/netify.d/netify-apps.conf, is dated 10 August 2023 on both routers
|
|
* checked, carries 199 application signatures, and contains ZERO entries for
|
|
* anthropic, claude, openai, chatgpt, huggingface, perplexity, mistral,
|
|
* copilot, bard or gemini. So this traffic arrives with detected_application 0
|
|
* and lands in generic HTTP/S alongside everything else.
|
|
*
|
|
* But the identifying data is already here. netifyd reports the TLS SNI as
|
|
* host_server_name, and flows to the real endpoints carry api.anthropic.com,
|
|
* claude.ai and api.openai.com with no code change on our side. The daemon is
|
|
* therefore the right layer: aggregation happens here, and the browser only
|
|
* ever sees summarised rows.
|
|
*
|
|
* THIS IS HOSTNAME MATCHING, NOT DPI, AND MUST BE DESCRIBED THAT WAY. It
|
|
* asserts a classification netifyd did not make. The technique is legitimate
|
|
* -- matching the SNI is the same mechanism netifyd itself uses for HTTP/S --
|
|
* but the honest framing is "we recognise the name", not "we inspected the
|
|
* protocol".
|
|
*
|
|
* Three limitations, stated here so they are not discovered later:
|
|
*
|
|
* - The list rots. New services appear constantly and nothing here updates
|
|
* itself. `ai_breakout '0'` turns the whole thing off.
|
|
* - ECH (Encrypted Client Hello) encrypts the SNI and would end this
|
|
* outright. netifyd has the identical exposure, so this is not worse than
|
|
* the status quo, but it is not durable either.
|
|
* - Region-encoded hostnames under a generic cloud domain cannot be matched
|
|
* by suffix without false positives. bedrock-runtime.us-east-1.amazonaws.com
|
|
* is the obvious example and is deliberately absent: no suffix of it is
|
|
* specific to Bedrock, and matching amazonaws.com would label every S3
|
|
* bucket on the network as AI.
|
|
*
|
|
* ACCOUNTING IS NOT AFFECTED. This changes an identity, never a byte count.
|
|
* account() never sees it, so byte conservation is exactly as it was and the
|
|
* existing conservation checks still hold.
|
|
*/
|
|
|
|
/* Keyed on a domain suffix, matched against the last two, three or four labels
|
|
* of the SNI. Value is [ app label, category, strong ].
|
|
*
|
|
* THE CATEGORY IS A SLUG, NOT A DISPLAY STRING, and it must stay one. Until
|
|
* 1.1.1 this column held "AI assistants" and its three siblings, which read
|
|
* correctly in English and could therefore never be translated: the browser
|
|
* renders a category by looking the slug up in a table of `_()` literals, and
|
|
* a string that arrives already rendered has no key to look up. Every netifyd
|
|
* category reaches the frontend as a lowercase slug; these four were the only
|
|
* ones that did not, and they were ours. See catLabel() in common.js, which
|
|
* carries the matching `_('AI assistants')`.
|
|
*
|
|
* `strong` means: override even when netifyd DID identify an application. It
|
|
* is set only where netifyd is known to recognise the PARENT brand and would
|
|
* otherwise swallow the AI service inside it -- Gemini reported as "Google",
|
|
* Copilot as "GitHub". Everywhere else the rule is conservative: netifyd's own
|
|
* answer wins, so the day Netify ships AI signatures this table goes quiet by
|
|
* itself instead of permanently shadowing better data.
|
|
*/
|
|
const AI_HOSTS = {
|
|
// ---- assistants and model vendors --------------------------------
|
|
"anthropic.com": [ "Anthropic (Claude)", "ai-assistants", false ],
|
|
"claude.ai": [ "Anthropic (Claude)", "ai-assistants", false ],
|
|
"openai.com": [ "OpenAI", "ai-assistants", false ],
|
|
"chatgpt.com": [ "OpenAI (ChatGPT)", "ai-assistants", false ],
|
|
"oaistatic.com": [ "OpenAI", "ai-assistants", false ],
|
|
"oaiusercontent.com": [ "OpenAI", "ai-assistants", false ],
|
|
"gemini.google.com": [ "Google Gemini", "ai-assistants", true ],
|
|
"aistudio.google.com": [ "Google Gemini", "ai-assistants", true ],
|
|
"generativelanguage.googleapis.com":
|
|
[ "Google Gemini", "ai-assistants", true ],
|
|
"x.ai": [ "xAI (Grok)", "ai-assistants", false ],
|
|
"grok.com": [ "xAI (Grok)", "ai-assistants", false ],
|
|
"mistral.ai": [ "Mistral", "ai-assistants", false ],
|
|
"cohere.com": [ "Cohere", "ai-assistants", false ],
|
|
"cohere.ai": [ "Cohere", "ai-assistants", false ],
|
|
"deepseek.com": [ "DeepSeek", "ai-assistants", false ],
|
|
"perplexity.ai": [ "Perplexity", "ai-assistants", false ],
|
|
"moonshot.cn": [ "Moonshot (Kimi)", "ai-assistants", false ],
|
|
"bigmodel.cn": [ "Zhipu", "ai-assistants", false ],
|
|
"dashscope.aliyuncs.com":
|
|
[ "Alibaba Qwen", "ai-assistants", true ],
|
|
"ai21.com": [ "AI21", "ai-assistants", false ],
|
|
"meta.ai": [ "Meta AI", "ai-assistants", false ],
|
|
"reka.ai": [ "Reka", "ai-assistants", false ],
|
|
"character.ai": [ "Character.AI", "ai-assistants", false ],
|
|
"poe.com": [ "Poe", "ai-assistants", false ],
|
|
"phind.com": [ "Phind", "ai-assistants", false ],
|
|
|
|
// ---- media generation --------------------------------------------
|
|
"midjourney.com": [ "Midjourney", "ai-media", false ],
|
|
"stability.ai": [ "Stability AI", "ai-media", false ],
|
|
"runwayml.com": [ "Runway", "ai-media", false ],
|
|
"elevenlabs.io": [ "ElevenLabs", "ai-media", false ],
|
|
"suno.com": [ "Suno", "ai-media", false ],
|
|
"suno.ai": [ "Suno", "ai-media", false ],
|
|
"udio.com": [ "Udio", "ai-media", false ],
|
|
"lumalabs.ai": [ "Luma", "ai-media", false ],
|
|
"bfl.ai": [ "Black Forest Labs", "ai-media", false ],
|
|
"ideogram.ai": [ "Ideogram", "ai-media", false ],
|
|
"leonardo.ai": [ "Leonardo", "ai-media", false ],
|
|
"pika.art": [ "Pika", "ai-media", false ],
|
|
"heygen.com": [ "HeyGen", "ai-media", false ],
|
|
"synthesia.io": [ "Synthesia", "ai-media", false ],
|
|
"descript.com": [ "Descript", "ai-media", false ],
|
|
|
|
// ---- coding assistants and dev tools -----------------------------
|
|
"githubcopilot.com": [ "GitHub Copilot", "ai-developer", true ],
|
|
"copilot.microsoft.com":[ "Microsoft Copilot", "ai-developer", true ],
|
|
"cursor.sh": [ "Cursor", "ai-developer", false ],
|
|
"cursor.com": [ "Cursor", "ai-developer", false ],
|
|
"codeium.com": [ "Codeium", "ai-developer", false ],
|
|
"windsurf.com": [ "Windsurf", "ai-developer", false ],
|
|
"tabnine.com": [ "Tabnine", "ai-developer", false ],
|
|
"sourcegraph.com": [ "Sourcegraph Cody", "ai-developer", false ],
|
|
"v0.dev": [ "v0", "ai-developer", false ],
|
|
"bolt.new": [ "Bolt", "ai-developer", false ],
|
|
"lovable.dev": [ "Lovable", "ai-developer", false ],
|
|
"replit.com": [ "Replit", "ai-developer", false ],
|
|
|
|
// ---- aggregators, inference APIs, GPU rental, AI infrastructure ---
|
|
"openrouter.ai": [ "OpenRouter", "ai-infrastructure", false ],
|
|
"together.ai": [ "Together AI", "ai-infrastructure", false ],
|
|
"together.xyz": [ "Together AI", "ai-infrastructure", false ],
|
|
"replicate.com": [ "Replicate", "ai-infrastructure", false ],
|
|
"replicate.delivery": [ "Replicate", "ai-infrastructure", false ],
|
|
"fireworks.ai": [ "Fireworks AI", "ai-infrastructure", false ],
|
|
"groq.com": [ "Groq", "ai-infrastructure", false ],
|
|
"deepinfra.com": [ "DeepInfra", "ai-infrastructure", false ],
|
|
"novita.ai": [ "Novita", "ai-infrastructure", false ],
|
|
"hyperbolic.xyz": [ "Hyperbolic", "ai-infrastructure", false ],
|
|
"anyscale.com": [ "Anyscale", "ai-infrastructure", false ],
|
|
"sambanova.ai": [ "SambaNova", "ai-infrastructure", false ],
|
|
"cerebras.ai": [ "Cerebras", "ai-infrastructure", false ],
|
|
"huggingface.co": [ "Hugging Face", "ai-infrastructure", false ],
|
|
"hf.co": [ "Hugging Face", "ai-infrastructure", false ],
|
|
"vast.ai": [ "Vast.ai", "ai-infrastructure", false ],
|
|
"runpod.io": [ "RunPod", "ai-infrastructure", false ],
|
|
"runpod.ai": [ "RunPod", "ai-infrastructure", false ],
|
|
"lambdalabs.com": [ "Lambda", "ai-infrastructure", false ],
|
|
"paperspace.com": [ "Paperspace", "ai-infrastructure", false ],
|
|
"coreweave.com": [ "CoreWeave", "ai-infrastructure", false ],
|
|
"modal.com": [ "Modal", "ai-infrastructure", false ],
|
|
"baseten.co": [ "Baseten", "ai-infrastructure", false ],
|
|
"beam.cloud": [ "Beam", "ai-infrastructure", false ],
|
|
"salad.com": [ "Salad", "ai-infrastructure", false ],
|
|
"tensordock.com": [ "TensorDock", "ai-infrastructure", false ],
|
|
"fly.io": [ "Fly.io", "ai-infrastructure", false ],
|
|
"wandb.ai": [ "Weights & Biases", "ai-infrastructure", false ],
|
|
"langchain.com": [ "LangChain", "ai-infrastructure", false ],
|
|
"pinecone.io": [ "Pinecone", "ai-infrastructure", false ],
|
|
"weaviate.io": [ "Weaviate", "ai-infrastructure", false ],
|
|
"qdrant.tech": [ "Qdrant", "ai-infrastructure", false ],
|
|
"trychroma.com": [ "Chroma", "ai-infrastructure", false ]
|
|
};
|
|
|
|
/* Match the SNI against AI_HOSTS by domain suffix.
|
|
*
|
|
* SUFFIX, ANCHORED ON A LABEL BOUNDARY -- never a substring test. A substring
|
|
* match on "anthropic.com" also matches "anthropic.com.attacker.example", which
|
|
* is a hostname anyone can register and point at anything. Splitting on labels
|
|
* and rebuilding the last N makes that impossible by construction rather than
|
|
* by care.
|
|
*
|
|
* Tries the last two labels first, then three, then four, so a specific entry
|
|
* (gemini.google.com) is reachable without a linear scan of the table and a
|
|
* generic parent cannot shadow it.
|
|
*/
|
|
function ai_match(host) {
|
|
let h = lc(trim(host ?? ""));
|
|
|
|
if (!length(h))
|
|
return null;
|
|
|
|
/* A trailing dot is legal in a FQDN and would otherwise produce an empty
|
|
* final label, shifting every suffix by one and matching nothing. */
|
|
if (substr(h, -1) == ".")
|
|
h = substr(h, 0, length(h) - 1);
|
|
|
|
let parts = split(h, ".");
|
|
let n = length(parts);
|
|
|
|
if (n < 2)
|
|
return null;
|
|
|
|
for (let take = 2; take <= 4 && take <= n; take++) {
|
|
let suffix = join(".", slice(parts, n - take));
|
|
|
|
/* User entries first, so `config hostmap` can override a built-in that
|
|
* is wrong as well as add one that is missing. */
|
|
let hit = (cfg?.hostmap ?? {})[suffix] ?? AI_HOSTS[suffix];
|
|
|
|
/* Type-checked rather than merely truthy.
|
|
*
|
|
* In a BROWSER this would be load-bearing: `{}["constructor"]` returns
|
|
* a function, so a hostname of "constructor" would read as a hit. In
|
|
* ucode it is not, because ucode objects have no prototype chain and
|
|
* that lookup is null. Verified by sabotage: replacing this with a
|
|
* truthiness test changes no test result.
|
|
*
|
|
* It stays because what it actually guards is a malformed entry in the
|
|
* table above -- a two-element row, or a string where an array belongs
|
|
* -- which would otherwise be indexed as if it were well formed and
|
|
* produce a null label on a live router. It is NOT a defence against a
|
|
* prototype-key hostname, and should not be described as one. */
|
|
if (type(hit) == "array" && length(hit) == 3)
|
|
return { app: hit[0], cat: hit[1], strong: hit[2], via: suffix };
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function flow_identify(fr, ev, f, now) {
|
|
let app_id = num(f.detected_application, 0);
|
|
let proto_id = num(f.detected_protocol, 0);
|
|
|
|
// Cap any tag taken straight off the wire (detected_*_name) so a single
|
|
// flow record cannot hold an unbounded string.
|
|
let cap = (s) => (length(s ?? "") > TAG_MAX) ? substr(s, 0, TAG_MAX) : s;
|
|
|
|
let key, label;
|
|
if (app_id > 0) {
|
|
key = "a" + app_id;
|
|
fr.tag = cap(f.detected_application_name ?? names.app[app_id]);
|
|
label = display_name(fr.tag);
|
|
} else if (proto_id > 0) {
|
|
key = "p" + proto_id;
|
|
fr.tag = cap(f.detected_protocol_name ?? names.proto[proto_id]);
|
|
label = display_name(fr.tag);
|
|
} else {
|
|
key = "unknown";
|
|
label = "Unknown";
|
|
fr.tag = "unknown";
|
|
}
|
|
if (!length(label ?? "")) {
|
|
label = display_name(f.detected_protocol_name ?? names.proto[proto_id]) ?? "Unknown";
|
|
if (!length(fr.tag ?? ""))
|
|
fr.tag = "unknown";
|
|
}
|
|
|
|
let cat = catnames.app[num(f.category?.application, -1)] ??
|
|
catnames.proto[num(f.category?.protocol, -1)] ?? "unclassified";
|
|
|
|
if (length(f.host_server_name ?? ""))
|
|
fr.host = substr(f.host_server_name, 0, 63);
|
|
|
|
// AI service breakout (10). Runs AFTER the netifyd identity is resolved and
|
|
// after `cat`, because it replaces both, and BEFORE attribution, because
|
|
// nothing below this point depends on the app identity.
|
|
//
|
|
// The gate is deliberately conservative: netifyd's own answer wins unless
|
|
// the entry is marked `strong`, which is set only where netifyd is known to
|
|
// recognise the parent brand and would swallow the AI service inside it.
|
|
// So the day Netify ships AI signatures, this table falls silent by itself
|
|
// rather than permanently shadowing better data.
|
|
//
|
|
// Identity only. No byte count is touched here, account() never sees this,
|
|
// and conservation is unaffected.
|
|
if (cfg.ai_breakout && length(fr.host ?? "")) {
|
|
let ai = ai_match(fr.host);
|
|
|
|
if (ai && (app_id == 0 || ai.strong)) {
|
|
// Key on the LABEL, not on the matched suffix: anthropic.com and
|
|
// claude.ai are one vendor and must aggregate into one row, not two
|
|
// identically-named ones.
|
|
let slug = lc(replace(ai.app, /[^a-zA-Z0-9]+/g, "-"));
|
|
slug = replace(slug, /^-+/, "");
|
|
slug = replace(slug, /-+$/, "");
|
|
|
|
key = "ai:" + slug;
|
|
label = ai.app;
|
|
// The icon pack is keyed on this tag. An `appflow.` prefix keeps the
|
|
// namespace honest: nothing here pretends to have come from netifyd.
|
|
fr.tag = "appflow." + slug;
|
|
cat = ai.cat;
|
|
stats.ai_labeled++;
|
|
}
|
|
}
|
|
|
|
// Attribution (§3.2). netifyd's "local"/"other" labels follow its own
|
|
// topology view, not ours, so decide which endpoint is *our* device:
|
|
// - on an internal (LAN bridge) capture the device is the non-router side;
|
|
// - on an external (WAN) capture our own box is the device and the peer is
|
|
// upstream infrastructure, so the device is the router-owned side.
|
|
// Ports only ever arrive on the initial `flow` event. Keep them on the
|
|
// record so conntrack can still be asked about this flow later; see
|
|
// ct_nat_twin().
|
|
if (f.local_port != null) fr.local_port = f.local_port;
|
|
if (f.other_port != null) fr.other_port = f.other_port;
|
|
|
|
let lm = lc(f.local_mac ?? ""), om = lc(f.other_mac ?? "");
|
|
let rl = is_router(lm, f.local_ip), ro = is_router(om, f.other_ip);
|
|
let dev_local = ev.internal ? !(rl && !ro) : !(ro && !rl);
|
|
let dev_is_router = dev_local ? rl : ro;
|
|
let mac = dev_local ? lm : om;
|
|
let ip = dev_local ? f.local_ip : f.other_ip;
|
|
// Honour netifyd's other_type when it names a group endpoint. netifyd's
|
|
// enum documents "broadcast"/"multicast", though only "local"/"remote" were
|
|
// seen on this bench, so the group-bit MAC test below is the path that
|
|
// actually fires today; the other_type check is a forward-compatible
|
|
// preference that costs nothing and needs no re-verification to be safe (it
|
|
// only ever adds a positive on a documented value).
|
|
let ot = f.other_type;
|
|
let group = (ot == "multicast" || ot == "broadcast") ||
|
|
is_group_mac(mac) || is_group_mac(dev_local ? om : lm);
|
|
let dkey = group ? "multicast"
|
|
: (dev_is_router ? "router" : (length(mac) ? mac : (ip ?? "unknown")));
|
|
|
|
// Single-count each flow. The SHIPPING rule is documented in DESIGN §3.2
|
|
// (which this implements; the earlier ip_nat-pairing design was replaced):
|
|
// with netifyd watching the LAN bridge (internal) and WAN (external) at once,
|
|
// a client's traffic is captured twice. Rather than pair the two captures by
|
|
// a fragile remote-endpoint key, count deterministically -- an internal
|
|
// capture is always the device's own copy, while an external capture is
|
|
// counted only when the device is the router itself (router-originated
|
|
// traffic never crosses the bridge); an external client-flow capture is
|
|
// therefore a NAT twin and is shadowed.
|
|
//
|
|
// NOT order-independent in the strict sense: saw_internal is a process-wide
|
|
// capability latch ("is any internal capture running"), not per-flow evidence
|
|
// of a specific twin. A WAN-side flow identified BEFORE the first internal
|
|
// event is counted; the same flow after is shadowed. Left there, a
|
|
// pre-existing external stub identified while saw_internal was still false
|
|
// would stay counted alongside its internal twin for its whole life (a real
|
|
// double-count of the central total, not a brief startup blip) -- reshadow()
|
|
// re-applies this rule on later stats updates to close that window. The
|
|
// per-flow twin question, and the exact contents of local_mac/other_mac when
|
|
// ip_nat is set, remain the open bench-test item in DESIGN §3.2 / §8.
|
|
if (fr.internal)
|
|
saw_internal = true;
|
|
|
|
// Single-count decision for an EXTERNAL capture, once an internal capture is
|
|
// known to be running. Two cases, and they are not symmetric:
|
|
//
|
|
// device is not the router -> a LAN client seen on the WAN side, which only
|
|
// happens as the twin of its own internal
|
|
// capture. Shadow it.
|
|
// device IS the router -> ambiguous by construction, because netifyd
|
|
// reports the post-NAT identity. Ask conntrack.
|
|
//
|
|
// When conntrack cannot answer, shadow. The two possible errors are not
|
|
// equal: counting a twin double-counts a client's entire traffic, measured at
|
|
// 163-199% of wire truth, while shadowing router-originated traffic omits the
|
|
// router's own DNS, NTP and updates. Default to the smaller error.
|
|
let want_shadow = false;
|
|
if (saw_internal && !fr.internal && !group) {
|
|
if (!dev_is_router)
|
|
want_shadow = true;
|
|
else {
|
|
let twin = ct_nat_twin(f, now, fr);
|
|
want_shadow = (twin != false);
|
|
if (twin == false)
|
|
fr.ct_local = true; // a positive finding, remembered
|
|
if (twin == null)
|
|
stats.ct_unresolved++;
|
|
else if (twin)
|
|
stats.ct_twins++;
|
|
else
|
|
stats.ct_local++;
|
|
}
|
|
}
|
|
|
|
if (fr.app_key == key && fr.cat_key == cat && fr.dev_key == dkey &&
|
|
fr.shadow == want_shadow) {
|
|
if (!fr.dev_ip)
|
|
fr.dev_ip = ip;
|
|
return; // nothing structural changed
|
|
}
|
|
|
|
// Re-pointing an already-counted flow: bytes already attributed stay where
|
|
// they were (the accounting seam is append-only by design); only future
|
|
// bytes and the active-flow counts follow the new identity.
|
|
flow_detach(fr); // uses the OLD shadow state
|
|
if (want_shadow && !fr.shadow)
|
|
stats.shadowed++;
|
|
fr.shadow = want_shadow;
|
|
if (fr.shadow) { // a shadowed twin tracks nothing
|
|
fr.app_key = key; fr.cat_key = cat; fr.dev_key = dkey;
|
|
fr.dev_mac = length(mac) ? mac : null; fr.dev_ip = ip;
|
|
fr.dev_local = dev_local;
|
|
fr.a = null; fr.d = null; fr.c = null;
|
|
return;
|
|
}
|
|
fr.app_key = key;
|
|
fr.app_name = label ?? "Unknown";
|
|
fr.cat_key = cat;
|
|
fr.dev_key = dkey;
|
|
fr.dev_mac = length(mac) ? mac : null;
|
|
fr.dev_ip = ip;
|
|
fr.dev_local = dev_local;
|
|
fr.a = agg_make(apps, "app", key, fr.app_name, now);
|
|
fr.d = agg_make(devs, "device", dkey, device_name(dkey, fr.dev_mac, ip, now), now);
|
|
fr.c = agg_make(cats, "category", cat, cat, now);
|
|
if (fr.d) {
|
|
fr.d.mac = fr.dev_mac;
|
|
// Keep the friendliest address for display: an interface usually has
|
|
// both, and an IPv4 address means more to a user than a link-local v6.
|
|
if (ip && (!fr.d.ip || (index(ip, ".") >= 0 && index(fr.d.ip, ".") < 0)))
|
|
fr.d.ip = ip;
|
|
fr.d.router = (dkey == "router");
|
|
}
|
|
if (fr.a) {
|
|
fr.a.tag = fr.tag;
|
|
fr.a.cat = cat;
|
|
}
|
|
flow_attach(fr);
|
|
}
|
|
|
|
// Re-apply the single-count rule to a flow whose shadow decision was frozen at
|
|
// creation. flow_identify runs once per flow (at flow_new) and again only on a
|
|
// `flow` event; a pre-existing external stub, created from flow_stats before any
|
|
// internal event, is therefore identified with saw_internal still false, counted,
|
|
// and never revisited -- so it double-counts against its internal twin for its
|
|
// whole life. Once saw_internal latches, such a flow is an external non-router
|
|
// capture like any other and the shipping rule already says to shadow it; this
|
|
// just enforces that rule for flows that predated the latch, closing the window
|
|
// from "entire flow life" to "at most one stats interval". It only ever tightens
|
|
// (unshadowed -> shadowed), never resurrects. The residual sign question for an
|
|
// unidentifiable stub (a genuine NAT twin vs. pre-existing router-origin traffic,
|
|
// both landing in the "unknown" bucket) is the same §3.2 bench-test item.
|
|
function reshadow(fr, f, now) {
|
|
if (fr.shadow || !saw_internal || fr.internal)
|
|
return;
|
|
if (fr.dev_key == "multicast")
|
|
return;
|
|
if (fr.dev_key == "router") {
|
|
// A WAN-side capture of a NAT-ed client flow is classified "router",
|
|
// because netifyd reports the post-translation identity (§3.2). That is
|
|
// precisely the class this function exists to correct, so the blanket
|
|
// `return` that stood here until 2026-08-24 exempted the only flows that
|
|
// needed it: after a restart, an established flow's external copy stayed
|
|
// counted alongside its internal twin for the flow's whole remaining
|
|
// life, and stats.shadowed never moved because there was no transition.
|
|
// Ask conntrack, exactly as flow_identify does, and default the same way.
|
|
let twin = ct_nat_twin(f, now, fr);
|
|
if (twin == false) {
|
|
stats.ct_local++;
|
|
return; // conntrack says the router originated it
|
|
}
|
|
if (twin == null) {
|
|
stats.ct_unresolved++;
|
|
// ABSENT EVIDENCE IS NOT POSITIVE EVIDENCE, the same rule
|
|
// statusStrip() carries a comment about. If conntrack already told
|
|
// us this flow was router-originated, a later lookup that cannot
|
|
// answer -- because the connection closed and the entry aged out --
|
|
// is not a reason to throw the flow away. Only a positive "this is
|
|
// a NAT twin" below may overturn it.
|
|
if (fr.ct_local)
|
|
return;
|
|
} else {
|
|
stats.ct_twins++;
|
|
}
|
|
}
|
|
flow_detach(fr);
|
|
fr.shadow = true;
|
|
fr.a = null; fr.d = null; fr.c = null;
|
|
stats.shadowed++;
|
|
}
|
|
|
|
function flow_new(digest, ev, f, now, late) {
|
|
if (nflows >= cfg.flow_max) {
|
|
prune(now, true);
|
|
if (nflows >= cfg.flow_max) {
|
|
stats.flows_dropped++;
|
|
return null;
|
|
}
|
|
}
|
|
|
|
let fr = {
|
|
digest,
|
|
iface: ev.interface ?? "?",
|
|
internal: (ev.internal == true),
|
|
nat: (f.ip_nat == true),
|
|
proto: num(f.ip_protocol, 0),
|
|
host: null, tag: null,
|
|
app_key: null, app_name: "Unknown", cat_key: "unclassified",
|
|
dev_key: "unknown", dev_mac: null, dev_ip: null, dev_local: true,
|
|
a: null, d: null, c: null,
|
|
up: 0, down: 0, total: 0,
|
|
start: num(f.first_seen_at, now), seen: now,
|
|
shadow: false, late: (late == true)
|
|
};
|
|
flows[digest] = fr;
|
|
nflows++;
|
|
stats.flows_seen++;
|
|
if (late)
|
|
stats.flows_late++;
|
|
|
|
// Attribution and single-count shadowing are both decided in flow_identify,
|
|
// deterministically from this flow alone (no cross-flow pairing state).
|
|
flow_identify(fr, ev, f, now);
|
|
return fr;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Event handlers. Dispatch is table-driven: adding or renaming an event
|
|
* type is a one-line change.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function ev_hello(ev) {
|
|
agent.version = ev.agent_version ?? agent.version;
|
|
logmsg("info", "netify agent %s connected (%s)",
|
|
agent.version ?? "?", ev.build_version ?? "");
|
|
}
|
|
|
|
// In normal operation `names` is bounded by nDPI's catalogue (a few thousand
|
|
// ids from the definitions event). These caps make that bound explicit rather
|
|
// than incidental, so a buggy or hostile agent cannot grow the maps without
|
|
// limit: entries beyond NAME_MAX are dropped and every stored tag is truncated.
|
|
function store_name(tbl, id, tag) {
|
|
if (tbl[id] == null && length(tbl) >= NAME_MAX)
|
|
return;
|
|
tbl[id] = (length(tag ?? "") > TAG_MAX) ? substr(tag, 0, TAG_MAX) : tag;
|
|
}
|
|
|
|
function ev_definitions(ev) {
|
|
let na = 0, np = 0;
|
|
for (let e in ev.applications ?? []) {
|
|
if (e?.id == null) continue;
|
|
store_name(names.app, e.id, e.tag);
|
|
na++;
|
|
}
|
|
for (let e in ev.protocols ?? []) {
|
|
if (e?.id == null) continue;
|
|
store_name(names.proto, e.id, e.tag);
|
|
np++;
|
|
}
|
|
logmsg("info", "definitions: %d applications, %d protocols", na, np);
|
|
}
|
|
|
|
function ev_agent_status(ev) {
|
|
agent.version = ev.agent_version ?? agent.version;
|
|
agent.uptime = num(ev.uptime, agent.uptime);
|
|
agent.flow_count = num(ev.flow_count, 0);
|
|
agent.rss_kb = num(ev.maxrss_kb, 0);
|
|
agent.ts = num(ev.timestamp, 0) * 1000;
|
|
router_learn(ev.interfaces);
|
|
// ev.devices is netifyd's MAC -> IP[] map; used only to fill in an address
|
|
// for devices whose flows we have not seen an IP for yet.
|
|
for (let mac in ev.devices ?? {}) {
|
|
let d = devs[lc(mac)];
|
|
if (d && !d.ip)
|
|
d.ip = (ev.devices[mac] ?? [])[0];
|
|
}
|
|
}
|
|
|
|
function ev_flow(ev, f, now) {
|
|
let fr = flows[f.digest];
|
|
if (!fr) {
|
|
flow_new(f.digest, ev, f, now, false);
|
|
return;
|
|
}
|
|
fr.late = false;
|
|
flow_identify(fr, ev, f, now);
|
|
}
|
|
|
|
// local_bytes/other_bytes are *per-update deltas* (verified on 4.4.7: they
|
|
// return to zero in the final flow_purge while total_bytes, cumulative, does
|
|
// not). This shared helper accounts one such update, bounding the directional
|
|
// deltas by the growth of total_bytes so a self-inconsistent counter can never
|
|
// inflate the totals. Both flow_stats and flow_purge go through it, so the
|
|
// clamp guarantee in DESIGN §2.2 holds on every accounting path — including the
|
|
// purge path, whose payload shape is the one the interface doc could not
|
|
// capture live.
|
|
function flow_update(fr, f, now, final) {
|
|
let dl = num(f.local_bytes, 0), do_ = num(f.other_bytes, 0);
|
|
let tb = num(f.total_bytes, 0);
|
|
|
|
// Byte conservation (§2.5). What netifyd says this flow grew by, recorded
|
|
// before any clamping, so `bytes_reported` can be compared against what
|
|
// actually reached an aggregate. A late stub's first update is skipped:
|
|
// its total_bytes is the flow's whole pre-adoption history, which this
|
|
// daemon deliberately does not attribute, and counting it would show as
|
|
// permanent false drift rather than as the connect-blindness it is.
|
|
// Byte conservation (§2.5). netifyd's own statement of what moved, recorded
|
|
// BEFORE any clamp or shadow decision, so `leaked` in status exposes bytes
|
|
// that were reported to us and never reached an aggregate. It has to mirror
|
|
// the basis attribution actually uses: where a baseline exists the
|
|
// cumulative growth is authoritative, otherwise the directional deltas are
|
|
// all there is. A late stub's pre-adoption history is excluded by
|
|
// construction, because it lives in total_bytes and never in the deltas.
|
|
if (tb > 0 && fr.total > 0) {
|
|
let grew = tb - fr.total;
|
|
stats.bytes_reported += (grew < 0) ? tb : grew;
|
|
} else {
|
|
stats.bytes_reported += (dl + do_);
|
|
}
|
|
|
|
// Bound the deltas by total_bytes growth ONLY when total_bytes is usable
|
|
// (present and moving forward). A missing or reset counter must not scale
|
|
// the real directional deltas to zero (that would silently and permanently
|
|
// stop accounting a flow); in that case trust the deltas as given.
|
|
if (tb > 0 && fr.total > 0) {
|
|
let dt = tb - fr.total;
|
|
if (dt < 0)
|
|
dt = tb; // netifyd counter reset: treat tb as the delta
|
|
let sum = dl + do_;
|
|
if (sum > dt) {
|
|
// Directional deltas exceed the cumulative growth: clamp them down.
|
|
dl = (sum > 0) ? int(dl * dt / sum) : 0;
|
|
do_ = dt - dl;
|
|
} else if (sum == 0 && dt > 0) {
|
|
// The cumulative counter advanced but this update carries no
|
|
// directional bytes. This is the flow_purge case on 4.4.7:
|
|
// local_bytes/other_bytes are always 0 there while total_bytes is
|
|
// final, so the bytes since the last flow_stats would otherwise be
|
|
// lost (a few percent of every flow). Attribute dt in the same
|
|
// up/down proportion the flow has shown so far (account() maps
|
|
// local/other -> up/down by dev_local, so split in that frame),
|
|
// defaulting to download when the direction is not yet known.
|
|
let up_share = (fr.up + fr.down > 0) ? int(dt * fr.up / (fr.up + fr.down)) : 0;
|
|
let down_share = dt - up_share;
|
|
if (fr.up + fr.down == 0)
|
|
down_share = dt; // unknown: count as download
|
|
// local == up when dev_local, else local == down.
|
|
dl = fr.dev_local ? up_share : down_share;
|
|
do_ = fr.dev_local ? down_share : up_share;
|
|
}
|
|
} else if (final && tb > 0 && fr.total == 0 && !fr.late && (dl + do_) == 0) {
|
|
// PURGE-ONLY FLOW: a `flow` event created this record, no flow_stats ever
|
|
// arrived, and now the purge carries the final cumulative counter with
|
|
// both directional fields zero. The branch above cannot run because it
|
|
// requires fr.total > 0, so without this the flow's entire byte count
|
|
// would be silently discarded -- and stats.no_total would not move
|
|
// either, because tb > 0.
|
|
//
|
|
// MEASURED 2026-08-24 (EA8500, netifyd 4.4.7): this does NOT fire on
|
|
// this netifyd. Tapping the export socket for 170 s, 64 of 64 flows
|
|
// whose full lifecycle was observed carried a flow_stats before their
|
|
// purge, UDP DNS queries included. DESIGN §2.5 reached the same
|
|
// conclusion from byte totals; this reached it from the events.
|
|
//
|
|
// It is here because that behaviour is undocumented and the daemon had
|
|
// no defence if it ever changed. Late stubs are excluded: for those tb
|
|
// is pre-adoption history and injecting it would overcount.
|
|
stats.purge_only++;
|
|
dl = fr.dev_local ? 0 : tb; // account() maps local/other -> up/down
|
|
do_ = fr.dev_local ? tb : 0; // by dev_local; count it as download
|
|
stats.bytes_reported += tb;
|
|
} else if (tb <= 0) {
|
|
stats.no_total++;
|
|
}
|
|
if (tb > 0)
|
|
fr.total = tb;
|
|
|
|
account(fr, dl, do_, now);
|
|
}
|
|
|
|
function ev_flow_stats(ev, f, now) {
|
|
let fr = flows[f.digest];
|
|
if (!fr) {
|
|
// Stats for a digest we hold no flow event for. Two distinct causes
|
|
// with DIFFERENT outcomes; both characterised on hardware 2026-08-22
|
|
// (EA8500, netifyd 4.4.7, deconfounded LAN-local traffic):
|
|
//
|
|
// 1. A genuine stats-before-flow race on a NEW flow. The flow event
|
|
// lands shortly after and identity follows via flow_identify().
|
|
//
|
|
// 2. A flow already established when we connected or reconnected.
|
|
// netifyd 4.4.7 has no established-flow dump, so it emits
|
|
// flow_stats for such a flow but NEVER a flow event: identity
|
|
// never arrives and the bytes stay under Unknown until the flow
|
|
// ends. Measured by restarting this daemon mid-transfer, which
|
|
// stranded 10.7 MB of a correctly-classified HTTP flow (a fresh
|
|
// socket client saw 5 flow_stats and 0 flow events for it).
|
|
// Bounded to the connect/reconnect window, self-heals as flows
|
|
// cycle, and not fixable here without netifyd support.
|
|
//
|
|
// Either way the bytes are PARKED as Unknown, never dropped: byte
|
|
// conservation through this path is verified (DESIGN 2.5, 8).
|
|
fr = flow_new(f.digest, ev, f, now, true);
|
|
if (!fr)
|
|
return;
|
|
}
|
|
reshadow(fr, f, now); // close the stale-shadow double-count
|
|
flow_update(fr, f, now, false);
|
|
}
|
|
|
|
function ev_purge(ev, f, now) {
|
|
let fr = flows[f.digest];
|
|
stats.purges++;
|
|
if (!fr)
|
|
return;
|
|
// The purge carries the flow's final counter delta: account it, clamped like
|
|
// any other update, before the record goes away.
|
|
fr.reason = ev.reason ?? "purged";
|
|
flow_update(fr, f, now, true); // `final`: the purge-only rescue path
|
|
flow_drop(f.digest);
|
|
}
|
|
|
|
const HANDLERS = {
|
|
agent_hello: ev_hello,
|
|
definitions: ev_definitions,
|
|
agent_status: ev_agent_status
|
|
};
|
|
|
|
const FLOW_HANDLERS = {
|
|
flow: ev_flow,
|
|
flow_stats: ev_flow_stats,
|
|
flow_purge: ev_purge
|
|
};
|
|
|
|
function dispatch(ev, now) {
|
|
let t = ev.type;
|
|
if (!length(t ?? "")) {
|
|
stats.malformed++;
|
|
return;
|
|
}
|
|
|
|
stats.events++;
|
|
if (stats.by_type[t] != null)
|
|
stats.by_type[t]++;
|
|
else if (length(stats.by_type) < EVENT_TYPE_MAX)
|
|
stats.by_type[t] = 1;
|
|
stats.last_event_ms = now;
|
|
|
|
let h = HANDLERS[t];
|
|
if (h) {
|
|
h(ev);
|
|
return;
|
|
}
|
|
|
|
let f = ev.flow;
|
|
if (type(f) == "object" && length(f.digest ?? "")) {
|
|
h = FLOW_HANDLERS[t];
|
|
if (h) {
|
|
h(ev, f, now);
|
|
return;
|
|
}
|
|
// Unknown but flow-scoped: a purge-ish name finalises the flow, which
|
|
// keeps a future netifyd rename from leaking flow records.
|
|
if (match(t, PURGE_RE)) {
|
|
ev_purge(ev, f, now);
|
|
return;
|
|
}
|
|
}
|
|
|
|
stats.unknown++;
|
|
if (stats.unknown_types[t] == null && length(stats.unknown_types) < UNKNOWN_TYPE_MAX)
|
|
stats.unknown_types[t] = 0;
|
|
if (stats.unknown_types[t] != null)
|
|
stats.unknown_types[t]++;
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Wire protocol (§2.1): newline-delimited JSON, each payload preceded by a
|
|
* {"length": N} header line.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function feed(line, now) {
|
|
if (!length(line))
|
|
return;
|
|
|
|
let ev;
|
|
try {
|
|
ev = json(line);
|
|
} catch (e) {
|
|
stats.malformed++;
|
|
logonce("badjson", "warn", "malformed event line (%d bytes): %s", length(line), e);
|
|
return;
|
|
}
|
|
|
|
if (type(ev) != "object") {
|
|
stats.malformed++;
|
|
return;
|
|
}
|
|
|
|
// Frame header: sole key "length".
|
|
if (ev.length != null && length(ev) == 1) {
|
|
stats.frames++;
|
|
return;
|
|
}
|
|
|
|
dispatch(ev, now);
|
|
}
|
|
|
|
function on_readable() {
|
|
let now = now_ms();
|
|
|
|
for (let i = 0; i < RX_DRAIN_MAX; i++) {
|
|
let data;
|
|
try {
|
|
data = sock.recv(RX_CHUNK, socket.MSG_DONTWAIT);
|
|
} catch (e) {
|
|
disconnect("recv error: " + e);
|
|
return;
|
|
}
|
|
|
|
if (data == null) {
|
|
let err = socket.error(true);
|
|
if (err == 11 || err == 4) // EAGAIN / EINTR
|
|
return;
|
|
disconnect(sprintf("recv failed: %s", socket.error()));
|
|
return;
|
|
}
|
|
|
|
if (length(data) == 0) {
|
|
disconnect("netifyd closed the connection");
|
|
return;
|
|
}
|
|
|
|
stats.rx_bytes += length(data);
|
|
rxbuf += data;
|
|
|
|
if (length(rxbuf) > RX_MAX) {
|
|
// Never let a peer that stops sending newlines grow us without
|
|
// bound (§3.3). Drop back to a line boundary and carry on.
|
|
stats.resyncs++;
|
|
logonce("rxmax", "warn", "receive buffer overflow, resyncing");
|
|
let cut = rindex(rxbuf, "\n");
|
|
rxbuf = (cut >= 0) ? substr(rxbuf, cut + 1) : "";
|
|
}
|
|
|
|
let parts = split(rxbuf, "\n");
|
|
rxbuf = parts[length(parts) - 1];
|
|
for (let n = 0; n < length(parts) - 1; n++)
|
|
feed(trim(parts[n]), now);
|
|
|
|
if (length(data) < RX_CHUNK)
|
|
return;
|
|
}
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Connection management: start before netifyd, survive its restart.
|
|
* ------------------------------------------------------------------ */
|
|
|
|
disconnect = function(why) {
|
|
if (sock_fd) {
|
|
try {
|
|
sock_fd.delete();
|
|
} catch (e) {
|
|
try { sock_fd.cancel(); } catch (e2) {}
|
|
}
|
|
sock_fd = null;
|
|
}
|
|
if (sock) {
|
|
try { sock.close(); } catch (e) {}
|
|
sock = null;
|
|
}
|
|
rxbuf = "";
|
|
if (connected) {
|
|
connected = false;
|
|
logmsg("warn", "disconnected: %s", why);
|
|
}
|
|
schedule_reconnect();
|
|
};
|
|
|
|
schedule_reconnect = function() {
|
|
if (retry_timer)
|
|
return;
|
|
retry_timer = uloop.timer(backoff_ms, function() {
|
|
retry_timer = null;
|
|
connect();
|
|
});
|
|
backoff_ms = clampi(backoff_ms * 2, RECONNECT_MIN_MS, RECONNECT_MAX_MS);
|
|
};
|
|
|
|
connect = function() {
|
|
try {
|
|
sock = socket.connect(cfg.socket_path);
|
|
} catch (e) {
|
|
sock = null;
|
|
}
|
|
|
|
if (!sock) {
|
|
stats.connect_fail++;
|
|
logonce("noconn", "warn", "waiting for netifyd on %s", cfg.socket_path);
|
|
schedule_reconnect();
|
|
return;
|
|
}
|
|
|
|
sock_fd = uloop.handle(sock, on_readable, uloop.ULOOP_READ);
|
|
// Count reconnects, not the first connect, so `socket.reconnects` reads as
|
|
// churn that actually happened.
|
|
if (ever_connected)
|
|
stats.reconnects++;
|
|
ever_connected = true;
|
|
connected = true;
|
|
backoff_ms = RECONNECT_MIN_MS;
|
|
// A new socket is a new netifyd lifetime. saw_internal is a capability
|
|
// latch about the CURRENT stream ("is internal capture active"), so it must
|
|
// not carry over: a netifyd restarted external-only would otherwise keep the
|
|
// stale `true` and shadow every client flow to zero. Cleared here so the new
|
|
// stream re-establishes it from its own events.
|
|
saw_internal = false;
|
|
router_bootstrap();
|
|
logmsg("info", "connected to %s", cfg.socket_path);
|
|
};
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* Housekeeping (§3.3)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
prune = function(now, force) {
|
|
let cutoff = now - cfg.idle_ms;
|
|
let dropped = 0;
|
|
|
|
for (let d in keys(flows)) {
|
|
let fr = flows[d];
|
|
if (fr && fr.seen < cutoff) {
|
|
flow_drop(d);
|
|
dropped++;
|
|
}
|
|
}
|
|
|
|
// Cap pressure: shed the least recently active tenth of the table.
|
|
//
|
|
// Counted separately into flows_shed as well as into the pruned total, so
|
|
// "the flow table is too small for this network" is visible instead of
|
|
// hiding under ordinary idle housekeeping. See the note beside flows_shed.
|
|
if (force && nflows >= cfg.flow_max) {
|
|
let list = [];
|
|
for (let d in flows)
|
|
push(list, [ d, flows[d].seen ]);
|
|
sort(list, (a, b) => a[1] - b[1]);
|
|
let want = int(cfg.flow_max / 10) + 1;
|
|
for (let i = 0; i < want && i < length(list); i++) {
|
|
flow_drop(list[i][0]);
|
|
dropped++;
|
|
stats.flows_shed++;
|
|
}
|
|
}
|
|
|
|
stats.flows_pruned += dropped;
|
|
|
|
// Aggregates with no live flows that have been silent for longer than the
|
|
// trailing-hour window carry no data anyone can still query.
|
|
let agg_cutoff = now - (SERIES_SLOTS * SERIES_MS) - 60000;
|
|
for (let tbl in [ apps, devs, cats ])
|
|
for (let k in keys(tbl)) {
|
|
let a = tbl[k];
|
|
let last = a.cur.last ? a.cur.last : a.cur.first;
|
|
if (a.cur.flows == 0 && last < agg_cutoff) {
|
|
delete tbl[k];
|
|
stats.aggs_pruned++;
|
|
}
|
|
}
|
|
};
|
|
|
|
function housekeeping() {
|
|
let now = now_ms();
|
|
stats.sweeps++;
|
|
let slot = int(now / SERIES_MS), dslot = int(now / DEVSER_MS);
|
|
|
|
prune(now, false);
|
|
|
|
// Advance the trailing-hour rings and roll the rate windows. Rolling here
|
|
// (not only in agg_bump) means an aggregate that has gone idle still sees
|
|
// its rate decay to zero, and it keeps agg_rate a pure read.
|
|
agg_roll(totals, now);
|
|
ring_sweep(totals.buckets, slot, SERIES_SLOTS);
|
|
for (let tbl in [ apps, devs, cats ])
|
|
for (let k in tbl) {
|
|
agg_roll(tbl[k], now);
|
|
ring_sweep(tbl[k].buckets, slot, SERIES_SLOTS);
|
|
for (let dk in tbl[k].devs ?? {})
|
|
ring_sweep(tbl[k].devs[dk].ring, dslot, DEVSER_SLOTS);
|
|
for (let hk in tbl[k].hosts ?? {})
|
|
ring_sweep(tbl[k].hosts[hk].ring, dslot, DEVSER_SLOTS);
|
|
}
|
|
|
|
// §7.3 seam: v1 has no persistent buckets, so this hands over nothing.
|
|
store.flush(totals.buckets, slot);
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* ubus surface (§5)
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function rss_kb() {
|
|
let raw;
|
|
try {
|
|
raw = fs.readfile("/proc/self/status");
|
|
} catch (e) {
|
|
return 0;
|
|
}
|
|
let m = raw ? match(raw, /VmRSS:[ \t]*([0-9]+)/) : null;
|
|
return m ? int(m[1]) : 0;
|
|
}
|
|
|
|
function envelope(now) {
|
|
return { generated_at: now, agent_connected: connected };
|
|
}
|
|
|
|
function metrics(a, now) {
|
|
let r = agg_rate(a, now);
|
|
return {
|
|
bytes_up: a.cur.up,
|
|
bytes_down: a.cur.down,
|
|
bytes_total: a.cur.up + a.cur.down,
|
|
rate_up: r[0],
|
|
rate_down: r[1],
|
|
rate_total: r[0] + r[1],
|
|
flows: a.cur.flows,
|
|
first_seen: a.cur.first,
|
|
last_seen: a.cur.last ? a.cur.last : a.cur.first
|
|
};
|
|
}
|
|
|
|
function app_row(a, now) {
|
|
let e = metrics(a, now);
|
|
e.key = a.key;
|
|
e.name = a.name;
|
|
e.tag = a.tag ?? a.key;
|
|
e.category = a.cat ?? null;
|
|
return e;
|
|
}
|
|
|
|
function dev_row(a, now) {
|
|
let e = metrics(a, now);
|
|
let pseudo = (a.key == "router" || a.key == "multicast" || a.key == "unknown");
|
|
e.key = a.key;
|
|
e.name = device_name(a.key, a.mac, a.ip, now);
|
|
// The pseudo-devices collapse many endpoints; a single leftover MAC/IP
|
|
// under them would be misleading, so present them without one.
|
|
e.mac = pseudo ? null : a.mac;
|
|
e.ip = pseudo ? null : a.ip;
|
|
e.is_router = (a.router == true);
|
|
return e;
|
|
}
|
|
|
|
function cat_row(a, now) {
|
|
let e = metrics(a, now);
|
|
e.key = a.key;
|
|
e.name = a.name;
|
|
return e;
|
|
}
|
|
|
|
const SORTERS = {
|
|
rate: (a, b) => (b.rate_total - a.rate_total) || (b.bytes_total - a.bytes_total),
|
|
bytes: (a, b) => b.bytes_total - a.bytes_total,
|
|
up: (a, b) => b.bytes_up - a.bytes_up,
|
|
down: (a, b) => b.bytes_down - a.bytes_down,
|
|
flows: (a, b) => b.flows - a.flows,
|
|
last: (a, b) => b.last_seen - a.last_seen,
|
|
name: (a, b) => (a.name < b.name) ? -1 : ((a.name > b.name) ? 1 : 0)
|
|
};
|
|
|
|
function rows_of(tbl, rowfn, now, sortkey, limit) {
|
|
let out = [];
|
|
for (let k in tbl)
|
|
push(out, rowfn(tbl[k], now));
|
|
sort(out, SORTERS[sortkey ?? "rate"] ?? SORTERS.rate);
|
|
return (limit > 0 && length(out) > limit) ? slice(out, 0, limit) : out;
|
|
}
|
|
|
|
function find_app(want) {
|
|
if (!length(want ?? ""))
|
|
return null;
|
|
if (apps[want])
|
|
return apps[want];
|
|
let lw = lc(want);
|
|
for (let k in apps) {
|
|
let a = apps[k];
|
|
if (lc(a.tag ?? "") == lw || lc(a.name) == lw)
|
|
return a;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function m_summary(req) {
|
|
let now = now_ms();
|
|
let n = clampi(num(req.args?.limit, cfg.top_n), 1, LIST_REPLY_MAX);
|
|
let r = envelope(now);
|
|
r.window = cfg.rate_window;
|
|
r.totals = metrics(totals, now);
|
|
r.totals.flows = nflows;
|
|
r.top_apps = rows_of(apps, app_row, now, "rate", n);
|
|
r.top_devices = rows_of(devs, dev_row, now, "rate", n);
|
|
r.top_categories = rows_of(cats, cat_row, now, "rate", n);
|
|
r.counts = {
|
|
apps: length(apps),
|
|
devices: length(devs),
|
|
categories: length(cats),
|
|
flows: nflows
|
|
};
|
|
return r;
|
|
}
|
|
|
|
/* Application totals restricted to one device.
|
|
*
|
|
* The mirror of devices_for_apps(): same bounded flow table, same join already
|
|
* present on every flow record, grouped the other way. No new aggregate.
|
|
*
|
|
* As with that function these are the bytes CURRENTLY TRACKED flows account
|
|
* for, not the application's lifetime totals, and the reply says so.
|
|
*/
|
|
function apps_for_device(dev, now) {
|
|
let acc = {};
|
|
|
|
for (let dg in flows) {
|
|
let fr = flows[dg];
|
|
|
|
if (!fr || fr.shadow || fr.dev_key != dev || !fr.app_key)
|
|
continue;
|
|
|
|
let e = acc[fr.app_key];
|
|
if (!e) {
|
|
e = { key: fr.app_key, cat: fr.cat_key, up: 0, down: 0, flows: 0 };
|
|
acc[fr.app_key] = e;
|
|
}
|
|
|
|
e.up += (fr.up ?? 0);
|
|
e.down += (fr.down ?? 0);
|
|
e.flows++;
|
|
}
|
|
|
|
let out = [];
|
|
for (let k in acc) {
|
|
let e = acc[k];
|
|
/* Names live on the aggregate, not on the flow, so look the app up. A
|
|
* flow whose app has already been pruned from the aggregate still gets
|
|
* a row, keyed rather than named, instead of vanishing. */
|
|
let a = apps[k];
|
|
|
|
push(out, {
|
|
key: k,
|
|
name: a ? a.name : k,
|
|
tag: a ? (a.tag ?? k) : k,
|
|
category: e.cat,
|
|
upload: e.up,
|
|
download: e.down,
|
|
bytes_total: e.up + e.down,
|
|
flows: e.flows
|
|
});
|
|
}
|
|
|
|
sort(out, (x, y) => y.bytes_total - x.bytes_total);
|
|
|
|
return out;
|
|
}
|
|
|
|
function m_apps(req) {
|
|
let now = now_ms();
|
|
let r = envelope(now);
|
|
r.window = cfg.rate_window;
|
|
let lim = clampi(num(req.args?.limit, LIST_REPLY_MAX), 1, LIST_REPLY_MAX);
|
|
let dev = req.args?.device;
|
|
|
|
/* `device` scopes the answer to one device: what is that thing doing. */
|
|
if (type(dev) == "string" && length(dev)) {
|
|
let rows = apps_for_device(dev, now);
|
|
|
|
r.from_flows = true;
|
|
r.device = dev;
|
|
r.total = length(rows);
|
|
r.apps = (length(rows) > lim) ? slice(rows, 0, lim) : rows;
|
|
|
|
return r;
|
|
}
|
|
|
|
r.from_flows = false;
|
|
r.apps = rows_of(apps, app_row, now, req.args?.sort, lim);
|
|
r.total = length(apps);
|
|
return r;
|
|
}
|
|
|
|
/* Device totals restricted to a set of application keys.
|
|
*
|
|
* Computed from the live flow table rather than from a new aggregate. Every
|
|
* flow already carries dev_key and app_key, so the join is there; the table is
|
|
* bounded by flow_max, so this is bounded too. No new memory.
|
|
*
|
|
* The numbers are what the CURRENTLY TRACKED flows account for, not the
|
|
* device's lifetime totals: a flow that has ended is no longer here. The reply
|
|
* carries from_flows so a caller cannot present it as cumulative history.
|
|
*
|
|
* Shadow flows are excluded for the same reason they are excluded everywhere
|
|
* else: they are the NAT-side duplicate and counting them would double.
|
|
*/
|
|
function devices_for_apps(keys, now) {
|
|
let want = {};
|
|
for (let k in (keys ?? []))
|
|
if (type(k) == "string" && length(k))
|
|
want[k] = true;
|
|
|
|
let acc = {};
|
|
for (let dg in flows) {
|
|
let fr = flows[dg];
|
|
|
|
if (!fr || fr.shadow || !fr.dev_key || !want[fr.app_key])
|
|
continue;
|
|
|
|
let e = acc[fr.dev_key];
|
|
if (!e) {
|
|
e = { key: fr.dev_key, mac: fr.dev_mac, ip: fr.dev_ip,
|
|
up: 0, down: 0, flows: 0 };
|
|
acc[fr.dev_key] = e;
|
|
}
|
|
if (!e.ip && fr.dev_ip)
|
|
e.ip = fr.dev_ip;
|
|
|
|
e.up += (fr.up ?? 0);
|
|
e.down += (fr.down ?? 0);
|
|
e.flows++;
|
|
}
|
|
|
|
let out = [];
|
|
for (let k in acc) {
|
|
let e = acc[k];
|
|
let pseudo = (k == "router" || k == "multicast" || k == "unknown");
|
|
|
|
push(out, {
|
|
key: k,
|
|
name: device_name(k, e.mac, e.ip, now),
|
|
mac: pseudo ? null : e.mac,
|
|
ip: pseudo ? null : e.ip,
|
|
upload: e.up,
|
|
download: e.down,
|
|
bytes_total: e.up + e.down,
|
|
flows: e.flows
|
|
});
|
|
}
|
|
|
|
sort(out, (x, y) => y.bytes_total - x.bytes_total);
|
|
|
|
return out;
|
|
}
|
|
|
|
function m_devices(req) {
|
|
let now = now_ms();
|
|
let r = envelope(now);
|
|
r.window = cfg.rate_window;
|
|
let lim = clampi(num(req.args?.limit, LIST_REPLY_MAX), 1, LIST_REPLY_MAX);
|
|
|
|
/* `apps` restricts the answer to the devices carrying those applications,
|
|
* which is how the UI answers "who is generating this traffic" once the
|
|
* user has filtered the application list. Matching stays in the UI, where
|
|
* the user's text lives; summing stays here, where the flows are. */
|
|
let keys = req.args?.apps;
|
|
|
|
if (type(keys) == "array" && length(keys)) {
|
|
let rows = devices_for_apps(keys, now);
|
|
|
|
r.from_flows = true;
|
|
r.total = length(rows);
|
|
r.devices = (length(rows) > lim) ? slice(rows, 0, lim) : rows;
|
|
|
|
return r;
|
|
}
|
|
|
|
r.from_flows = false;
|
|
r.devices = rows_of(devs, dev_row, now, req.args?.sort, lim);
|
|
r.total = length(devs);
|
|
return r;
|
|
}
|
|
|
|
function m_app_detail(req) {
|
|
let now = now_ms();
|
|
let a = find_app(req.args?.app);
|
|
let r = envelope(now);
|
|
if (!a) {
|
|
r.found = false;
|
|
r.app = req.args?.app;
|
|
return r;
|
|
}
|
|
|
|
let slot = int(now / SERIES_MS), dslot = int(now / DEVSER_MS);
|
|
r.found = true;
|
|
r.app = a.key;
|
|
r.tag = a.tag ?? a.key;
|
|
r.label = a.name;
|
|
// The detail drawer is opened from the Statistics table, which is a
|
|
// trailing-hour view. Report the byte totals over that same window (from the
|
|
// hour ring) so the drawer header cannot disagree with the row that opened it;
|
|
// the earlier metrics(a) returned session-lifetime bytes, which read as a
|
|
// gross mismatch under an "last 60 minutes" caption. Rate and flow context
|
|
// stay as-is (a rate is inherently current).
|
|
let at = ring_total(a.buckets, slot, SERIES_SLOTS);
|
|
r.metrics = metrics(a, now);
|
|
r.metrics.bytes_down = at[0];
|
|
r.metrics.bytes_up = at[1];
|
|
r.metrics.bytes_total = at[0] + at[1];
|
|
r.metrics.window = "hour";
|
|
r.time_series = ring_series(a.buckets, slot, SERIES_SLOTS, SERIES_MS);
|
|
|
|
let ds = [];
|
|
for (let k in a.devs ?? {}) {
|
|
let e = a.devs[k];
|
|
let t = ring_total(e.ring, dslot, DEVSER_SLOTS);
|
|
push(ds, {
|
|
key: k,
|
|
mac: e.mac,
|
|
ip: e.ip,
|
|
name: device_name(k, e.mac, e.ip, now),
|
|
download: t[0],
|
|
upload: t[1],
|
|
total: t[0] + t[1],
|
|
last_seen: e.last
|
|
});
|
|
}
|
|
sort(ds, (x, y) => (y.total - x.total));
|
|
r.devices = ds;
|
|
|
|
let hs = [];
|
|
for (let h in a.hosts ?? {}) {
|
|
let ht = ring_total(a.hosts[h].ring, dslot, DEVSER_SLOTS);
|
|
push(hs, {
|
|
host: h,
|
|
download: ht[0],
|
|
upload: ht[1],
|
|
total: ht[0] + ht[1],
|
|
last_seen: a.hosts[h].last
|
|
});
|
|
}
|
|
sort(hs, (x, y) => (y.total - x.total));
|
|
r.hosts = hs;
|
|
|
|
return r;
|
|
}
|
|
|
|
function m_flows(req) {
|
|
let now = now_ms();
|
|
let want_app = req.args?.app, want_dev = req.args?.device;
|
|
let limit = clampi(num(req.args?.limit, 50), 1, FLOWS_REPLY_MAX);
|
|
let out = [];
|
|
|
|
for (let d in flows) {
|
|
let fr = flows[d];
|
|
if (length(want_app ?? "") && fr.app_key != want_app && fr.tag != want_app)
|
|
continue;
|
|
if (length(want_dev ?? "") && fr.dev_key != want_dev)
|
|
continue;
|
|
push(out, {
|
|
digest: substr(d, 0, 16),
|
|
app: fr.app_key,
|
|
app_name: fr.app_name,
|
|
category: fr.cat_key,
|
|
device: fr.dev_key,
|
|
device_name: device_name(fr.dev_key, fr.dev_mac, fr.dev_ip, now),
|
|
host: fr.host,
|
|
interface: fr.iface,
|
|
internal: fr.internal,
|
|
nat: fr.nat,
|
|
shadow: fr.shadow,
|
|
ip_protocol: fr.proto,
|
|
upload: fr.up,
|
|
download: fr.down,
|
|
// netifyd's cumulative lifetime counter for the flow, NOT the sum of
|
|
// what appflow attributed (up+down): a stub flow adopted mid-life has
|
|
// a large lifetime but small attributed bytes. Named distinctly so a
|
|
// consumer never confuses it with the aggregate `bytes_total`.
|
|
lifetime_bytes: fr.total,
|
|
first_seen: fr.start,
|
|
last_seen: fr.seen
|
|
});
|
|
}
|
|
|
|
sort(out, (x, y) => y.last_seen - x.last_seen);
|
|
let r = envelope(now);
|
|
r.total = length(out);
|
|
r.truncated = (length(out) > limit);
|
|
r.flows = r.truncated ? slice(out, 0, limit) : out;
|
|
return r;
|
|
}
|
|
|
|
// GL-parity trailing-hour view. Only range "hour" exists in v1; the argument
|
|
// is required so the contract can grow (day/week) without a breaking change.
|
|
function m_stats(req) {
|
|
let range = req.args?.range ?? "hour";
|
|
if (range != "hour")
|
|
return ubus.STATUS_INVALID_ARGUMENT;
|
|
|
|
let now = now_ms();
|
|
let slot = int(now / SERIES_MS);
|
|
let r = envelope(now);
|
|
r.range = "hour";
|
|
r.range_start = (slot - SERIES_SLOTS + 1) * SERIES_MS;
|
|
r.range_end = now;
|
|
|
|
let t = ring_total(totals.buckets, slot, SERIES_SLOTS);
|
|
r.all = { download: t[0], upload: t[1], total: t[0] + t[1] };
|
|
|
|
let list = [];
|
|
for (let k in apps) {
|
|
let a = apps[k];
|
|
let at = ring_total(a.buckets, slot, SERIES_SLOTS);
|
|
if ((at[0] + at[1]) == 0 && a.cur.flows == 0)
|
|
continue;
|
|
push(list, {
|
|
app: a.tag ?? a.key,
|
|
key: a.key,
|
|
label: a.name,
|
|
// An application's category is a property of its flows; the
|
|
// aggregate carries the most recent one seen.
|
|
category: a.cat ?? null,
|
|
download: at[0],
|
|
upload: at[1],
|
|
total: at[0] + at[1],
|
|
active_flows: a.cur.flows
|
|
});
|
|
}
|
|
sort(list, (x, y) => y.total - x.total);
|
|
|
|
// Only the leaders carry a time series -- twelve buckets per application
|
|
// would dominate the reply for no benefit.
|
|
let top = [];
|
|
for (let i = 0; i < length(list) && i < 10; i++) {
|
|
let e = list[i];
|
|
push(top, {
|
|
app: e.app, key: e.key, label: e.label, category: e.category,
|
|
download: e.download, upload: e.upload, total: e.total,
|
|
active_flows: e.active_flows,
|
|
time_series: ring_series(apps[e.key].buckets, slot, SERIES_SLOTS, SERIES_MS)
|
|
});
|
|
}
|
|
|
|
r.applications = list;
|
|
r.top_apps = top;
|
|
return r;
|
|
}
|
|
|
|
function m_status(req) {
|
|
let now = now_ms();
|
|
let r = envelope(now);
|
|
r.version = VERSION;
|
|
r.uptime = int((mono_ms() - start_mono) / 1000);
|
|
r.started_at = start_ms;
|
|
r.socket = { path: cfg.socket_path, connected: connected, reconnects: stats.reconnects,
|
|
connect_failures: stats.connect_fail };
|
|
r.config = { flow_max: cfg.flow_max, rate_window: cfg.rate_window,
|
|
top_n: cfg.top_n, idle_timeout: cfg.idle_timeout,
|
|
// 0/1 rather than a JSON boolean, so every value in this
|
|
// object is numeric and one reader parses all of them.
|
|
ai_breakout: cfg.ai_breakout ? 1 : 0 };
|
|
r.flows = { tracked: nflows, seen: stats.flows_seen, late: stats.flows_late,
|
|
pruned: stats.flows_pruned, shed: stats.flows_shed,
|
|
dropped: stats.flows_dropped,
|
|
shadowed: stats.shadowed, purged: stats.purges,
|
|
no_total: stats.no_total, ai_labeled: stats.ai_labeled };
|
|
// NAT twin resolution (3.2). `available` false means conntrack could not
|
|
// be read, in which case external captures are shadowed on the safe
|
|
// default and the router's own traffic is not attributed. A persistently
|
|
// high `unresolved` means entries are expiring before the flow is
|
|
// classified, which has the same effect for those flows.
|
|
r.conntrack = { available: conntrack.avail, entries: conntrack.entries,
|
|
twins: stats.ct_twins, router_local: stats.ct_local,
|
|
unresolved: stats.ct_unresolved };
|
|
// Byte conservation (§2.5). `leaked` is what netifyd reported growing
|
|
// that neither reached an aggregate nor was deliberately shadowed as a NAT
|
|
// twin. It should sit at 0. A number that climbs is a byte-accounting
|
|
// defect, which is exactly the class that was previously invisible here.
|
|
r.bytes = { reported: stats.bytes_reported,
|
|
attributed: stats.bytes_attributed,
|
|
shadowed: stats.bytes_shadowed,
|
|
leaked: stats.bytes_reported - stats.bytes_attributed - stats.bytes_shadowed,
|
|
purge_only_rescued: stats.purge_only };
|
|
r.aggregates = { apps: length(apps), devices: length(devs),
|
|
categories: length(cats), pruned: stats.aggs_pruned,
|
|
refused: stats.aggs_refused };
|
|
r.events = { total: stats.events, frames: stats.frames, malformed: stats.malformed,
|
|
unknown: stats.unknown, unknown_types: stats.unknown_types,
|
|
by_type: stats.by_type, rx_bytes: stats.rx_bytes, resyncs: stats.resyncs,
|
|
last_event_age_ms: stats.last_event_ms ? (now - stats.last_event_ms) : -1 };
|
|
r.agent = { version: agent.version, uptime: agent.uptime,
|
|
flow_count: agent.flow_count, maxrss_kb: agent.rss_kb ?? 0,
|
|
last_status_at: agent.ts };
|
|
r.housekeeping = { sweeps: stats.sweeps, interval_ms: PRUNE_MS };
|
|
r.memory = { rss_kb: rss_kb() };
|
|
r.last_reset = stats.last_reset_ms;
|
|
return r;
|
|
}
|
|
|
|
// "Clear" — zero every accumulator while keeping flow tracking intact, so the
|
|
// numbers restart from now without losing the live flow context.
|
|
function m_reset(req) {
|
|
let now = now_ms();
|
|
agg_reset(totals, now);
|
|
for (let tbl in [ apps, devs, cats ])
|
|
for (let k in tbl) {
|
|
agg_reset(tbl[k], now);
|
|
delete tbl[k].devs;
|
|
delete tbl[k].hosts;
|
|
}
|
|
// Zero each flow's attributed counters, but NOT flows[].total: that mirrors
|
|
// netifyd's cumulative lifetime counter and is the baseline the next
|
|
// flow_stats delta is measured against. Clearing it would make the next
|
|
// update compute a false multi-KB spike. flows[].lifetime_bytes therefore
|
|
// survives a reset by design.
|
|
for (let d in flows) {
|
|
flows[d].up = 0;
|
|
flows[d].down = 0;
|
|
}
|
|
stats.last_reset_ms = now;
|
|
logmsg("info", "statistics reset via ubus");
|
|
let r = envelope(now);
|
|
r.reset = true;
|
|
return r;
|
|
}
|
|
|
|
function ubus_publish(conn) {
|
|
// Every args policy accepts ubus_rpc_session: rpcd injects it into all
|
|
// LuCI-proxied calls, and an unknown typed field is rejected as EINVAL.
|
|
const SID = { ubus_rpc_session: "" };
|
|
return conn.publish("appflow", {
|
|
summary: { call: m_summary, args: { limit: 0, ...SID } },
|
|
/* `device` must be declared or ubus rejects the call outright, which
|
|
* presents as the method returning nothing at all. */
|
|
apps: { call: m_apps, args: { sort: "", limit: 0, device: "", ...SID } },
|
|
/* `apps` is an array of application keys. It MUST be declared here or
|
|
* ubus rejects the call outright rather than ignoring the argument,
|
|
* which presents as the method returning nothing at all. */
|
|
devices: { call: m_devices, args: { sort: "", limit: 0, apps: [], ...SID } },
|
|
app_detail: { call: m_app_detail, args: { app: "", ...SID } },
|
|
flows: { call: m_flows, args: { limit: 0, app: "", device: "", ...SID } },
|
|
stats: { call: m_stats, args: { range: "", ...SID } },
|
|
status: { call: m_status, args: { ...SID } },
|
|
reset: { call: m_reset, args: { ...SID } }
|
|
});
|
|
// Reserved for §7.4: history { since, until } — deliberately not published
|
|
// in v1 so the method name stays free for the persistent-store version.
|
|
}
|
|
|
|
/* ------------------------------------------------------------------ *
|
|
* main
|
|
* ------------------------------------------------------------------ */
|
|
|
|
function main() {
|
|
start_ms = now_ms();
|
|
start_mono = mono_ms();
|
|
|
|
cfg_load();
|
|
totals = agg_new("total", "total", "Total", start_ms);
|
|
load_categories();
|
|
router_bootstrap();
|
|
|
|
uloop.init();
|
|
|
|
let conn;
|
|
try {
|
|
conn = ubus.connect();
|
|
} catch (e) {
|
|
logmsg("error", "ubus connect failed: %s", e);
|
|
exit(1);
|
|
}
|
|
|
|
try {
|
|
ubus_obj = ubus_publish(conn);
|
|
} catch (e) {
|
|
logmsg("error", "ubus publish failed: %s", e);
|
|
exit(1);
|
|
}
|
|
|
|
for (let sig in [ "SIGTERM", "SIGINT" ])
|
|
try {
|
|
uloop.signal(sig, function() { uloop.end(); });
|
|
} catch (e) {
|
|
// signal handling is a nicety; procd will SIGKILL if we ignore it
|
|
}
|
|
|
|
uloop.interval(PRUNE_MS, housekeeping);
|
|
|
|
logmsg("info", "%s %s started (flow_max=%d rate_window=%ds)",
|
|
NAME, VERSION, cfg.flow_max, cfg.rate_window);
|
|
connect();
|
|
|
|
uloop.run();
|
|
uloop.done();
|
|
|
|
if (sock)
|
|
try { sock.close(); } catch (e) {}
|
|
logmsg("info", "exiting");
|
|
}
|
|
|
|
main();
|