feat(api, web): add Linux nft destination port hits for blocked IPs

Track tcp/udp dports via deny_port_hits, expose aggregate and per-IP ports in UI; install-link re-run refreshes nft rules.

Co-authored-by: Cursor <[email protected]>

packages/db/migrations/010_agent_port_block_stats.sql

@@ -0,0 +1,21 @@
+-- Per-(ip, protocol, port) deny drop counters from Linux nft dynamic set.
+CREATE TABLE IF NOT EXISTS agent_port_block_stats (
+  id TEXT PRIMARY KEY NOT NULL,
+  agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+  ip TEXT NOT NULL,
+  port INTEGER NOT NULL,
+  protocol TEXT NOT NULL,
+  packets INTEGER NOT NULL DEFAULT 0,
+  last_reported_packets INTEGER NOT NULL DEFAULT 0,
+  first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+  last_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_port_block_stats_agent_ip_port_proto
+  ON agent_port_block_stats(agent_id, ip, port, protocol);
+
+CREATE INDEX IF NOT EXISTS idx_agent_port_block_stats_agent_packets
+  ON agent_port_block_stats(agent_id, packets);
+
+CREATE INDEX IF NOT EXISTS idx_agent_port_block_stats_agent_port_proto
+  ON agent_port_block_stats(agent_id, port, protocol);

packages/db/src/repositories/index.ts

@@ -61,9 +61,21 @@ export {
   listIpBlockStats,
   deleteIpBlockStatsForAgent,
   resetIpBlockStatsBaselines,
+  upsertPortBlockStats,
+  listPortBlockStatsAggregate,
+  listPortBlockStatsForIps,
+  mapTopPortsByIp,
+  deletePortBlockStatsForAgent,
+  resetPortBlockStatsBaselines,
   PRESENCE_REHIT_STALE_MS,
 } from './stats.js'
-export type { UpsertIpBlockStatsOptions, IpHitInput } from './stats.js'
+export type {
+  UpsertIpBlockStatsOptions,
+  IpHitInput,
+  PortHitInput,
+  PortBlockAggregateRow,
+  PortBlockPerIpRow,
+} from './stats.js'
 
 export {
   getSetting,
@@ -142,6 +154,12 @@ import {
   listIpBlockStats,
   deleteIpBlockStatsForAgent,
   resetIpBlockStatsBaselines,
+  upsertPortBlockStats,
+  listPortBlockStatsAggregate,
+  listPortBlockStatsForIps,
+  mapTopPortsByIp,
+  deletePortBlockStatsForAgent,
+  resetPortBlockStatsBaselines,
 } from './stats.js'
 import {
   getSetting,
@@ -212,6 +230,12 @@ export const repos = {
   listIpBlockStats,
   deleteIpBlockStatsForAgent,
   resetIpBlockStatsBaselines,
+  upsertPortBlockStats,
+  listPortBlockStatsAggregate,
+  listPortBlockStatsForIps,
+  mapTopPortsByIp,
+  deletePortBlockStatsForAgent,
+  resetPortBlockStatsBaselines,
   getSetting,
   setSetting,
   listSettings,

packages/db/src/repositories/stats.ts

@@ -1,6 +1,10 @@
-import { and, eq, desc } from 'drizzle-orm'
+import { and, eq, desc, sql, inArray } from 'drizzle-orm'
 import type { Db } from '../client.js'
-import { agentIpBlockStats, agentStatsSamples } from '../schema.js'
+import {
+  agentIpBlockStats,
+  agentPortBlockStats,
+  agentStatsSamples,
+} from '../schema.js'
 
 export function insertStatsSample(
   db: Db,
@@ -172,3 +176,169 @@ export function resetIpBlockStatsBaselines(db: Db, agentId: string) {
     .where(eq(agentIpBlockStats.agentId, agentId))
     .run()
 }
+
+export type PortHitInput = {
+  ip: string
+  port: number
+  protocol: 'tcp' | 'udp' | string
+  packets: number
+}
+
+/**
+ * Upsert per-(ip, proto, port) deny counters (Linux nft absolute deltas).
+ * deny_port_hits is not flushed on policy apply (timeout ages elements), so
+ * baselines are not auto-reset on Traffic flush — only on stats/reset delete.
+ */
+export function upsertPortBlockStats(
+  db: Db,
+  agentId: string,
+  hits: PortHitInput[],
+  now = new Date().toISOString(),
+) {
+  for (const hit of hits) {
+    const ip = hit.ip.trim()
+    const protocol = String(hit.protocol).toLowerCase()
+    if (!ip || (protocol !== 'tcp' && protocol !== 'udp')) continue
+    const port = Math.floor(hit.port)
+    if (port < 1 || port > 65535) continue
+    const reported = Math.max(0, Math.floor(hit.packets))
+    const existing = db
+      .select()
+      .from(agentPortBlockStats)
+      .where(
+        and(
+          eq(agentPortBlockStats.agentId, agentId),
+          eq(agentPortBlockStats.ip, ip),
+          eq(agentPortBlockStats.port, port),
+          eq(agentPortBlockStats.protocol, protocol),
+        ),
+      )
+      .get()
+
+    if (!existing) {
+      db.insert(agentPortBlockStats)
+        .values({
+          id: crypto.randomUUID(),
+          agentId,
+          ip,
+          port,
+          protocol,
+          packets: reported,
+          lastReportedPackets: reported,
+          firstSeenAt: now,
+          lastSeenAt: now,
+        })
+        .run()
+      continue
+    }
+
+    const prevReported = existing.lastReportedPackets ?? 0
+    const delta =
+      reported >= prevReported ? reported - prevReported : reported
+    const packets = (existing.packets ?? 0) + delta
+    db.update(agentPortBlockStats)
+      .set({
+        packets,
+        lastReportedPackets: reported,
+        ...(delta > 0 ? { lastSeenAt: now } : {}),
+      })
+      .where(eq(agentPortBlockStats.id, existing.id))
+      .run()
+  }
+}
+
+export type PortBlockAggregateRow = {
+  port: number
+  protocol: string
+  packets: number
+  lastSeenAt: string
+}
+
+export function listPortBlockStatsAggregate(
+  db: Db,
+  agentId: string,
+  limit = 50,
+): PortBlockAggregateRow[] {
+  const rows = db
+    .select({
+      port: agentPortBlockStats.port,
+      protocol: agentPortBlockStats.protocol,
+      packets: sql<number>`sum(${agentPortBlockStats.packets})`.mapWith(Number),
+      lastSeenAt: sql<string>`max(${agentPortBlockStats.lastSeenAt})`,
+    })
+    .from(agentPortBlockStats)
+    .where(eq(agentPortBlockStats.agentId, agentId))
+    .groupBy(agentPortBlockStats.port, agentPortBlockStats.protocol)
+    .orderBy(sql`sum(${agentPortBlockStats.packets}) desc`)
+    .limit(limit)
+    .all()
+  return rows.map((r) => ({
+    port: r.port,
+    protocol: r.protocol,
+    packets: r.packets ?? 0,
+    lastSeenAt: r.lastSeenAt,
+  }))
+}
+
+export type PortBlockPerIpRow = {
+  ip: string
+  port: number
+  protocol: string
+  packets: number
+}
+
+/** Raw rows for a set of IPs (caller picks top-N per IP). */
+export function listPortBlockStatsForIps(
+  db: Db,
+  agentId: string,
+  ips: string[],
+): PortBlockPerIpRow[] {
+  if (ips.length === 0) return []
+  return db
+    .select({
+      ip: agentPortBlockStats.ip,
+      port: agentPortBlockStats.port,
+      protocol: agentPortBlockStats.protocol,
+      packets: agentPortBlockStats.packets,
+    })
+    .from(agentPortBlockStats)
+    .where(
+      and(
+        eq(agentPortBlockStats.agentId, agentId),
+        inArray(agentPortBlockStats.ip, ips),
+      ),
+    )
+    .orderBy(desc(agentPortBlockStats.packets))
+    .all()
+}
+
+/** Top `perIpLimit` ports per IP, keyed by IP. */
+export function mapTopPortsByIp(
+  db: Db,
+  agentId: string,
+  ips: string[],
+  perIpLimit = 5,
+): Map<string, PortBlockPerIpRow[]> {
+  const all = listPortBlockStatsForIps(db, agentId, ips)
+  const map = new Map<string, PortBlockPerIpRow[]>()
+  for (const row of all) {
+    const list = map.get(row.ip) ?? []
+    if (list.length >= perIpLimit) continue
+    list.push(row)
+    map.set(row.ip, list)
+  }
+  return map
+}
+
+export function deletePortBlockStatsForAgent(db: Db, agentId: string) {
+  db.delete(agentPortBlockStats)
+    .where(eq(agentPortBlockStats.agentId, agentId))
+    .run()
+}
+
+export function resetPortBlockStatsBaselines(db: Db, agentId: string) {
+  db.update(agentPortBlockStats)
+    .set({ lastReportedPackets: 0 })
+    .where(eq(agentPortBlockStats.agentId, agentId))
+    .run()
+}

packages/db/src/schema.ts

@@ -228,6 +228,42 @@ export const agentIpBlockStats = sqliteTable(
   }),
 )
 
+/** Per-(ip, proto, dport) deny hits from Linux nft dynamic concat set. */
+export const agentPortBlockStats = sqliteTable(
+  'agent_port_block_stats',
+  {
+    id: text('id').primaryKey(),
+    agentId: text('agent_id')
+      .notNull()
+      .references(() => agents.id, { onDelete: 'cascade' }),
+    ip: text('ip').notNull(),
+    port: integer('port').notNull(),
+    protocol: text('protocol').notNull(), // tcp | udp
+    packets: integer('packets').notNull().default(0),
+    lastReportedPackets: integer('last_reported_packets').notNull().default(0),
+    firstSeenAt: text('first_seen_at')
+      .notNull()
+      .default(sql`(strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))`),
+    lastSeenAt: text('last_seen_at')
+      .notNull()
+      .default(sql`(strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))`),
+  },
+  (t) => ({
+    agentIpPortProto: uniqueIndex(
+      'idx_agent_port_block_stats_agent_ip_port_proto',
+    ).on(t.agentId, t.ip, t.port, t.protocol),
+    agentPackets: index('idx_agent_port_block_stats_agent_packets').on(
+      t.agentId,
+      t.packets,
+    ),
+    agentPortProto: index('idx_agent_port_block_stats_agent_port_proto').on(
+      t.agentId,
+      t.port,
+      t.protocol,
+    ),
+  }),
+)
+
 /** Short install invite links (`/agent-install/:id` and `/:slug`). */
 export const agentInstallLinks = sqliteTable(
   'agent_install_links',
@@ -291,6 +327,7 @@ export const schema = {
   ipOverrides,
   agentStatsSamples,
   agentIpBlockStats,
+  agentPortBlockStats,
   agentInstallLinks,
   auditLog,
 }

packages/shared/src/contracts.ts

@@ -226,6 +226,13 @@ export const applyReportIpHitSchema = z.object({
   packets: z.number().int().nonnegative(),
 })
 
+export const applyReportPortHitSchema = z.object({
+  ip: z.string().min(1).max(64),
+  port: z.number().int().min(1).max(65535),
+  protocol: z.enum(['tcp', 'udp']),
+  packets: z.number().int().nonnegative(),
+})
+
 export const applyReportBodySchema = z.object({
   status: z.string(),
   prefix_count: z.number().int().optional(),
@@ -236,6 +243,14 @@ export const applyReportBodySchema = z.object({
   source: z.string().optional(),
   /** Linux nft/ipset per-element drop counters (top-N, packets > 0). */
   ip_hits: z.array(applyReportIpHitSchema).max(200).optional(),
+  /** Linux nft dynamic set per-(ip, proto, dport) deny hits (top-N). */
+  port_hits: z.array(applyReportPortHitSchema).max(500).optional(),
+})
+
+export const agentIpPortStatSchema = z.object({
+  port: z.number().int(),
+  protocol: z.enum(['tcp', 'udp']),
+  packets: z.number().int(),
 })
 
 export const agentIpBlockStatSchema = z.object({
@@ -243,6 +258,14 @@ export const agentIpBlockStatSchema = z.object({
   packets: z.number().int(),
   first_seen_at: z.string(),
   last_seen_at: z.string(),
+  ports: z.array(agentIpPortStatSchema).optional(),
+})
+
+export const agentPortBlockStatSchema = z.object({
+  port: z.number().int(),
+  protocol: z.enum(['tcp', 'udp']),
+  packets: z.number().int(),
+  last_seen_at: z.string(),
 })
 
 export const agentPolicySchema = z.object({