Add per-device VPN and direct routing policies

test/server/dataplane-client.test.js

@@ -4,8 +4,8 @@ import { createDataplaneClient } from '../../src/server/dataplaneClient.js';
 
 test('control uses the dataplane socket protocol', async () => {
   const requests = [];
-  const send = async (socketPath, pathname, method) => {
-    requests.push(`${method} ${pathname} ${socketPath}`);
+  const send = async (socketPath, pathname, method, body) => {
+    requests.push({ method, pathname, socketPath, body });
     return {
       running: pathname !== '/stop',
       startedAt: 'now',
@@ -24,15 +24,20 @@ test('control uses the dataplane socket protocol', async () => {
   assert.equal(devices.running, true);
   const traffic = await client.observeTraffic();
   assert.equal(traffic.running, true);
+  await client.observeDevicePolicy();
+  await client.applyDevicePolicies([{ id: 'dev_0011223344556677' }]);
   assert.equal(client.running, true);
   await client.restart();
   assert.equal((await client.stop()).running, false);
-  assert.deepEqual(requests, [
+  assert.deepEqual(requests.map(({ method, pathname, socketPath }) => `${method} ${pathname} ${socketPath}`), [
     'GET /status /run/dataplane.sock',
     'POST /apply /run/dataplane.sock',
     'GET /devices /run/dataplane.sock',
     'GET /device-traffic /run/dataplane.sock',
+    'GET /device-policy /run/dataplane.sock',
+    'PUT /device-policy /run/dataplane.sock',
     'POST /restart /run/dataplane.sock',
     'POST /stop /run/dataplane.sock',
   ]);
+  assert.deepEqual(requests[5].body, { devices: [{ id: 'dev_0011223344556677' }] });
 });

test/server/deploy-split.test.js

@@ -17,7 +17,7 @@ test('gateway deploy updates control without recreating dataplane', () => {
   assert.match(deploy, /up -d --no-deps --wait[^\n]+vpn-proxy-control/);
   assert.match(workflow, /UPDATE_DATAPLANE="\$\{UPDATE_DATAPLANE\}"/);
   assert.match(workflow, /src\/server\/\(config\|dataplane\|gatewayRouting\|singboxRuntime\|version\)/);
-  assert.match(workflow, /src\/server\/\(adapters\/neighbors\|services\/deviceTrafficService\)/);
+  assert.match(workflow, /src\/server\/\(adapters\/neighbors\|services\/\(deviceTrafficService\|devicePolicyService\)\)/);
   assert.match(workflow, /src\/shared\/errors/);
   assert.doesNotMatch(workflow, /dataplaneClient/);
 });

test/server/device-inventory.test.js

@@ -10,6 +10,7 @@ import {
   DEVICE_INVENTORY_SCHEMA_VERSION,
   migrateDeviceInventoryState,
 } from '../../src/server/services/deviceInventoryService.js';
+import { fingerprintDirectDevices } from '../../src/server/services/devicePolicyService.js';
 import { createJsonStore } from '../../src/server/services/stateStore.js';
 
 test('device inventory discovers, merges, persists metadata and expires anonymous devices', async (t) => {
@@ -196,6 +197,143 @@ test('device traffic totals persist exact deltas across polls and process epochs
   assert.equal(snapshot.source.traffic.error, 'traffic unavailable');
 });
 
+test('pinned device policy persists, reconciles the full set, and keeps the last applied mode on failure', async (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-policy-'));
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
+  const store = createJsonStore({
+    filePath: path.join(directory, 'devices.json'),
+    defaultValue: {},
+    migrate: migrateDeviceInventoryState,
+    initializeMissing: true,
+    backupWhen: () => true,
+  });
+  const observedAt = '2026-08-07T12:00:00.000Z';
+  const mac = '00:11:22:33:44:55';
+  let observations = [{
+    ip: '192.168.50.7',
+    mac,
+    interface: 'eth0',
+    observedAt,
+    active: true,
+  }];
+  let activeDevices = [];
+  let generation = 0;
+  let policyEpoch = 'policy-epoch-a';
+  let failApply = false;
+  let malformedAck = false;
+  const appliedSets = [];
+  const policySnapshot = () => ({
+    epoch: policyEpoch,
+    generation: `policy-rules-${generation}`,
+    fingerprint: fingerprintDirectDevices(activeDevices),
+    observedAt,
+    appliedIds: activeDevices.map(({ id }) => id),
+  });
+  const createService = () => createDeviceInventoryService({
+    store,
+    observe: () => ({ observedAt, observations, error: null }),
+    observePolicy: policySnapshot,
+    applyPolicies: async (devices) => {
+      appliedSets.push(structuredClone(devices));
+      if (failApply) throw new Error('iptables unavailable');
+      if (malformedAck) return { ...policySnapshot(), fingerprint: 'invalid' };
+      activeDevices = structuredClone(devices);
+      generation += 1;
+      return policySnapshot();
+    },
+  });
+
+  let service = createService();
+  let snapshot = await service.refresh();
+  const id = snapshot.devices[0].id;
+  assert.equal(snapshot.devices[0].desiredPolicy, 'vpn');
+  assert.equal(snapshot.devices[0].appliedPolicy, 'vpn');
+
+  snapshot = service.update(id, { pinned: true }, snapshot.revision);
+  snapshot = await service.setPolicy(id, 'direct', snapshot.revision);
+  assert.deepEqual(appliedSets.at(-1), [{ id, ip: '192.168.50.7', mac, interface: 'eth0' }]);
+  assert.equal(snapshot.devices[0].desiredPolicy, 'direct');
+  assert.equal(snapshot.devices[0].appliedPolicy, 'direct');
+  assert.equal(snapshot.devices[0].policyStatus, 'applied');
+  assert.equal(store.read().schemaVersion, 2);
+  assert.equal(store.read().policy.schemaVersion, 1);
+
+  service = createService();
+  snapshot = await service.reconcilePolicies();
+  assert.equal(snapshot.devices[0].appliedPolicy, 'direct');
+  assert.throws(
+    () => service.update(id, { pinned: false }, snapshot.revision),
+    (error) => error.code === 'REQUEST_INVALID',
+  );
+
+  policyEpoch = 'policy-epoch-b';
+  activeDevices = [];
+  failApply = true;
+  snapshot = await service.refresh();
+  assert.equal(snapshot.devices[0].desiredPolicy, 'direct');
+  assert.equal(snapshot.devices[0].appliedPolicy, 'vpn');
+  assert.equal(snapshot.devices[0].policyStatus, 'failed');
+  failApply = false;
+  snapshot = await service.refresh();
+  assert.equal(snapshot.devices[0].appliedPolicy, 'direct');
+  assert.equal(snapshot.devices[0].policyStatus, 'applied');
+
+  observations = [
+    { ...observations[0], interface: 'eth0' },
+    { ...observations[0], interface: 'eth1' },
+  ];
+  snapshot = await service.refresh();
+  assert.deepEqual(appliedSets.at(-1), []);
+  assert.equal(snapshot.devices[0].confidence, 'ambiguous');
+  assert.equal(snapshot.devices[0].desiredPolicy, 'direct');
+  assert.equal(snapshot.devices[0].appliedPolicy, 'vpn');
+  assert.equal(snapshot.devices[0].policyStatus, 'pending');
+
+  snapshot = await service.setPolicy(id, 'vpn', snapshot.revision);
+  assert.equal(snapshot.devices[0].policyStatus, 'applied');
+  observations = [{ ...observations[0], interface: 'eth0' }];
+  snapshot = await service.refresh();
+  snapshot = await service.setPolicy(id, 'direct', snapshot.revision);
+  const secondMac = '00:11:22:33:44:66';
+  observations.push({
+    ip: '192.168.50.8',
+    mac: secondMac,
+    interface: 'eth0',
+    observedAt,
+    active: true,
+  });
+  snapshot = await service.refresh();
+  const secondId = snapshot.devices.find((device) => device.mac === secondMac).id;
+  snapshot = service.update(secondId, { pinned: true }, snapshot.revision);
+  failApply = true;
+  await assert.rejects(
+    service.setPolicy(secondId, 'direct', snapshot.revision),
+    (error) => error.code === 'DEVICE_POLICY_APPLY_FAILED',
+  );
+  snapshot = service.snapshot();
+  const byId = new Map(snapshot.devices.map((device) => [device.id, device]));
+  assert.equal(byId.get(id).desiredPolicy, 'direct');
+  assert.equal(byId.get(id).appliedPolicy, 'direct');
+  assert.equal(byId.get(id).policyStatus, 'applied');
+  assert.equal(byId.get(secondId).desiredPolicy, 'direct');
+  assert.equal(byId.get(secondId).appliedPolicy, 'vpn');
+  assert.equal(byId.get(secondId).policyStatus, 'failed');
+  assert.equal(snapshot.source.policy.error, 'iptables unavailable');
+  failApply = false;
+  malformedAck = true;
+  await assert.rejects(
+    service.setPolicy(secondId, 'direct', snapshot.revision),
+    (error) => error.code === 'DEVICE_POLICY_APPLY_FAILED',
+  );
+  snapshot = service.snapshot();
+  assert.equal(snapshot.devices.find((device) => device.id === secondId).appliedPolicy, 'vpn');
+  assert.match(snapshot.source.policy.error, /acknowledgement/);
+  assert.throws(
+    () => service.setPolicy(secondId, 'block', snapshot.revision),
+    (error) => error.code === 'REQUEST_INVALID',
+  );
+});
+
 test('device inventory v1 migration creates a versioned backup', (t) => {
   const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-migration-'));
   t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
@@ -210,11 +348,47 @@ test('device inventory v1 migration creates a versioned backup', (t) => {
 
   const migrated = store.read();
   assert.equal(migrated.schemaVersion, DEVICE_INVENTORY_SCHEMA_VERSION);
+  assert.equal(migrated.policy.schemaVersion, 1);
   assert.deepEqual(migrated.traffic.baselinesByMac, {});
   assert.match(store.migration?.backupPath || '', /\.backup-v1-/);
   assert.ok(fs.existsSync(store.migration.backupPath));
 });
 
+test('device inventory backs up and normalizes a malformed additive policy checkpoint', (t) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-policy-migration-'));
+  t.after(() => fs.rmSync(directory, { recursive: true, force: true }));
+  const filePath = path.join(directory, 'devices.json');
+  fs.writeFileSync(filePath, JSON.stringify({
+    schemaVersion: 2,
+    revision: 3,
+    devices: [],
+    traffic: { baselinesByMac: {}, totalsByMac: {}, rebaselineMacs: [] },
+    policy: {
+      schemaVersion: 1,
+      defaultMode: 'vpn',
+      dataplaneEpoch: 42,
+      byMac: {
+        broken: { desired: 'direct', applied: 'vpn', status: 'failed' },
+      },
+    },
+  }));
+  const store = createJsonStore({
+    filePath,
+    defaultValue: {},
+    migrate: migrateDeviceInventoryState,
+    backupWhen: () => true,
+  });
+
+  const migrated = store.read();
+  assert.equal(migrated.schemaVersion, 2);
+  assert.equal(migrated.policy.schemaVersion, 1);
+  assert.equal(migrated.policy.dataplaneEpoch, null);
+  assert.deepEqual(migrated.policy.byMac, {});
+  assert.match(migrated.policy.lastError, /device policy checkpoint/);
+  assert.match(store.migration?.backupPath || '', /\.backup-v2-/);
+  assert.ok(fs.existsSync(store.migration.backupPath));
+});
+
 test('device inventory backs up malformed v2 traffic and re-baselines without double counting', async (t) => {
   const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-device-corrupt-traffic-'));
   t.after(() => fs.rmSync(directory, { recursive: true, force: true }));

test/server/device-policy.test.js

@@ -0,0 +1,125 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  buildDevicePolicyRestore,
+  createDevicePolicyService,
+  fingerprintDirectDevices,
+  normalizeDirectDevices,
+} from '../../src/server/services/devicePolicyService.js';
+
+const directDevice = {
+  id: 'dev_0011223344556677',
+  ip: '192.168.50.7',
+  mac: '00:11:22:33:44:55',
+  interface: 'eth0',
+};
+
+test('device policy rules match the full identity before the TPROXY fallback', () => {
+  assert.deepEqual(normalizeDirectDevices([{ ...directDevice, mac: directDevice.mac.toUpperCase() }]), [directDevice]);
+  assert.throws(() => normalizeDirectDevices([directDevice, directDevice]), /повторяющаяся/);
+  assert.throws(() => normalizeDirectDevices([{ ...directDevice, interface: 'br-user' }]), /identity/);
+
+  const restore = buildDevicePolicyRestore({
+    devices: [directDevice],
+    chain: 'VPN_PROXY_DEVICE_POLICY',
+    slot: 'B',
+    tproxyPort: 7895,
+    tproxyMark: '1',
+  });
+  assert.equal(restore, [
+    '*mangle',
+    '-F VPN_PROXY_DEVICE_POLICY_B',
+    '-A VPN_PROXY_DEVICE_POLICY_B -i eth0 -s 192.168.50.7 -m mac --mac-source 00:11:22:33:44:55 -m comment --comment harbor-policy:dev_0011223344556677:direct -j RETURN',
+    '-A VPN_PROXY_DEVICE_POLICY_B -p tcp -j TPROXY --on-port 7895 --tproxy-mark 1/1',
+    '-A VPN_PROXY_DEVICE_POLICY_B -p udp -j TPROXY --on-port 7895 --tproxy-mark 1/1',
+    'COMMIT',
+    '',
+  ].join('\n'));
+  assert.match(fingerprintDirectDevices([directDevice]), /^[a-f0-9]{64}$/);
+});
+
+test('device policy swaps one prepared slot and preserves the active slot on failure', async () => {
+  const commands = [];
+  let failSwap = false;
+  const generations = ['epoch-a', 'rules-b', 'rules-a'];
+  const service = createDevicePolicyService({
+    chain: 'VPN_PROXY_DEVICE_POLICY',
+    tproxyPort: 7895,
+    tproxyMark: '1',
+    nextGeneration: () => generations.shift(),
+    now: () => new Date('2026-08-07T12:00:00.000Z'),
+    run: (command, args, options) => {
+      commands.push({ command, args, options });
+      if (failSwap && args.includes('-R')) return { status: 1, stderr: 'swap failed' };
+      return { status: 0, stdout: '', stderr: '' };
+    },
+  });
+
+  const initial = service.snapshot();
+  assert.equal((await service.apply([])).generation, initial.generation);
+  assert.equal(commands.length, 0);
+
+  const applied = await service.apply([directDevice]);
+  assert.deepEqual(applied.appliedIds, [directDevice.id]);
+  assert.equal(applied.changed, true);
+  assert.deepEqual(commands.at(-1).args, [
+    '-w', '1', '-t', 'mangle', '-R', 'VPN_PROXY_DEVICE_POLICY', '1', '-j', 'VPN_PROXY_DEVICE_POLICY_B',
+  ]);
+  assert.equal(commands[0].command, 'iptables-restore');
+  assert.match(commands[0].options.input, /--mac-source 00:11:22:33:44:55/);
+  assert.equal(commands.length, 2);
+  assert.equal(commands.every(({ options }) => options.timeout === 2_000), true);
+
+  const commandCount = commands.length;
+  assert.equal((await service.apply([directDevice])).changed, false);
+  assert.equal(commands.length, commandCount);
+
+  failSwap = true;
+  await assert.rejects(service.apply([]), /swap failed/);
+  assert.deepEqual(service.snapshot().appliedIds, [directDevice.id]);
+  assert.equal(service.snapshot().generation, applied.generation);
+
+  const prepareCommands = [];
+  const prepareFailure = createDevicePolicyService({
+    chain: 'VPN_PROXY_DEVICE_POLICY',
+    tproxyPort: 7895,
+    tproxyMark: '1',
+    nextGeneration: () => 'prepare-epoch',
+    run: (command, args, options) => {
+      prepareCommands.push({ command, args, options });
+      return command === 'iptables-restore'
+        ? { status: 1, stderr: 'prepare failed' }
+        : { status: 0, stdout: '', stderr: '' };
+    },
+  });
+  await assert.rejects(prepareFailure.apply([directDevice]), /prepare failed/);
+  assert.equal(prepareCommands.some(({ args }) => args.includes('-R')), false);
+  assert.deepEqual(prepareFailure.snapshot().appliedIds, []);
+});
+
+test('the maximum policy set stays inside the control request deadline', async () => {
+  const devices = Array.from({ length: 512 }, (_, index) => {
+    const mac = index.toString(16).padStart(12, '0').match(/../g).join(':');
+    return {
+      id: `dev_${index.toString(16).padStart(16, '0')}`,
+      ip: `192.168.${50 + Math.floor(index / 254)}.${(index % 254) + 1}`,
+      mac,
+      interface: 'eth0',
+    };
+  });
+  const calls = [];
+  const service = createDevicePolicyService({
+    chain: 'VPN_PROXY_DEVICE_POLICY',
+    tproxyPort: 7895,
+    tproxyMark: '1',
+    nextGeneration: () => 'generation',
+    run: (command, args, options) => {
+      calls.push({ command, args, options });
+      return { status: 0, stdout: '', stderr: '' };
+    },
+  });
+
+  await service.apply(devices);
+  assert.deepEqual(calls.map(({ command }) => command), ['iptables-restore', 'iptables']);
+  assert.equal(calls.reduce((sum, { options }) => sum + options.timeout, 0), 4_000);
+});

test/server/entrypoint-tproxy.test.js

@@ -18,9 +18,12 @@ test('gateway keeps direct forwarding active while TProxy interception is switch
   assert.doesNotMatch(entrypoint, /TPROXY_BYPASS_SOURCE_CIDRS|DIRECT_BYPASS_CACHE|ipset/);
   assert.match(entrypoint, /-t raw -I PREROUTING 1 -j "\$TRAFFIC_UPLOAD_CHAIN"/);
   assert.match(entrypoint, /-t mangle -I POSTROUTING 1 -j "\$TRAFFIC_DOWNLOAD_CHAIN"/);
+  assert.match(entrypoint, /-A "\$TPROXY_CHAIN" -j "\$DEVICE_POLICY_CHAIN"/);
+  assert.match(entrypoint, /-A "\$\{DEVICE_POLICY_CHAIN\}_A" -p tcp -j TPROXY/);
+  assert.match(entrypoint, /-A "\$\{DEVICE_POLICY_CHAIN\}_A" -p udp -j TPROXY/);
   assert.match(entrypoint, /setup_tproxy\s+if ! setup_device_traffic/);
   assert.match(entrypoint, /if ! setup_device_traffic; then[\s\S]*VPN routing remains active/);
-  assert.match(entrypoint, /export TRAFFIC_UPLOAD_CHAIN TRAFFIC_DOWNLOAD_CHAIN BYPASS_CIDRS/);
+  assert.match(entrypoint, /export TPROXY_PORT TPROXY_MARK DEVICE_POLICY_CHAIN TRAFFIC_UPLOAD_CHAIN TRAFFIC_DOWNLOAD_CHAIN BYPASS_CIDRS/);
   assert.match(entrypoint, /ipt_traffic\(\) \{\s+iptables -w 1/);
 });