873 lines
36 KiB
JavaScript
873 lines
36 KiB
JavaScript
import crypto from 'node:crypto';
|
|
import fs from 'node:fs';
|
|
import net from 'node:net';
|
|
import { HarborError } from '../../shared/errors.js';
|
|
import { isDeviceInterface } from '../adapters/neighbors.js';
|
|
import { fingerprintDirectDevices } from './devicePolicyService.js';
|
|
|
|
export const DEVICE_INVENTORY_SCHEMA_VERSION = 2;
|
|
const ONLINE_MS = 2 * 60 * 1000;
|
|
const RECENT_MS = 24 * 60 * 60 * 1000;
|
|
const RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
|
|
const TRAFFIC_HISTORY_LIMIT = 120;
|
|
const COUNTER_PATTERN = /^\d+$/;
|
|
const DEVICE_ID_PATTERN = /^dev_[a-f0-9]{16}$/;
|
|
const MAC_PATTERN = /^[0-9a-f]{2}(?::[0-9a-f]{2}){5}$/;
|
|
const FINGERPRINT_PATTERN = /^[a-f0-9]{64}$/;
|
|
const POLICY_MODES = new Set(['vpn', 'direct']);
|
|
const POLICY_STATUSES = new Set(['applied', 'applying', 'pending', 'failed']);
|
|
const PROXY_RECOVERY_ERROR = 'Повреждённый proxy traffic checkpoint восстановлен из корректных данных';
|
|
|
|
const DEFAULT_DEVICE_POLICY = Object.freeze({
|
|
desired: 'vpn',
|
|
applied: 'vpn',
|
|
status: 'applied',
|
|
appliedAt: null,
|
|
error: null,
|
|
operationId: null,
|
|
});
|
|
|
|
const DEFAULT_POLICY_STATE = {
|
|
schemaVersion: 1,
|
|
defaultMode: 'vpn',
|
|
dataplaneEpoch: null,
|
|
generation: null,
|
|
fingerprint: null,
|
|
lastAppliedAt: null,
|
|
lastError: null,
|
|
byMac: {},
|
|
};
|
|
|
|
const DEFAULT_PROXY_TRAFFIC = {
|
|
schemaVersion: 1,
|
|
lastObservedAt: null,
|
|
lastError: null,
|
|
baselinesByMac: {},
|
|
totalsByMac: {},
|
|
rebaselineMacs: [],
|
|
};
|
|
|
|
const DEFAULT_STATE = {
|
|
schemaVersion: DEVICE_INVENTORY_SCHEMA_VERSION,
|
|
revision: 0,
|
|
lastObservedAt: null,
|
|
lastError: null,
|
|
policy: DEFAULT_POLICY_STATE,
|
|
traffic: {
|
|
epoch: null,
|
|
generation: null,
|
|
lastObservedAt: null,
|
|
lastError: null,
|
|
baselinesByMac: {},
|
|
totalsByMac: {},
|
|
rebaselineMacs: [],
|
|
proxy: DEFAULT_PROXY_TRAFFIC,
|
|
},
|
|
devices: [],
|
|
};
|
|
|
|
const normalizeMac = (value) => String(value || '').trim().toLowerCase();
|
|
const deviceId = (mac) => `dev_${crypto.createHash('sha256').update(mac).digest('hex').slice(0, 16)}`;
|
|
const isPrivateMac = (mac) => (Number.parseInt(mac.slice(0, 2), 16) & 2) !== 0;
|
|
const recordEntries = (value) => value && typeof value === 'object' && !Array.isArray(value)
|
|
? Object.entries(value)
|
|
: [];
|
|
const parseStoredCounter = (value) => {
|
|
const counter = String(value ?? '');
|
|
return COUNTER_PATTERN.test(counter) ? BigInt(counter).toString() : null;
|
|
};
|
|
|
|
function normalizeProxyTraffic(value, devices) {
|
|
const proxy = value && typeof value === 'object' && !Array.isArray(value) ? value : {};
|
|
if (Number.isSafeInteger(proxy.schemaVersion) && proxy.schemaVersion > 1) {
|
|
throw new Error(`Unsupported proxy traffic schemaVersion: ${proxy.schemaVersion}`);
|
|
}
|
|
const rebaselineMacs = new Set((Array.isArray(proxy.rebaselineMacs) ? proxy.rebaselineMacs : [])
|
|
.map(normalizeMac).filter((mac) => MAC_PATTERN.test(mac)));
|
|
let recovered = value !== undefined && (
|
|
proxy !== value || proxy.schemaVersion !== 1
|
|
|| !proxy.baselinesByMac || typeof proxy.baselinesByMac !== 'object' || Array.isArray(proxy.baselinesByMac)
|
|
|| !proxy.totalsByMac || typeof proxy.totalsByMac !== 'object' || Array.isArray(proxy.totalsByMac)
|
|
);
|
|
if (recovered) {
|
|
for (const device of devices) rebaselineMacs.add(normalizeMac(device.mac));
|
|
}
|
|
const baselinesByMac = {};
|
|
for (const [rawMac, baseline] of recordEntries(proxy.baselinesByMac)) {
|
|
const mac = normalizeMac(rawMac);
|
|
const uploadBytes = parseStoredCounter(baseline?.uploadBytes);
|
|
const downloadBytes = parseStoredCounter(baseline?.downloadBytes);
|
|
if (!MAC_PATTERN.test(mac) || typeof baseline?.epoch !== 'string' || !baseline.epoch
|
|
|| uploadBytes == null || downloadBytes == null) {
|
|
if (MAC_PATTERN.test(mac)) rebaselineMacs.add(mac);
|
|
recovered = true;
|
|
continue;
|
|
}
|
|
baselinesByMac[mac] = { epoch: baseline.epoch, uploadBytes, downloadBytes };
|
|
}
|
|
const totalsByMac = {};
|
|
for (const [rawMac, total] of recordEntries(proxy.totalsByMac)) {
|
|
const mac = normalizeMac(rawMac);
|
|
const uploadBytes = parseStoredCounter(total?.uploadBytes);
|
|
const downloadBytes = parseStoredCounter(total?.downloadBytes);
|
|
if (!MAC_PATTERN.test(mac) || uploadBytes == null || downloadBytes == null) {
|
|
if (MAC_PATTERN.test(mac)) rebaselineMacs.add(mac);
|
|
recovered = true;
|
|
continue;
|
|
}
|
|
totalsByMac[mac] = {
|
|
uploadBytes,
|
|
downloadBytes,
|
|
observedAt: typeof total?.observedAt === 'string' ? total.observedAt : null,
|
|
};
|
|
}
|
|
for (const mac of new Set([...Object.keys(baselinesByMac), ...Object.keys(totalsByMac)])) {
|
|
if (!Object.hasOwn(baselinesByMac, mac) || !Object.hasOwn(totalsByMac, mac)) {
|
|
rebaselineMacs.add(mac);
|
|
recovered = true;
|
|
}
|
|
}
|
|
return {
|
|
schemaVersion: 1,
|
|
lastObservedAt: typeof proxy.lastObservedAt === 'string' ? proxy.lastObservedAt : null,
|
|
lastError: recovered
|
|
? PROXY_RECOVERY_ERROR
|
|
: typeof proxy.lastError === 'string' ? proxy.lastError : null,
|
|
baselinesByMac,
|
|
totalsByMac,
|
|
rebaselineMacs: [...rebaselineMacs],
|
|
};
|
|
}
|
|
|
|
function normalizePolicyState(value) {
|
|
const policy = value && typeof value === 'object' && !Array.isArray(value) ? value : {};
|
|
const byMac = {};
|
|
let recovered = value !== undefined && (
|
|
policy.schemaVersion !== 1
|
|
|| policy.defaultMode !== 'vpn'
|
|
|| !policy.byMac || typeof policy.byMac !== 'object' || Array.isArray(policy.byMac)
|
|
|| (policy.dataplaneEpoch != null && typeof policy.dataplaneEpoch !== 'string')
|
|
|| (policy.generation != null && typeof policy.generation !== 'string')
|
|
|| (policy.fingerprint != null && typeof policy.fingerprint !== 'string')
|
|
|| (policy.lastAppliedAt != null && typeof policy.lastAppliedAt !== 'string')
|
|
);
|
|
for (const [rawMac, entry] of recordEntries(policy.byMac)) {
|
|
const mac = normalizeMac(rawMac);
|
|
if (!MAC_PATTERN.test(mac) || !POLICY_MODES.has(entry?.desired)
|
|
|| !POLICY_MODES.has(entry?.applied) || !POLICY_STATUSES.has(entry?.status)) {
|
|
recovered = true;
|
|
continue;
|
|
}
|
|
byMac[mac] = {
|
|
desired: entry.desired,
|
|
applied: entry.applied,
|
|
status: entry.status,
|
|
appliedAt: typeof entry.appliedAt === 'string' ? entry.appliedAt : null,
|
|
error: typeof entry.error === 'string' ? entry.error : null,
|
|
operationId: typeof entry.operationId === 'string' ? entry.operationId : null,
|
|
};
|
|
}
|
|
return {
|
|
...DEFAULT_POLICY_STATE,
|
|
dataplaneEpoch: typeof policy.dataplaneEpoch === 'string' ? policy.dataplaneEpoch : null,
|
|
generation: typeof policy.generation === 'string' ? policy.generation : null,
|
|
fingerprint: typeof policy.fingerprint === 'string' ? policy.fingerprint : null,
|
|
lastAppliedAt: typeof policy.lastAppliedAt === 'string' ? policy.lastAppliedAt : null,
|
|
schemaVersion: 1,
|
|
defaultMode: 'vpn',
|
|
lastError: recovered
|
|
? 'Повреждённый device policy checkpoint восстановлен из корректных данных'
|
|
: typeof policy.lastError === 'string' ? policy.lastError : null,
|
|
byMac,
|
|
};
|
|
}
|
|
|
|
export function parseOuiVendors(text) {
|
|
const vendors = new Map();
|
|
for (const line of String(text || '').split(/\r?\n/)) {
|
|
const match = line.match(/^([0-9a-f]{2}(?:-[0-9a-f]{2}){2})\s+\(hex\)\s+(.+)$/i);
|
|
if (match) vendors.set(match[1].replaceAll('-', '').toLowerCase(), match[2].trim());
|
|
}
|
|
return vendors;
|
|
}
|
|
|
|
export function createVendorLookup(filePath = '/usr/share/ieee-data/oui.txt') {
|
|
let vendors;
|
|
return (mac) => {
|
|
if (!mac || isPrivateMac(mac)) return null;
|
|
if (!vendors) {
|
|
try {
|
|
vendors = parseOuiVendors(fs.readFileSync(filePath, 'utf8'));
|
|
} catch {
|
|
vendors = new Map();
|
|
}
|
|
}
|
|
return vendors.get(mac.replaceAll(':', '').slice(0, 6)) || null;
|
|
};
|
|
}
|
|
|
|
export function migrateDeviceInventoryState(value) {
|
|
const state = value && typeof value === 'object' && !Array.isArray(value) ? value : {};
|
|
const version = Number.isSafeInteger(state.schemaVersion) ? state.schemaVersion : 0;
|
|
if (version < 0 || version > DEVICE_INVENTORY_SCHEMA_VERSION) {
|
|
throw new Error(`Unsupported device inventory schemaVersion: ${version}`);
|
|
}
|
|
const traffic = state.traffic && typeof state.traffic === 'object' && !Array.isArray(state.traffic)
|
|
? state.traffic
|
|
: {};
|
|
const devices = Array.isArray(state.devices)
|
|
? state.devices.filter((device) => isDeviceInterface(device?.interface))
|
|
: [];
|
|
const proxyTraffic = normalizeProxyTraffic(traffic.proxy, devices);
|
|
const rebaselineMacs = new Set((Array.isArray(traffic.rebaselineMacs) ? traffic.rebaselineMacs : [])
|
|
.map(normalizeMac).filter((mac) => MAC_PATTERN.test(mac)));
|
|
let recoveredTraffic = version >= 2 && (
|
|
traffic !== state.traffic
|
|
|| !traffic.baselinesByMac || typeof traffic.baselinesByMac !== 'object' || Array.isArray(traffic.baselinesByMac)
|
|
|| !traffic.totalsByMac || typeof traffic.totalsByMac !== 'object' || Array.isArray(traffic.totalsByMac)
|
|
);
|
|
if (recoveredTraffic) {
|
|
for (const device of devices) rebaselineMacs.add(normalizeMac(device.mac));
|
|
}
|
|
const baselinesByMac = {};
|
|
for (const [rawMac, baseline] of recordEntries(traffic.baselinesByMac)) {
|
|
const mac = normalizeMac(rawMac);
|
|
const uploadBytes = parseStoredCounter(baseline?.uploadBytes);
|
|
const downloadBytes = parseStoredCounter(baseline?.downloadBytes);
|
|
if (!MAC_PATTERN.test(mac) || typeof baseline?.epoch !== 'string' || !baseline.epoch
|
|
|| uploadBytes == null || downloadBytes == null) {
|
|
if (MAC_PATTERN.test(mac)) rebaselineMacs.add(mac);
|
|
recoveredTraffic = true;
|
|
continue;
|
|
}
|
|
baselinesByMac[mac] = { epoch: baseline.epoch, uploadBytes, downloadBytes };
|
|
}
|
|
const totalsByMac = {};
|
|
for (const [rawMac, total] of recordEntries(traffic.totalsByMac)) {
|
|
const mac = normalizeMac(rawMac);
|
|
const uploadBytes = parseStoredCounter(total?.uploadBytes);
|
|
const downloadBytes = parseStoredCounter(total?.downloadBytes);
|
|
if (!MAC_PATTERN.test(mac) || uploadBytes == null || downloadBytes == null) {
|
|
if (MAC_PATTERN.test(mac)) rebaselineMacs.add(mac);
|
|
recoveredTraffic = true;
|
|
continue;
|
|
}
|
|
totalsByMac[mac] = {
|
|
uploadBytes,
|
|
downloadBytes,
|
|
observedAt: typeof total?.observedAt === 'string' ? total.observedAt : null,
|
|
};
|
|
}
|
|
for (const mac of new Set([...Object.keys(baselinesByMac), ...Object.keys(totalsByMac)])) {
|
|
if (!Object.hasOwn(baselinesByMac, mac) || !Object.hasOwn(totalsByMac, mac)) {
|
|
rebaselineMacs.add(mac);
|
|
recoveredTraffic = true;
|
|
}
|
|
}
|
|
return {
|
|
...DEFAULT_STATE,
|
|
...state,
|
|
schemaVersion: DEVICE_INVENTORY_SCHEMA_VERSION,
|
|
revision: Number.isSafeInteger(state.revision) ? state.revision : 0,
|
|
policy: normalizePolicyState(state.policy),
|
|
traffic: {
|
|
...DEFAULT_STATE.traffic,
|
|
...traffic,
|
|
lastError: recoveredTraffic
|
|
? 'Повреждённый traffic checkpoint восстановлен из корректных данных'
|
|
: traffic.lastError || null,
|
|
baselinesByMac,
|
|
totalsByMac,
|
|
rebaselineMacs: [...rebaselineMacs].filter((mac) => MAC_PATTERN.test(mac)),
|
|
proxy: proxyTraffic,
|
|
},
|
|
devices,
|
|
};
|
|
}
|
|
|
|
function deviceStatus(lastSeenAt, now) {
|
|
const age = now.getTime() - new Date(lastSeenAt).getTime();
|
|
if (age <= ONLINE_MS) return 'online';
|
|
if (age <= RECENT_MS) return 'recent';
|
|
return 'offline';
|
|
}
|
|
|
|
export function createDeviceInventoryService({
|
|
store,
|
|
observe,
|
|
observeTraffic = null,
|
|
observePolicy = null,
|
|
applyPolicies = null,
|
|
vendor = () => null,
|
|
now = () => new Date(),
|
|
}) {
|
|
let refreshPromise = null;
|
|
let policyQueue = Promise.resolve();
|
|
const trafficHistoryByMac = new Map();
|
|
const trafficCursorByMac = new Map();
|
|
|
|
function captureTrafficHistory(state) {
|
|
const knownMacs = new Set(state.devices.map(({ mac }) => mac));
|
|
for (const device of state.devices) {
|
|
const traffic = state.traffic.totalsByMac[device.mac];
|
|
const proxy = state.traffic.proxy.totalsByMac[device.mac];
|
|
const signature = `${traffic?.observedAt || ''}|${proxy?.observedAt || ''}`;
|
|
if (signature === '|') continue;
|
|
const gateway = BigInt(traffic?.uploadBytes || '0') + BigInt(traffic?.downloadBytes || '0');
|
|
const proxyTotal = BigInt(proxy?.uploadBytes || '0') + BigInt(proxy?.downloadBytes || '0');
|
|
const previous = trafficCursorByMac.get(device.mac);
|
|
trafficCursorByMac.set(device.mac, { signature, gateway, proxy: proxyTotal });
|
|
if (!previous || previous.signature === signature) continue;
|
|
const observedAt = [traffic?.observedAt, proxy?.observedAt].filter(Boolean).sort().at(-1);
|
|
const samples = trafficHistoryByMac.get(device.mac) || [];
|
|
trafficHistoryByMac.set(device.mac, [...samples, {
|
|
observedAt,
|
|
gatewayBytes: gateway > previous.gateway ? (gateway - previous.gateway).toString() : '0',
|
|
proxyBytes: proxyTotal > previous.proxy ? (proxyTotal - previous.proxy).toString() : '0',
|
|
}].slice(-TRAFFIC_HISTORY_LIMIT));
|
|
}
|
|
for (const mac of trafficCursorByMac.keys()) {
|
|
if (!knownMacs.has(mac)) {
|
|
trafficCursorByMac.delete(mac);
|
|
trafficHistoryByMac.delete(mac);
|
|
}
|
|
}
|
|
}
|
|
|
|
function serializePolicy(action) {
|
|
const result = policyQueue.then(action, action);
|
|
policyQueue = result.catch(() => {});
|
|
return result;
|
|
}
|
|
|
|
function policyFor(state, mac) {
|
|
return state.policy.byMac[mac] || DEFAULT_DEVICE_POLICY;
|
|
}
|
|
|
|
function policyIdentity(device) {
|
|
return Boolean(device) && device.confidence !== 'ambiguous'
|
|
&& net.isIPv4(String(device.ip || '')) && MAC_PATTERN.test(device.mac)
|
|
&& isDeviceInterface(device.interface);
|
|
}
|
|
|
|
function directDevices(state) {
|
|
return state.devices
|
|
.filter((device) => policyFor(state, device.mac).desired === 'direct' && policyIdentity(device))
|
|
.map(({ id, ip, mac, interface: deviceInterface }) => ({
|
|
id,
|
|
ip,
|
|
mac,
|
|
interface: deviceInterface,
|
|
}));
|
|
}
|
|
|
|
function validatePolicyAck(result, requested) {
|
|
const appliedIds = Array.isArray(result?.appliedIds) ? result.appliedIds : [];
|
|
const expectedIds = new Set(requested.map(({ id }) => id));
|
|
if (typeof result?.epoch !== 'string' || !result.epoch
|
|
|| typeof result.generation !== 'string' || !result.generation
|
|
|| !FINGERPRINT_PATTERN.test(result.fingerprint)
|
|
|| typeof result.observedAt !== 'string' || !result.observedAt
|
|
|| result.fingerprint !== fingerprintDirectDevices(requested)
|
|
|| appliedIds.length !== expectedIds.size
|
|
|| new Set(appliedIds).size !== appliedIds.length
|
|
|| appliedIds.some((id) => !DEVICE_ID_PATTERN.test(id) || !expectedIds.has(id))) {
|
|
throw new Error('Dataplane вернул невалидный device policy acknowledgement');
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function snapshot() {
|
|
const state = migrateDeviceInventoryState(store.read());
|
|
const current = now();
|
|
const rank = { online: 0, recent: 1, offline: 2 };
|
|
const devices = state.devices.map((device) => {
|
|
const traffic = state.traffic.totalsByMac[device.mac];
|
|
const proxyTraffic = state.traffic.proxy.totalsByMac[device.mac];
|
|
const policy = policyFor(state, device.mac);
|
|
return {
|
|
...device,
|
|
status: deviceStatus(device.lastSeenAt, current),
|
|
uploadBytes: traffic?.uploadBytes || '0',
|
|
downloadBytes: traffic?.downloadBytes || '0',
|
|
trafficObservedAt: traffic?.observedAt || null,
|
|
proxyUploadBytes: proxyTraffic?.uploadBytes || '0',
|
|
proxyDownloadBytes: proxyTraffic?.downloadBytes || '0',
|
|
proxyTrafficObservedAt: proxyTraffic?.observedAt || null,
|
|
trafficHistory: trafficHistoryByMac.get(device.mac) || [],
|
|
desiredPolicy: policy.desired,
|
|
appliedPolicy: policy.applied,
|
|
policyStatus: policy.status,
|
|
policyAppliedAt: policy.appliedAt,
|
|
policyError: policy.error,
|
|
};
|
|
}).sort((left, right) => (
|
|
Number(right.pinned) - Number(left.pinned)
|
|
|| rank[left.status] - rank[right.status]
|
|
|| String(right.lastSeenAt).localeCompare(String(left.lastSeenAt))
|
|
));
|
|
return {
|
|
revision: state.revision,
|
|
trafficHistoryCapacity: TRAFFIC_HISTORY_LIMIT,
|
|
source: {
|
|
kind: 'neighbor',
|
|
lastObservedAt: state.lastObservedAt,
|
|
error: state.lastError,
|
|
traffic: {
|
|
lastObservedAt: state.traffic.lastObservedAt,
|
|
error: state.traffic.lastError,
|
|
proxy: {
|
|
lastObservedAt: state.traffic.proxy.lastObservedAt,
|
|
error: state.traffic.proxy.lastError,
|
|
},
|
|
},
|
|
policy: {
|
|
lastAppliedAt: state.policy.lastAppliedAt,
|
|
error: state.policy.lastError,
|
|
},
|
|
},
|
|
devices,
|
|
};
|
|
}
|
|
|
|
function markPolicyEpoch(observed) {
|
|
if (typeof observed?.epoch !== 'string' || !observed.epoch || !Array.isArray(observed.appliedIds)) return;
|
|
store.update((stored) => {
|
|
const state = migrateDeviceInventoryState(stored);
|
|
if (!state.policy.dataplaneEpoch || state.policy.dataplaneEpoch === observed.epoch) return state;
|
|
const appliedIds = new Set(observed.appliedIds);
|
|
const devicesByMac = new Map(state.devices.map((device) => [device.mac, device]));
|
|
const byMac = {};
|
|
for (const [mac, entry] of Object.entries(state.policy.byMac)) {
|
|
const device = devicesByMac.get(mac);
|
|
if (!device) continue;
|
|
const applied = appliedIds.has(device.id) ? 'direct' : 'vpn';
|
|
if (entry.desired === 'vpn' && applied === 'vpn') continue;
|
|
byMac[mac] = {
|
|
...entry,
|
|
applied,
|
|
status: entry.desired === applied ? 'applied' : 'pending',
|
|
appliedAt: observed.observedAt || entry.appliedAt,
|
|
error: entry.desired === applied
|
|
? null
|
|
: 'Dataplane перезапущен, маршрут ожидает повторного применения',
|
|
operationId: null,
|
|
};
|
|
}
|
|
return {
|
|
...state,
|
|
revision: state.revision + 1,
|
|
policy: {
|
|
...state.policy,
|
|
dataplaneEpoch: observed.epoch,
|
|
generation: typeof observed.generation === 'string' ? observed.generation : null,
|
|
fingerprint: FINGERPRINT_PATTERN.test(observed.fingerprint) ? observed.fingerprint : null,
|
|
lastAppliedAt: observed.observedAt || state.policy.lastAppliedAt,
|
|
lastError: null,
|
|
byMac,
|
|
},
|
|
};
|
|
});
|
|
}
|
|
|
|
function commitPolicySuccess(result) {
|
|
store.update((stored) => {
|
|
const state = migrateDeviceInventoryState(stored);
|
|
const appliedIds = new Set(Array.isArray(result.appliedIds) ? result.appliedIds : []);
|
|
const devicesByMac = new Map(state.devices.map((device) => [device.mac, device]));
|
|
const byMac = {};
|
|
for (const [mac, entry] of Object.entries(state.policy.byMac)) {
|
|
const device = devicesByMac.get(mac);
|
|
if (!device) continue;
|
|
if (entry.desired === 'vpn' && !appliedIds.has(device?.id)) continue;
|
|
const applied = appliedIds.has(device?.id) ? 'direct' : 'vpn';
|
|
byMac[mac] = {
|
|
...entry,
|
|
applied,
|
|
status: entry.desired === applied ? 'applied' : 'pending',
|
|
appliedAt: result.observedAt || state.policy.lastAppliedAt,
|
|
error: entry.desired === applied
|
|
? null
|
|
: 'Gateway должен однозначно распознать устройство',
|
|
operationId: null,
|
|
};
|
|
}
|
|
const policy = {
|
|
...state.policy,
|
|
dataplaneEpoch: result.epoch,
|
|
generation: result.generation,
|
|
fingerprint: result.fingerprint,
|
|
lastAppliedAt: result.observedAt || state.policy.lastAppliedAt,
|
|
lastError: null,
|
|
byMac,
|
|
};
|
|
if (JSON.stringify(policy) === JSON.stringify(state.policy)) return state;
|
|
return { ...state, revision: state.revision + 1, policy };
|
|
});
|
|
}
|
|
|
|
function commitPolicyFailure(error) {
|
|
store.update((stored) => {
|
|
const state = migrateDeviceInventoryState(stored);
|
|
const message = error.message || String(error);
|
|
const byMac = Object.fromEntries(Object.entries(state.policy.byMac).map(([mac, entry]) => [mac, {
|
|
...entry,
|
|
status: entry.status === 'applied' && entry.desired === entry.applied ? 'applied' : 'failed',
|
|
error: entry.status === 'applied' && entry.desired === entry.applied ? null : message,
|
|
operationId: null,
|
|
}]));
|
|
return {
|
|
...state,
|
|
revision: state.revision + 1,
|
|
policy: { ...state.policy, lastError: message, byMac },
|
|
};
|
|
});
|
|
}
|
|
|
|
async function reconcileLocked(observedPolicy, throwOnError) {
|
|
if (!applyPolicies) return snapshot();
|
|
markPolicyEpoch(observedPolicy);
|
|
const state = migrateDeviceInventoryState(store.read());
|
|
try {
|
|
const requested = directDevices(state);
|
|
const result = validatePolicyAck(await applyPolicies(requested), requested);
|
|
commitPolicySuccess(result);
|
|
return snapshot();
|
|
} catch (cause) {
|
|
commitPolicyFailure(cause);
|
|
if (throwOnError) throw new HarborError('DEVICE_POLICY_APPLY_FAILED', { cause });
|
|
return snapshot();
|
|
}
|
|
}
|
|
|
|
async function performRefresh() {
|
|
const [result, trafficResult, policyResult] = await Promise.all([
|
|
Promise.resolve().then(() => observe()).catch((error) => ({
|
|
observedAt: now().toISOString(),
|
|
observations: [],
|
|
error: error.message || String(error),
|
|
})),
|
|
observeTraffic
|
|
? Promise.resolve().then(() => observeTraffic())
|
|
.catch((error) => ({ transportError: error.message || String(error) }))
|
|
: null,
|
|
observePolicy
|
|
? Promise.resolve().then(() => observePolicy())
|
|
.catch((error) => ({ transportError: error.message || String(error) }))
|
|
: null,
|
|
]);
|
|
const observedAt = result?.observedAt || now().toISOString();
|
|
const observations = (Array.isArray(result?.observations) ? result.observations : [])
|
|
.filter((observation) => isDeviceInterface(observation?.interface));
|
|
const identitiesByMac = new Map();
|
|
for (const observation of observations) {
|
|
const mac = normalizeMac(observation.mac);
|
|
if (!mac || !net.isIPv4(String(observation.ip || ''))) continue;
|
|
if (!identitiesByMac.has(mac)) identitiesByMac.set(mac, new Set());
|
|
identitiesByMac.get(mac).add(`${observation.ip}|${observation.interface || ''}`);
|
|
}
|
|
return serializePolicy(async () => {
|
|
const nextState = store.update((stored) => {
|
|
const state = migrateDeviceInventoryState(stored);
|
|
const byMac = new Map(state.devices.map((device) => [device.mac, device]));
|
|
for (const observation of observations) {
|
|
const mac = normalizeMac(observation.mac);
|
|
if (!mac) continue;
|
|
const previous = byMac.get(mac);
|
|
const lastSeenAt = observation.active || !previous
|
|
? observation.observedAt || observedAt
|
|
: previous.lastSeenAt;
|
|
byMac.set(mac, {
|
|
id: previous?.id || deviceId(mac),
|
|
alias: previous?.alias || '',
|
|
pinned: previous?.pinned === true,
|
|
hostname: previous?.hostname || null,
|
|
manufacturer: previous?.manufacturer || vendor(mac),
|
|
mac,
|
|
ip: String(observation.ip || previous?.ip || ''),
|
|
interface: String(observation.interface || previous?.interface || ''),
|
|
firstSeenAt: previous?.firstSeenAt || observation.observedAt || observedAt,
|
|
lastSeenAt,
|
|
source: 'neighbor',
|
|
confidence: identitiesByMac.get(mac)?.size > 1
|
|
? 'ambiguous'
|
|
: isPrivateMac(mac) ? 'medium' : 'high',
|
|
});
|
|
}
|
|
const cutoff = new Date(observedAt).getTime() - RETENTION_MS;
|
|
const devices = [...byMac.values()].filter((device) => (
|
|
device.pinned || device.alias || new Date(device.lastSeenAt).getTime() >= cutoff
|
|
));
|
|
let traffic = state.traffic;
|
|
if (trafficResult) {
|
|
if (trafficResult.transportError) {
|
|
traffic = { ...traffic, lastError: trafficResult.transportError };
|
|
} else {
|
|
try {
|
|
if (typeof trafficResult.epoch !== 'string' || !trafficResult.epoch) {
|
|
throw new Error('Dataplane не вернул traffic epoch');
|
|
}
|
|
const rows = Array.isArray(trafficResult.devices) ? trafficResult.devices : [];
|
|
const processByMac = new Map();
|
|
const proxyByMac = new Map();
|
|
let proxyRows = 0;
|
|
let legacyRows = 0;
|
|
let proxySampleError = null;
|
|
for (const row of rows) {
|
|
const mac = normalizeMac(row?.mac);
|
|
const upload = String(row?.uploadBytes ?? '');
|
|
const download = String(row?.downloadBytes ?? '');
|
|
if (!MAC_PATTERN.test(mac) || !COUNTER_PATTERN.test(upload) || !COUNTER_PATTERN.test(download)) {
|
|
throw new Error('Dataplane вернул невалидный traffic counter');
|
|
}
|
|
const previous = processByMac.get(mac) || { upload: 0n, download: 0n };
|
|
processByMac.set(mac, {
|
|
upload: previous.upload + BigInt(upload),
|
|
download: previous.download + BigInt(download),
|
|
});
|
|
const hasProxyUpload = Object.hasOwn(row || {}, 'proxyUploadBytes');
|
|
const hasProxyDownload = Object.hasOwn(row || {}, 'proxyDownloadBytes');
|
|
if (!hasProxyUpload && !hasProxyDownload) {
|
|
legacyRows += 1;
|
|
continue;
|
|
}
|
|
if (!hasProxyUpload || !hasProxyDownload) {
|
|
proxySampleError = 'Dataplane вернул неполный proxy traffic counter';
|
|
continue;
|
|
}
|
|
const proxyUpload = String(row.proxyUploadBytes);
|
|
const proxyDownload = String(row.proxyDownloadBytes);
|
|
if (!COUNTER_PATTERN.test(proxyUpload) || !COUNTER_PATTERN.test(proxyDownload)) {
|
|
proxySampleError = 'Dataplane вернул невалидный proxy traffic counter';
|
|
continue;
|
|
}
|
|
proxyRows += 1;
|
|
const previousProxy = proxyByMac.get(mac) || { upload: 0n, download: 0n };
|
|
proxyByMac.set(mac, {
|
|
upload: previousProxy.upload + BigInt(proxyUpload),
|
|
download: previousProxy.download + BigInt(proxyDownload),
|
|
});
|
|
}
|
|
if (proxyRows && legacyRows) proxySampleError = 'Dataplane смешал старый и новый proxy traffic contract';
|
|
const knownMacs = new Set(devices.map((device) => device.mac));
|
|
const baselinesByMac = { ...traffic.baselinesByMac };
|
|
const totalsByMac = { ...traffic.totalsByMac };
|
|
const rebaselineMacs = new Set(traffic.rebaselineMacs);
|
|
for (const [mac, processTotal] of processByMac) {
|
|
if (!knownMacs.has(mac)) continue;
|
|
const baseline = baselinesByMac[mac];
|
|
const recovering = rebaselineMacs.has(mac);
|
|
const sameEpoch = !recovering && baseline?.epoch === trafficResult.epoch;
|
|
const baselineUpload = sameEpoch ? BigInt(baseline.uploadBytes) : 0n;
|
|
const baselineDownload = sameEpoch ? BigInt(baseline.downloadBytes) : 0n;
|
|
if (processTotal.upload < baselineUpload || processTotal.download < baselineDownload) {
|
|
throw new Error('Dataplane traffic counter уменьшился внутри одного epoch');
|
|
}
|
|
const total = totalsByMac[mac] || { uploadBytes: '0', downloadBytes: '0' };
|
|
totalsByMac[mac] = {
|
|
uploadBytes: (BigInt(total.uploadBytes)
|
|
+ (recovering ? 0n : processTotal.upload - baselineUpload)).toString(),
|
|
downloadBytes: (BigInt(total.downloadBytes)
|
|
+ (recovering ? 0n : processTotal.download - baselineDownload)).toString(),
|
|
observedAt: trafficResult.observedAt || traffic.lastObservedAt,
|
|
};
|
|
baselinesByMac[mac] = {
|
|
epoch: trafficResult.epoch,
|
|
uploadBytes: processTotal.upload.toString(),
|
|
downloadBytes: processTotal.download.toString(),
|
|
};
|
|
rebaselineMacs.delete(mac);
|
|
}
|
|
for (const mac of Object.keys(totalsByMac)) {
|
|
if (!knownMacs.has(mac)) {
|
|
delete totalsByMac[mac];
|
|
delete baselinesByMac[mac];
|
|
rebaselineMacs.delete(mac);
|
|
}
|
|
}
|
|
for (const mac of rebaselineMacs) {
|
|
if (!knownMacs.has(mac)) {
|
|
delete totalsByMac[mac];
|
|
delete baselinesByMac[mac];
|
|
rebaselineMacs.delete(mac);
|
|
}
|
|
}
|
|
const proxyBaselines = { ...traffic.proxy.baselinesByMac };
|
|
const proxyTotals = { ...traffic.proxy.totalsByMac };
|
|
const proxyRebaseline = new Set(traffic.proxy.rebaselineMacs);
|
|
for (const mac of new Set([
|
|
...Object.keys(proxyTotals),
|
|
...Object.keys(proxyBaselines),
|
|
...proxyRebaseline,
|
|
])) {
|
|
if (knownMacs.has(mac)) continue;
|
|
delete proxyTotals[mac];
|
|
delete proxyBaselines[mac];
|
|
proxyRebaseline.delete(mac);
|
|
}
|
|
let proxy = {
|
|
...traffic.proxy,
|
|
lastError: !proxyRebaseline.size && traffic.proxy.lastError === PROXY_RECOVERY_ERROR
|
|
? null
|
|
: traffic.proxy.lastError,
|
|
baselinesByMac: proxyBaselines,
|
|
totalsByMac: proxyTotals,
|
|
rebaselineMacs: [...proxyRebaseline],
|
|
};
|
|
if (proxySampleError) {
|
|
proxy = { ...proxy, lastError: proxySampleError };
|
|
} else if (proxyRows) {
|
|
try {
|
|
const nextProxyBaselines = { ...proxyBaselines };
|
|
const nextProxyTotals = { ...proxyTotals };
|
|
const nextProxyRebaseline = new Set(proxyRebaseline);
|
|
for (const [mac, processTotal] of proxyByMac) {
|
|
if (!knownMacs.has(mac)) continue;
|
|
const baseline = nextProxyBaselines[mac];
|
|
const recovering = nextProxyRebaseline.has(mac);
|
|
const sameEpoch = !recovering && baseline?.epoch === trafficResult.epoch;
|
|
const baselineUpload = sameEpoch ? BigInt(baseline.uploadBytes) : 0n;
|
|
const baselineDownload = sameEpoch ? BigInt(baseline.downloadBytes) : 0n;
|
|
if (processTotal.upload < baselineUpload || processTotal.download < baselineDownload) {
|
|
throw new Error('Dataplane proxy traffic counter уменьшился внутри одного epoch');
|
|
}
|
|
const total = nextProxyTotals[mac] || { uploadBytes: '0', downloadBytes: '0' };
|
|
nextProxyTotals[mac] = {
|
|
uploadBytes: (BigInt(total.uploadBytes)
|
|
+ (recovering ? 0n : processTotal.upload - baselineUpload)).toString(),
|
|
downloadBytes: (BigInt(total.downloadBytes)
|
|
+ (recovering ? 0n : processTotal.download - baselineDownload)).toString(),
|
|
observedAt: trafficResult.observedAt || proxy.lastObservedAt,
|
|
};
|
|
nextProxyBaselines[mac] = {
|
|
epoch: trafficResult.epoch,
|
|
uploadBytes: processTotal.upload.toString(),
|
|
downloadBytes: processTotal.download.toString(),
|
|
};
|
|
nextProxyRebaseline.delete(mac);
|
|
}
|
|
proxy = {
|
|
...proxy,
|
|
lastObservedAt: trafficResult.observedAt || proxy.lastObservedAt,
|
|
lastError: trafficResult.source?.error
|
|
|| (nextProxyRebaseline.size ? proxy.lastError : null),
|
|
baselinesByMac: nextProxyBaselines,
|
|
totalsByMac: nextProxyTotals,
|
|
rebaselineMacs: [...nextProxyRebaseline],
|
|
};
|
|
} catch (error) {
|
|
proxy = { ...proxy, lastError: error.message || String(error) };
|
|
}
|
|
}
|
|
traffic = {
|
|
...traffic,
|
|
epoch: trafficResult.epoch,
|
|
generation: trafficResult.generation || traffic.generation,
|
|
lastObservedAt: trafficResult.observedAt || traffic.lastObservedAt,
|
|
lastError: trafficResult.source?.error
|
|
|| (rebaselineMacs.size ? traffic.lastError : null),
|
|
baselinesByMac,
|
|
totalsByMac,
|
|
rebaselineMacs: [...rebaselineMacs],
|
|
proxy,
|
|
};
|
|
} catch (error) {
|
|
traffic = { ...traffic, lastError: error.message || String(error) };
|
|
}
|
|
}
|
|
}
|
|
return {
|
|
...state,
|
|
revision: state.revision + 1,
|
|
lastObservedAt: observedAt,
|
|
lastError: result?.error || null,
|
|
traffic,
|
|
devices,
|
|
};
|
|
});
|
|
captureTrafficHistory(nextState);
|
|
if (policyResult?.transportError) commitPolicyFailure(new Error(policyResult.transportError));
|
|
return reconcileLocked(policyResult, false);
|
|
});
|
|
}
|
|
|
|
function refresh() {
|
|
if (!refreshPromise) {
|
|
refreshPromise = performRefresh().finally(() => {
|
|
refreshPromise = null;
|
|
});
|
|
}
|
|
return refreshPromise;
|
|
}
|
|
|
|
function update(id, patch, expectedRevision) {
|
|
if (!patch || typeof patch !== 'object' || Array.isArray(patch)) {
|
|
throw new HarborError('REQUEST_INVALID');
|
|
}
|
|
const aliasProvided = Object.hasOwn(patch, 'alias');
|
|
const pinProvided = Object.hasOwn(patch, 'pinned');
|
|
if (!Number.isSafeInteger(expectedRevision) || expectedRevision < 0
|
|
|| (!aliasProvided && !pinProvided)
|
|
|| (aliasProvided && (typeof patch.alias !== 'string' || patch.alias.length > 64))
|
|
|| (pinProvided && typeof patch.pinned !== 'boolean')) {
|
|
throw new HarborError('REQUEST_INVALID');
|
|
}
|
|
store.update((stored) => {
|
|
const state = migrateDeviceInventoryState(stored);
|
|
if (state.revision !== expectedRevision) throw new HarborError('STATE_CONFLICT');
|
|
const index = state.devices.findIndex((device) => device.id === id);
|
|
if (index < 0) throw new HarborError('DEVICE_NOT_FOUND');
|
|
const devices = [...state.devices];
|
|
devices[index] = {
|
|
...devices[index],
|
|
...(aliasProvided ? { alias: patch.alias.trim() } : {}),
|
|
...(pinProvided ? { pinned: patch.pinned } : {}),
|
|
};
|
|
return { ...state, revision: state.revision + 1, devices };
|
|
});
|
|
return snapshot();
|
|
}
|
|
|
|
function setPolicy(id, mode, expectedRevision) {
|
|
if (!Number.isSafeInteger(expectedRevision) || expectedRevision < 0 || !POLICY_MODES.has(mode)) {
|
|
throw new HarborError('REQUEST_INVALID');
|
|
}
|
|
return serializePolicy(async () => {
|
|
store.update((stored) => {
|
|
const state = migrateDeviceInventoryState(stored);
|
|
if (state.revision !== expectedRevision) throw new HarborError('STATE_CONFLICT');
|
|
const device = state.devices.find((candidate) => candidate.id === id);
|
|
if (!device) throw new HarborError('DEVICE_NOT_FOUND');
|
|
if (mode === 'direct' && !policyIdentity(device)) throw new HarborError('DEVICE_IDENTITY_AMBIGUOUS');
|
|
const current = policyFor(state, device.mac);
|
|
if (current.desired === mode && current.status === 'applied') return state;
|
|
const byMac = {
|
|
...state.policy.byMac,
|
|
[device.mac]: {
|
|
...current,
|
|
desired: mode,
|
|
status: 'applying',
|
|
error: null,
|
|
operationId: crypto.randomUUID(),
|
|
},
|
|
};
|
|
return {
|
|
...state,
|
|
revision: state.revision + 1,
|
|
policy: { ...state.policy, lastError: null, byMac },
|
|
};
|
|
});
|
|
return reconcileLocked(null, true);
|
|
});
|
|
}
|
|
|
|
async function reconcilePolicies() {
|
|
const observed = observePolicy
|
|
? await Promise.resolve().then(() => observePolicy()).catch(() => null)
|
|
: null;
|
|
return serializePolicy(() => reconcileLocked(observed, true));
|
|
}
|
|
|
|
return { snapshot, refresh, update, setPolicy, reconcilePolicies };
|
|
}
|