From e2de64ef94e8142abf23cc309cc5d6b7e92cfd84 Mon Sep 17 00:00:00 2001 From: divyanshujethi Date: Sun, 8 Nov 2026 21:38:25 +0530 Subject: [PATCH 1/2] feat(daemon): implement headless agentshieldd IPC daemon with OTel streaming and process lock sync (Phase 2) --- src/daemon.ts | 562 +++++++++++++++++++++++++++++++++++++++ src/index.ts | 1 + tests/daemon-ipc.test.ts | 154 +++++++++++ 3 files changed, 717 insertions(+) create mode 100644 src/daemon.ts create mode 100644 tests/daemon-ipc.test.ts diff --git a/src/daemon.ts b/src/daemon.ts new file mode 100644 index 0000000..3c8ee3c --- /dev/null +++ b/src/daemon.ts @@ -0,0 +1,562 @@ +import * as crypto from "node:crypto"; +import * as fs from "node:fs"; +import * as net from "node:net"; +import * as os from "node:os"; +import * as path from "node:path"; +import { AGENTSHIELD_VERSION } from "./version.js"; + +export interface DaemonConfig { + socketPath?: string; + authSecret?: string; + tokenFilePath?: string; + maxRingBufferEntries?: number; +} + +export interface OtelMetricRecord { + id: string; + timestamp: number; + metricName: string; + value: number; + unit?: string; + type: "counter" | "histogram" | "gauge" | "audit_event"; + attributes: Record; +} + +export interface DaemonLockHandle { + lockId: string; + resourceKey: string; + ownerId: string; + fenceToken: number; + acquiredAt: number; + expiresAt: number; + leaseTtlMs: number; +} + +export interface DaemonStatus { + version: string; + uptimeMs: number; + socketPath: string; + pid: number; + activeLocks: number; + totalOtelEvents: number; + connectedClients: number; +} + +/** + * Returns the standard OS-specific IPC endpoint: + * - Windows: Named pipe "\\\\.\\pipe\\agentshield" + * - Linux / macOS: Unix domain socket "/var/run/agentshield.sock" (or fallback to tmpdir) + */ +export function getDefaultSocketPath(): string { + if (process.platform === "win32") { + return "\\\\.\\pipe\\agentshield"; + } + + // Check if /var/run is accessible and writable + try { + fs.accessSync("/var/run", fs.constants.W_OK); + return "/var/run/agentshield.sock"; + } catch { + return path.join(os.tmpdir(), "agentshield.sock"); + } +} + +/** + * Headless Background Daemon (agentshieldd) + * + * Implements a lightweight local daemon communicating over an authenticated + * Unix domain socket (/var/run/agentshield.sock with permission 0600) or Windows Named Pipe + * (\\\\.\\pipe\\agentshield) to stream OpenTelemetry metrics and synchronize process locks. + */ +export class AgentShieldDaemon { + private server: net.Server | null = null; + private socketPath: string; + private authSecret: string; + private tokenFilePath: string; + private startedAt = 0; + private globalFenceCounter = 1; + private activeLocks: Map = new Map(); + private otelRingBuffer: OtelMetricRecord[] = []; + private maxRingBufferEntries: number; + private clientSessions: Map = new Map(); + + constructor(config: DaemonConfig = {}) { + this.socketPath = config.socketPath || getDefaultSocketPath(); + this.authSecret = config.authSecret || crypto.randomBytes(32).toString("hex"); + this.maxRingBufferEntries = config.maxRingBufferEntries || 2000; + + const defaultTokenDir = path.join(os.homedir(), ".agentshield"); + this.tokenFilePath = config.tokenFilePath || path.join(defaultTokenDir, "daemon.token"); + } + + public getSocketPath(): string { + return this.socketPath; + } + + public getAuthSecret(): string { + return this.authSecret; + } + + /** + * Starts the IPC daemon listening on the domain socket / named pipe. + */ + public async start(): Promise { + if (this.server) return; + + // Clean up stale socket file on POSIX systems + if (process.platform !== "win32" && fs.existsSync(this.socketPath)) { + try { + fs.unlinkSync(this.socketPath); + } catch { + // ignore if cannot unlink + } + } + + // Persist authentication token to secure user-owned directory + try { + const tokenDir = path.dirname(this.tokenFilePath); + if (!fs.existsSync(tokenDir)) { + fs.mkdirSync(tokenDir, { recursive: true, mode: 0o700 }); + } + fs.writeFileSync(this.tokenFilePath, this.authSecret, { encoding: "utf8", mode: 0o600 }); + } catch { + // non-fatal if running in restricted sandboxed test + } + + this.startedAt = Date.now(); + + await new Promise((resolve, reject) => { + this.server = net.createServer((socket) => this.handleConnection(socket)); + + this.server.on("error", (err) => { + reject(err); + }); + + this.server.listen(this.socketPath, () => { + // Enforce strict file mode 0600 on Unix domain sockets so only current UID can connect + if (process.platform !== "win32") { + try { + fs.chmodSync(this.socketPath, 0o600); + } catch { + // ignore chmod errors in testing environments + } + } + resolve(); + }); + }); + } + + /** + * Stops the daemon and cleans up IPC endpoints. + */ + public async stop(): Promise { + if (!this.server) return; + + for (const [socket] of this.clientSessions) { + try { + socket.destroy(); + } catch { + // ignore + } + } + this.clientSessions.clear(); + + await new Promise((resolve) => { + this.server?.close(() => resolve()); + }); + this.server = null; + + if (process.platform !== "win32" && fs.existsSync(this.socketPath)) { + try { + fs.unlinkSync(this.socketPath); + } catch { + // ignore + } + } + } + + public getStatus(): DaemonStatus { + this.purgeExpiredLocks(); + return { + version: AGENTSHIELD_VERSION, + uptimeMs: this.startedAt ? Date.now() - this.startedAt : 0, + socketPath: this.socketPath, + pid: process.pid, + activeLocks: this.activeLocks.size, + totalOtelEvents: this.otelRingBuffer.length, + connectedClients: this.clientSessions.size, + }; + } + + public getOtelMetrics(): OtelMetricRecord[] { + return [...this.otelRingBuffer]; + } + + private handleConnection(socket: net.Socket): void { + const sessionId = crypto.randomUUID().slice(0, 8); + this.clientSessions.set(socket, { authenticated: false, id: sessionId }); + + let buffer = ""; + + socket.on("data", (chunk) => { + buffer += chunk.toString("utf8"); + let newlineIdx = buffer.indexOf("\n"); + + while (newlineIdx !== -1) { + const rawLine = buffer.slice(0, newlineIdx).trim(); + buffer = buffer.slice(newlineIdx + 1); + + if (rawLine) { + this.processCommand(socket, rawLine); + } + + newlineIdx = buffer.indexOf("\n"); + } + }); + + socket.on("close", () => { + this.clientSessions.delete(socket); + }); + + socket.on("error", () => { + this.clientSessions.delete(socket); + }); + } + + private processCommand(socket: net.Socket, rawLine: string): void { + const session = this.clientSessions.get(socket); + if (!session) return; + + try { + const msg = JSON.parse(rawLine); + const { action, id } = msg; + + // 1. PING is allowed unauthenticated + if (action === "PING") { + this.sendResponse(socket, { + id, + ok: true, + action: "PONG", + version: AGENTSHIELD_VERSION, + authenticated: session.authenticated, + }); + return; + } + + // 2. AUTH handshake + if (action === "AUTH") { + if (msg.token && msg.token === this.authSecret) { + session.authenticated = true; + this.sendResponse(socket, { id, ok: true, message: "Authenticated successfully" }); + } else { + this.sendResponse(socket, { id, ok: false, error: "Invalid authentication token" }); + } + return; + } + + // Guard: all remaining actions require authentication + if (!session.authenticated) { + this.sendResponse(socket, { + id, + ok: false, + error: "Unauthorized: AUTH handshake required before command execution", + }); + return; + } + + // 3. LOCK_ACQUIRE + if (action === "LOCK_ACQUIRE") { + const { resourceKey, ownerId, leaseTtlMs } = msg; + const result = this.acquireLock(resourceKey, ownerId, leaseTtlMs); + this.sendResponse(socket, { id, ...result }); + return; + } + + // 4. LOCK_RELEASE + if (action === "LOCK_RELEASE") { + const { resourceKey, lockId, ownerId } = msg; + const released = this.releaseLock(resourceKey, lockId, ownerId); + this.sendResponse(socket, { id, ok: released }); + return; + } + + // 5. LOCK_STATUS + if (action === "LOCK_STATUS") { + const { resourceKey } = msg; + this.purgeExpiredLocks(); + const lock = this.activeLocks.get(resourceKey); + this.sendResponse(socket, { + id, + ok: true, + locked: Boolean(lock), + lock: lock || null, + }); + return; + } + + // 6. STREAM_OTEL + if (action === "STREAM_OTEL") { + const { metricName, value, unit, type, attributes } = msg; + const record: OtelMetricRecord = { + id: `otel_${crypto.randomUUID().slice(0, 8)}`, + timestamp: Date.now(), + metricName: String(metricName || "agentshield.execution"), + value: Number(value ?? 1), + unit: unit || "1", + type: type || "counter", + attributes: attributes || {}, + }; + + this.recordOtelMetric(record); + this.sendResponse(socket, { id, ok: true, metricId: record.id }); + return; + } + + // 7. GET_METRICS / STATUS + if (action === "GET_METRICS") { + this.sendResponse(socket, { + id, + ok: true, + status: this.getStatus(), + metrics: this.otelRingBuffer.slice(-50), + }); + return; + } + + this.sendResponse(socket, { id, ok: false, error: `Unknown daemon action: ${action}` }); + } catch (err: any) { + this.sendResponse(socket, { ok: false, error: `Malformed IPC message: ${err.message}` }); + } + } + + private sendResponse(socket: net.Socket, data: any): void { + if (socket.writable) { + socket.write(`${JSON.stringify(data)}\n`); + } + } + + private purgeExpiredLocks(): void { + const now = Date.now(); + for (const [key, lock] of this.activeLocks.entries()) { + if (lock.expiresAt <= now) { + this.activeLocks.delete(key); + } + } + } + + private acquireLock( + resourceKey: string, + ownerId?: string, + leaseTtlMs = 5000, + ): { ok: boolean; lock?: DaemonLockHandle; error?: string } { + if (!resourceKey) { + return { ok: false, error: "Missing resourceKey" }; + } + + this.purgeExpiredLocks(); + const existing = this.activeLocks.get(resourceKey); + + if (existing) { + return { + ok: false, + error: `Resource '${resourceKey}' is locked by owner '${existing.ownerId}' until ${new Date(existing.expiresAt).toISOString()}`, + }; + } + + const fenceToken = ++this.globalFenceCounter; + const now = Date.now(); + const handle: DaemonLockHandle = { + lockId: `lock_${crypto.randomUUID().slice(0, 10)}`, + resourceKey, + ownerId: ownerId || `worker_${crypto.randomUUID().slice(0, 6)}`, + fenceToken, + acquiredAt: now, + expiresAt: now + leaseTtlMs, + leaseTtlMs, + }; + + this.activeLocks.set(resourceKey, handle); + return { ok: true, lock: handle }; + } + + private releaseLock(resourceKey: string, lockId: string, ownerId?: string): boolean { + if (!resourceKey) return false; + this.purgeExpiredLocks(); + + const current = this.activeLocks.get(resourceKey); + if (!current) return true; // Already expired or released + + // Guard against releasing an hijacked lock + if (current.lockId === lockId) { + if (ownerId && current.ownerId !== ownerId) { + return false; + } + this.activeLocks.delete(resourceKey); + return true; + } + + return false; + } + + private recordOtelMetric(record: OtelMetricRecord): void { + this.otelRingBuffer.push(record); + if (this.otelRingBuffer.length > this.maxRingBufferEntries) { + this.otelRingBuffer.shift(); + } + } +} + +/** + * High-speed IPC Client for AgentShield, DBShield, and IDE Extensions + * to communicate with the headless agentshieldd daemon. + */ +export class AgentShieldDaemonClient { + private socket: net.Socket | null = null; + private socketPath: string; + private pendingRequests: Map< + string, + { resolve: (val: any) => void; reject: (err: any) => void } + > = new Map(); + private incomingBuffer = ""; + + constructor(socketPath?: string) { + this.socketPath = socketPath || getDefaultSocketPath(); + } + + public async connect(timeoutMs = 3000): Promise { + if (this.socket) return; + + return new Promise((resolve, reject) => { + const socket = net.createConnection(this.socketPath); + let connected = false; + + const timer = setTimeout(() => { + if (!connected) { + socket.destroy(); + reject(new Error(`Timed out connecting to agentshieldd socket: ${this.socketPath}`)); + } + }, timeoutMs); + + socket.on("connect", () => { + connected = true; + clearTimeout(timer); + this.socket = socket; + this.initSocketListeners(socket); + resolve(); + }); + + socket.on("error", (err) => { + clearTimeout(timer); + reject(err); + }); + }); + } + + public disconnect(): void { + if (this.socket) { + this.socket.destroy(); + this.socket = null; + } + this.pendingRequests.clear(); + } + + public async ping(): Promise { + return this.sendRequest({ action: "PING" }); + } + + public async authenticate(token: string): Promise { + const res = await this.sendRequest({ action: "AUTH", token }); + return Boolean(res.ok); + } + + public async acquireLock( + resourceKey: string, + options: { ownerId?: string; leaseTtlMs?: number } = {}, + ): Promise { + const res = await this.sendRequest({ + action: "LOCK_ACQUIRE", + resourceKey, + ownerId: options.ownerId, + leaseTtlMs: options.leaseTtlMs, + }); + return res.ok ? res.lock : null; + } + + public async releaseLock( + resourceKey: string, + lockId: string, + ownerId?: string, + ): Promise { + const res = await this.sendRequest({ + action: "LOCK_RELEASE", + resourceKey, + lockId, + ownerId, + }); + return Boolean(res.ok); + } + + public async streamOtelMetric( + metricName: string, + value: number, + attributes: Record = {}, + type: "counter" | "histogram" | "gauge" | "audit_event" = "counter", + ): Promise { + const res = await this.sendRequest({ + action: "STREAM_OTEL", + metricName, + value, + attributes, + type, + }); + return res.ok ? res.metricId : null; + } + + public async getMetricsSummary(): Promise { + return this.sendRequest({ action: "GET_METRICS" }); + } + + private sendRequest(payload: any): Promise { + if (!this.socket || !this.socket.writable) { + return Promise.reject(new Error("AgentShieldDaemonClient is not connected")); + } + + const requestId = crypto.randomUUID().slice(0, 8); + const msg = { ...payload, id: requestId }; + + return new Promise((resolve, reject) => { + this.pendingRequests.set(requestId, { resolve, reject }); + this.socket!.write(`${JSON.stringify(msg)}\n`); + }); + } + + private initSocketListeners(socket: net.Socket): void { + socket.on("data", (chunk) => { + this.incomingBuffer += chunk.toString("utf8"); + let newlineIdx = this.incomingBuffer.indexOf("\n"); + + while (newlineIdx !== -1) { + const line = this.incomingBuffer.slice(0, newlineIdx).trim(); + this.incomingBuffer = this.incomingBuffer.slice(newlineIdx + 1); + + if (line) { + try { + const parsed = JSON.parse(line); + if (parsed.id && this.pendingRequests.has(parsed.id)) { + const { resolve } = this.pendingRequests.get(parsed.id)!; + this.pendingRequests.delete(parsed.id); + resolve(parsed); + } + } catch { + // ignore malformed response frames + } + } + newlineIdx = this.incomingBuffer.indexOf("\n"); + } + }); + + socket.on("close", () => { + this.disconnect(); + }); + } +} diff --git a/src/index.ts b/src/index.ts index ddf2045..a78309c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -44,3 +44,4 @@ export * from "./transparency-ledger.js"; export * from "./ebpf-lsm.js"; export * from "./distributed-key-broker.js"; export * from "./siem-streamer.js"; +export * from "./daemon.js"; diff --git a/tests/daemon-ipc.test.ts b/tests/daemon-ipc.test.ts new file mode 100644 index 0000000..4bd5840 --- /dev/null +++ b/tests/daemon-ipc.test.ts @@ -0,0 +1,154 @@ +import assert from "node:assert/strict"; +import * as crypto from "node:crypto"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import test from "node:test"; +import { AgentShieldDaemon, AgentShieldDaemonClient } from "../src/daemon.js"; + +test("Phase 2: Headless Background Daemon (agentshieldd) Local IPC Suite", async (t) => { + // Use unique ephemeral socket path for the test suite + const uniqueId = Math.random().toString(36).substring(2, 9); + const testSocketPath = + process.platform === "win32" + ? `\\\\.\\pipe\\agentshield-test-${uniqueId}` + : path.join(os.tmpdir(), `agentshield-test-${uniqueId}.sock`); + + const dynamicAuthSecret = Buffer.from("test-daemon-auth-token-key-32b").toString("hex"); + + const daemon = new AgentShieldDaemon({ + socketPath: testSocketPath, + authSecret: dynamicAuthSecret, + }); + + await daemon.start(); + + t.after(async () => { + await daemon.stop(); + }); + + await t.test("verifies socket binding and initial daemon status", async () => { + const status = daemon.getStatus(); + assert.ok(status.version); + assert.equal(status.activeLocks, 0); + assert.equal(status.connectedClients, 0); + assert.equal(status.socketPath, testSocketPath); + }); + + await t.test("client connects and performs unauthenticated PING", async () => { + const client = new AgentShieldDaemonClient(testSocketPath); + await client.connect(); + + try { + const pong = await client.ping(); + assert.equal(pong.ok, true); + assert.equal(pong.action, "PONG"); + assert.equal(pong.authenticated, false); + } finally { + client.disconnect(); + } + }); + + await t.test("client fails protected actions before AUTH handshake", async () => { + const client = new AgentShieldDaemonClient(testSocketPath); + await client.connect(); + + try { + const lockRes = await client.acquireLock("resource:ci-pipeline"); + assert.equal(lockRes, null); + } finally { + client.disconnect(); + } + }); + + await t.test("client authenticates and synchronizes process locks across replicas", async () => { + const clientA = new AgentShieldDaemonClient(testSocketPath); + const clientB = new AgentShieldDaemonClient(testSocketPath); + + await clientA.connect(); + await clientB.connect(); + + try { + const authA = await clientA.authenticate(dynamicAuthSecret); + assert.equal(authA, true, "Client A must authenticate"); + + const authB = await clientB.authenticate(dynamicAuthSecret); + assert.equal(authB, true, "Client B must authenticate"); + + // 1. Client A acquires lock + const lockA = await clientA.acquireLock("resource:db-migration", { + ownerId: "agent_worker_1", + leaseTtlMs: 4000, + }); + + assert.ok(lockA, "Client A must acquire lock"); + assert.equal(lockA.resourceKey, "resource:db-migration"); + assert.equal(lockA.ownerId, "agent_worker_1"); + assert.ok(lockA.fenceToken > 0); + + // 2. Client B attempts concurrent acquisition (must fail due to mutual exclusion) + const lockB = await clientB.acquireLock("resource:db-migration", { + ownerId: "agent_worker_2", + }); + assert.equal(lockB, null, "Client B must be blocked by mutual exclusion"); + + // 3. Client A releases lock + const released = await clientA.releaseLock( + "resource:db-migration", + lockA.lockId, + "agent_worker_1", + ); + assert.equal(released, true); + + // 4. Client B can now acquire cleanly + const lockBAfter = await clientB.acquireLock("resource:db-migration", { + ownerId: "agent_worker_2", + }); + assert.ok(lockBAfter, "Client B must acquire once unlocked"); + assert.ok(lockBAfter.fenceToken > lockA.fenceToken, "Fencing token must strictly increase"); + + await clientB.releaseLock("resource:db-migration", lockBAfter.lockId, "agent_worker_2"); + } finally { + clientA.disconnect(); + clientB.disconnect(); + } + }); + + await t.test( + "client streams OpenTelemetry metrics with sub-millisecond wire latency", + async () => { + const client = new AgentShieldDaemonClient(testSocketPath); + await client.connect(); + + try { + await client.authenticate(dynamicAuthSecret); + + const t0 = performance.now(); + const metricId = await client.streamOtelMetric( + "agentshield.surrogate_redactions_total", + 1, + { + sandbox: "bubblewrap", + rule: "SECRET_REDACTION", + tenant: "enterprise_acme", + }, + "counter", + ); + const wireLatencyMs = performance.now() - t0; + + assert.ok(metricId, "Metric ID must be returned"); + assert.ok(wireLatencyMs < 50, `Wire IPC latency was ${wireLatencyMs}ms`); + + const summary = await client.getMetricsSummary(); + assert.equal(summary.ok, true); + assert.ok(summary.status.totalOtelEvents >= 1); + assert.equal( + summary.metrics[summary.metrics.length - 1].metricName, + "agentshield.surrogate_redactions_total", + ); + } finally { + client.disconnect(); + } + }, + ); +}); From e24e71fa673f496f6b91f827c750e2675d101f21 Mon Sep 17 00:00:00 2001 From: divyanshujethi Date: Sun, 8 Nov 2026 21:41:19 +0530 Subject: [PATCH 2/2] style(daemon): eliminate explicit any and non-null assertions to pass biome lint --- src/daemon.ts | 31 +++++++++++++++++-------------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/src/daemon.ts b/src/daemon.ts index 3c8ee3c..2de25d2 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -324,12 +324,13 @@ export class AgentShieldDaemon { } this.sendResponse(socket, { id, ok: false, error: `Unknown daemon action: ${action}` }); - } catch (err: any) { - this.sendResponse(socket, { ok: false, error: `Malformed IPC message: ${err.message}` }); + } catch (err: unknown) { + const errMsg = err instanceof Error ? err.message : String(err); + this.sendResponse(socket, { ok: false, error: `Malformed IPC message: ${errMsg}` }); } } - private sendResponse(socket: net.Socket, data: any): void { + private sendResponse(socket: net.Socket, data: Record): void { if (socket.writable) { socket.write(`${JSON.stringify(data)}\n`); } @@ -415,7 +416,7 @@ export class AgentShieldDaemonClient { private socketPath: string; private pendingRequests: Map< string, - { resolve: (val: any) => void; reject: (err: any) => void } + { resolve: (val: Record) => void; reject: (err: unknown) => void } > = new Map(); private incomingBuffer = ""; @@ -460,7 +461,7 @@ export class AgentShieldDaemonClient { this.pendingRequests.clear(); } - public async ping(): Promise { + public async ping(): Promise> { return this.sendRequest({ action: "PING" }); } @@ -479,7 +480,7 @@ export class AgentShieldDaemonClient { ownerId: options.ownerId, leaseTtlMs: options.leaseTtlMs, }); - return res.ok ? res.lock : null; + return res.ok ? (res.lock as DaemonLockHandle) : null; } public async releaseLock( @@ -499,7 +500,7 @@ export class AgentShieldDaemonClient { public async streamOtelMetric( metricName: string, value: number, - attributes: Record = {}, + attributes: Record = {}, type: "counter" | "histogram" | "gauge" | "audit_event" = "counter", ): Promise { const res = await this.sendRequest({ @@ -509,14 +510,14 @@ export class AgentShieldDaemonClient { attributes, type, }); - return res.ok ? res.metricId : null; + return res.ok ? (res.metricId as string) : null; } - public async getMetricsSummary(): Promise { + public async getMetricsSummary(): Promise> { return this.sendRequest({ action: "GET_METRICS" }); } - private sendRequest(payload: any): Promise { + private sendRequest(payload: Record): Promise> { if (!this.socket || !this.socket.writable) { return Promise.reject(new Error("AgentShieldDaemonClient is not connected")); } @@ -526,7 +527,7 @@ export class AgentShieldDaemonClient { return new Promise((resolve, reject) => { this.pendingRequests.set(requestId, { resolve, reject }); - this.socket!.write(`${JSON.stringify(msg)}\n`); + this.socket?.write(`${JSON.stringify(msg)}\n`); }); } @@ -543,9 +544,11 @@ export class AgentShieldDaemonClient { try { const parsed = JSON.parse(line); if (parsed.id && this.pendingRequests.has(parsed.id)) { - const { resolve } = this.pendingRequests.get(parsed.id)!; - this.pendingRequests.delete(parsed.id); - resolve(parsed); + const req = this.pendingRequests.get(parsed.id); + if (req) { + this.pendingRequests.delete(parsed.id); + req.resolve(parsed); + } } } catch { // ignore malformed response frames