mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 16:03:26 +00:00
fix(daemon): keep audit-only evidence off the main thread and out of false negatives
Three audit-only daemon evidence defects:
- The Windows CIM probe emitted the "absent" marker whenever $p was empty, so a
non-terminating Get-CimInstance failure (Winmgmt down, corrupt WMI repository,
access denied) exited 0 with {"exists":false} and was recorded as proof the
daemon process was gone. The script now sets $ErrorActionPreference and reports
query failure explicitly; only a query that ran and found nothing yields
'missing', everything else stays 'unavailable'.
- probeMacosProcess read the process start time through execFileSync('ps') on the
Electron main thread. It now uses the async readMacosProcessStartedAtMs, matching
the async command-line sibling in the same probe.
- daemon_audit_eligibility fired on every successful inventory, so routine
listProcesses traffic could drain the shared 1,000-event per-session telemetry
ceiling. Repeated identical observations now collapse to one heartbeat per
5 minutes; a changed observation still emits immediately.
The evidence stays strictly audit-only — no lifecycle, routing or recovery path
consumes the observation.
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -1,11 +1,20 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { validate } from '../telemetry/validator'
|
||||
import { recordAuthenticatedInventory, type DaemonAuditContext } from './daemon-audit-classifier'
|
||||
import { DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
import { DaemonServer } from './daemon-server'
|
||||
import { getDaemonSocketPath } from './daemon-spawner'
|
||||
|
||||
const { trackMock } = vi.hoisted(() => ({ trackMock: vi.fn() }))
|
||||
vi.mock('../telemetry/client', () => ({ track: trackMock }))
|
||||
|
||||
import { trackDaemonAuditEligibility } from './daemon-audit-eligibility-event'
|
||||
import {
|
||||
createDaemonAuditEligibilityTracker,
|
||||
trackDaemonAuditEligibility
|
||||
} from './daemon-audit-eligibility-event'
|
||||
|
||||
const context: DaemonAuditContext = {
|
||||
protocolGeneration: 23,
|
||||
@@ -48,4 +57,76 @@ describe('daemon audit eligibility telemetry', () => {
|
||||
trackDaemonAuditEligibility(recordAuthenticatedInventory(context, null))
|
||||
).not.toThrow()
|
||||
})
|
||||
|
||||
it('collapses repeated identical observations into one heartbeat per window', () => {
|
||||
let nowMs = 1_700_000_000_000
|
||||
const trackEligibility = createDaemonAuditEligibilityTracker(() => nowMs)
|
||||
|
||||
for (let index = 0; index < 60; index += 1) {
|
||||
nowMs += 1_000
|
||||
trackEligibility(recordAuthenticatedInventory(context, null))
|
||||
}
|
||||
|
||||
expect(trackMock).toHaveBeenCalledOnce()
|
||||
|
||||
nowMs += 5 * 60_000
|
||||
trackEligibility(recordAuthenticatedInventory(context, null))
|
||||
expect(trackMock).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('emits immediately when the observation changes', () => {
|
||||
let nowMs = 1_700_000_000_000
|
||||
const trackEligibility = createDaemonAuditEligibilityTracker(() => nowMs)
|
||||
|
||||
trackEligibility(recordAuthenticatedInventory(context, null))
|
||||
nowMs += 1_000
|
||||
trackEligibility(
|
||||
recordAuthenticatedInventory(context, {
|
||||
identity: { pid: 42, startedAtMs: nowMs, launchNonce: 'launch-a' }
|
||||
})
|
||||
)
|
||||
|
||||
expect(trackMock).toHaveBeenCalledTimes(2)
|
||||
expect(trackMock.mock.calls[1][1]).toMatchObject({ exact_incarnation: 'endpoint-identity' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('daemon audit eligibility inventory volume', () => {
|
||||
let dir: string
|
||||
let socketPath: string
|
||||
let tokenPath: string
|
||||
let server: DaemonServer
|
||||
let adapter: DaemonPtyAdapter
|
||||
|
||||
beforeEach(() => {
|
||||
dir = mkdtempSync(join(tmpdir(), 'daemon-audit-eligibility-'))
|
||||
socketPath = getDaemonSocketPath(dir)
|
||||
tokenPath = join(dir, 'daemon.token')
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
adapter?.dispose()
|
||||
await server?.shutdown()
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
it('does not emit one eligibility event per successful inventory', async () => {
|
||||
server = new DaemonServer({
|
||||
socketPath,
|
||||
tokenPath,
|
||||
spawnSubprocess: () => {
|
||||
throw new Error('Test must not create a PTY')
|
||||
}
|
||||
})
|
||||
await server.start()
|
||||
adapter = new DaemonPtyAdapter({ socketPath, tokenPath })
|
||||
|
||||
for (let index = 0; index < 40; index += 1) {
|
||||
await adapter.listProcesses()
|
||||
}
|
||||
|
||||
expect(
|
||||
trackMock.mock.calls.filter(([name]) => name === 'daemon_audit_eligibility')
|
||||
).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,29 +1,61 @@
|
||||
import { track } from '../telemetry/client'
|
||||
import type { EventProps } from '../../shared/telemetry-events'
|
||||
import type { DaemonAuditObservation } from './daemon-audit-classifier'
|
||||
|
||||
// Why: a steady daemon repeats a byte-identical observation on every listProcesses call, so
|
||||
// repeats are re-sent only as an occasional heartbeat — the shared per-session telemetry
|
||||
// ceiling is 1,000 events for the whole app and audit data must not crowd it out.
|
||||
const REPEATED_OBSERVATION_INTERVAL_MS = 5 * 60_000
|
||||
|
||||
export function trackDaemonAuditEligibility(observation: DaemonAuditObservation): void {
|
||||
try {
|
||||
track('daemon_audit_eligibility', {
|
||||
state: observation.state,
|
||||
reason: observation.reason,
|
||||
trigger: observation.trigger,
|
||||
evidence_sources: [...observation.evidenceSources],
|
||||
protocol_generation: observation.context.protocolGeneration,
|
||||
provider: observation.context.provider,
|
||||
endpoint_kind: observation.context.endpointKind,
|
||||
profile_scope: observation.context.profileScope ? 'configured' : 'unspecified',
|
||||
exact_incarnation: exactIncarnationKind(observation),
|
||||
reachability: observation.reachability,
|
||||
inventory_authority: observation.inventoryAuthority,
|
||||
process_liveness: observation.processLiveness,
|
||||
process_reason: observation.processReason,
|
||||
endpoint_state: observation.endpointState
|
||||
})
|
||||
track('daemon_audit_eligibility', auditEligibilityProperties(observation))
|
||||
} catch {
|
||||
// Audit telemetry cannot affect daemon availability.
|
||||
}
|
||||
}
|
||||
|
||||
export function createDaemonAuditEligibilityTracker(
|
||||
now: () => number = Date.now
|
||||
): (observation: DaemonAuditObservation) => void {
|
||||
let lastProperties: string | null = null
|
||||
let lastTrackedAtMs = 0
|
||||
return (observation) => {
|
||||
const properties = JSON.stringify(auditEligibilityProperties(observation))
|
||||
const observedAtMs = now()
|
||||
if (
|
||||
properties === lastProperties &&
|
||||
observedAtMs - lastTrackedAtMs < REPEATED_OBSERVATION_INTERVAL_MS
|
||||
) {
|
||||
return
|
||||
}
|
||||
lastProperties = properties
|
||||
lastTrackedAtMs = observedAtMs
|
||||
trackDaemonAuditEligibility(observation)
|
||||
}
|
||||
}
|
||||
|
||||
function auditEligibilityProperties(
|
||||
observation: DaemonAuditObservation
|
||||
): EventProps<'daemon_audit_eligibility'> {
|
||||
return {
|
||||
state: observation.state,
|
||||
reason: observation.reason,
|
||||
trigger: observation.trigger,
|
||||
evidence_sources: [...observation.evidenceSources],
|
||||
protocol_generation: observation.context.protocolGeneration,
|
||||
provider: observation.context.provider,
|
||||
endpoint_kind: observation.context.endpointKind,
|
||||
profile_scope: observation.context.profileScope ? 'configured' : 'unspecified',
|
||||
exact_incarnation: exactIncarnationKind(observation),
|
||||
reachability: observation.reachability,
|
||||
inventory_authority: observation.inventoryAuthority,
|
||||
process_liveness: observation.processLiveness,
|
||||
process_reason: observation.processReason,
|
||||
endpoint_state: observation.endpointState
|
||||
}
|
||||
}
|
||||
|
||||
function exactIncarnationKind(
|
||||
observation: DaemonAuditObservation
|
||||
): 'endpoint-identity' | 'endpoint-identity-linux-ticks' | 'unavailable' {
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
import type * as ChildProcessModule from 'node:child_process'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { ExactDaemonIncarnation } from './daemon-incarnation-evidence-types'
|
||||
|
||||
const { execFileMock, execFileSyncMock } = vi.hoisted(() => ({
|
||||
execFileMock: vi.fn(
|
||||
(
|
||||
_file: string,
|
||||
_args: readonly string[],
|
||||
_options: unknown,
|
||||
callback: (error: Error | null, result: { stdout: string; stderr: string }) => void
|
||||
) => {
|
||||
callback(null, { stdout: `${new Date(1_700_000_000_000).toString()}\n`, stderr: '' })
|
||||
}
|
||||
),
|
||||
execFileSyncMock: vi.fn(() => '')
|
||||
}))
|
||||
|
||||
vi.mock('node:child_process', async (importOriginal) => ({
|
||||
...(await importOriginal<typeof ChildProcessModule>()),
|
||||
execFile: execFileMock,
|
||||
execFileSync: execFileSyncMock
|
||||
}))
|
||||
|
||||
const { probeDaemonProcessIdentity } = await import('./daemon-incarnation-evidence')
|
||||
|
||||
const endpoint = { socketPath: '/runtime/daemon.sock', tokenPath: '/runtime/daemon.token' }
|
||||
const exactIncarnation: ExactDaemonIncarnation = {
|
||||
identity: { pid: 42, startedAtMs: 1_700_000_000_000, launchNonce: 'launch-a' }
|
||||
}
|
||||
|
||||
describe('daemon audit evidence main-thread cost', () => {
|
||||
it('reads the macOS process start time without a synchronous main-thread spawn', async () => {
|
||||
await expect(
|
||||
probeDaemonProcessIdentity(exactIncarnation, endpoint, {
|
||||
platform: 'darwin',
|
||||
signalProcess: () => 'occupied',
|
||||
readCommandLine: async () =>
|
||||
`node daemon-entry --socket ${endpoint.socketPath} --token ${endpoint.tokenPath}`
|
||||
})
|
||||
).resolves.toMatchObject({ state: 'present', reason: 'macos_identity_match' })
|
||||
|
||||
expect(execFileSyncMock).not.toHaveBeenCalled()
|
||||
expect(execFileMock).toHaveBeenCalledWith(
|
||||
'ps',
|
||||
['-p', '42', '-o', 'lstart='],
|
||||
expect.anything(),
|
||||
expect.any(Function)
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -70,6 +70,6 @@ export type DaemonProcessProbeDependencies = {
|
||||
readLinuxStat?: (pid: number) => Promise<LinuxStatEvidence>
|
||||
readBootIdentity?: () => Promise<string | undefined>
|
||||
readCommandLine?: (pid: number, platform: NodeJS.Platform) => Promise<string | undefined>
|
||||
readProcessStartedAtMs?: (pid: number) => number | null
|
||||
readProcessStartedAtMs?: (pid: number) => Promise<number | null>
|
||||
queryWindowsProcess?: (pid: number) => Promise<WindowsProcessEvidence>
|
||||
}
|
||||
|
||||
@@ -61,7 +61,7 @@ describe('daemon process identity evidence', () => {
|
||||
})
|
||||
|
||||
it('proves Linux pid reuse from native start ticks without derived milliseconds', async () => {
|
||||
const readProcessStartedAtMs = vi.fn(() => exactIncarnation.identity.startedAtMs)
|
||||
const readProcessStartedAtMs = vi.fn(async () => exactIncarnation.identity.startedAtMs)
|
||||
|
||||
await expect(
|
||||
probeDaemonProcessIdentity(
|
||||
@@ -247,7 +247,7 @@ describe('daemon process identity evidence', () => {
|
||||
signalProcess: () => 'occupied',
|
||||
readCommandLine: async () =>
|
||||
`node daemon-entry --socket ${endpoint.socketPath} --token ${endpoint.tokenPath}`,
|
||||
readProcessStartedAtMs: () => exactIncarnation.identity.startedAtMs + 2_500
|
||||
readProcessStartedAtMs: async () => exactIncarnation.identity.startedAtMs + 2_500
|
||||
})
|
||||
).resolves.toMatchObject({ state: 'unknown', reason: 'macos_start_time_mismatch' })
|
||||
})
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
inspectProcessSignal,
|
||||
queryWindowsProcess,
|
||||
readLinuxStat,
|
||||
readMacosProcessStartedAtMs,
|
||||
readProcessCommandLine
|
||||
} from './daemon-process-inspection'
|
||||
|
||||
@@ -34,6 +35,11 @@ export {
|
||||
|
||||
const POSIX_START_TIME_TOLERANCE_MS = 1_500
|
||||
|
||||
// Why: linux start time comes from procfs via the shared helper; only darwin's `ps` spawn
|
||||
// had to move off the main thread.
|
||||
const readLinuxProcessStartedAtMs = async (pid: number): Promise<number | null> =>
|
||||
getProcessStartedAtMs(pid)
|
||||
|
||||
export async function probeDaemonProcessIdentity(
|
||||
exactIncarnation: ExactDaemonIncarnation | null,
|
||||
endpoint: { socketPath: string; tokenPath: string },
|
||||
@@ -134,7 +140,7 @@ async function probeLinuxProcess(
|
||||
])
|
||||
}
|
||||
|
||||
const startedAtMs = (dependencies.readProcessStartedAtMs ?? getProcessStartedAtMs)(
|
||||
const startedAtMs = await (dependencies.readProcessStartedAtMs ?? readLinuxProcessStartedAtMs)(
|
||||
exactIncarnation.identity.pid
|
||||
)
|
||||
if (startedAtMs === null) {
|
||||
@@ -172,7 +178,7 @@ async function probeMacosProcess(
|
||||
if (!commandLineMatchesDaemon(commandLine, endpoint.socketPath, endpoint.tokenPath)) {
|
||||
return unknown('command_line_mismatch', ['process_command_line'])
|
||||
}
|
||||
const startedAtMs = (dependencies.readProcessStartedAtMs ?? getProcessStartedAtMs)(
|
||||
const startedAtMs = await (dependencies.readProcessStartedAtMs ?? readMacosProcessStartedAtMs)(
|
||||
exactIncarnation.identity.pid
|
||||
)
|
||||
if (startedAtMs === null) {
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { queryWindowsProcess, readProcessCommandLine } from './daemon-process-inspection'
|
||||
import {
|
||||
queryWindowsProcess,
|
||||
readMacosProcessStartedAtMs,
|
||||
readProcessCommandLine
|
||||
} from './daemon-process-inspection'
|
||||
|
||||
describe('daemon process inspection', () => {
|
||||
it('falls back to ps when Linux procfs returns an empty command line', async () => {
|
||||
@@ -23,6 +27,50 @@ describe('daemon process inspection', () => {
|
||||
expect(runCommand).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('asks PowerShell to report a failed CIM query instead of an absent process', async () => {
|
||||
const runCommand = vi.fn(
|
||||
async (_file: string, _args: string[], _timeoutMs: number) =>
|
||||
'{"status":"present","cmd":"daemon","start":1}'
|
||||
)
|
||||
|
||||
await queryWindowsProcess(42, { runCommand })
|
||||
|
||||
const script = runCommand.mock.calls[0]?.[1].at(-1) ?? ''
|
||||
expect(script).toContain("$ErrorActionPreference = 'Stop'")
|
||||
expect(script).toMatch(/catch \{[^}]*query_failed/)
|
||||
})
|
||||
|
||||
it('keeps a failed CIM query indeterminate instead of proving the process gone', async () => {
|
||||
const runCommand = vi.fn(async () => '{"status":"query_failed"}')
|
||||
|
||||
await expect(queryWindowsProcess(42, { runCommand })).resolves.toEqual({
|
||||
status: 'unavailable'
|
||||
})
|
||||
})
|
||||
|
||||
it('never reads a probe result without a success marker as proof of absence', async () => {
|
||||
const runCommand = vi.fn(async () => '{"exists":false}')
|
||||
|
||||
await expect(queryWindowsProcess(42, { runCommand })).resolves.toEqual({
|
||||
status: 'unavailable'
|
||||
})
|
||||
})
|
||||
|
||||
it('reports absence only from a CIM query that ran and found nothing', async () => {
|
||||
const runCommand = vi.fn(async () => '{"status":"missing"}')
|
||||
|
||||
await expect(queryWindowsProcess(42, { runCommand })).resolves.toEqual({ status: 'missing' })
|
||||
})
|
||||
|
||||
it('reads the macOS start time through an async spawn', async () => {
|
||||
const runCommand = vi.fn(async () => 'Sat Jan 1 00:00:00 2028\n')
|
||||
|
||||
await expect(readMacosProcessStartedAtMs(42, { runCommand })).resolves.toBe(
|
||||
Date.parse('Sat Jan 1 00:00:00 2028')
|
||||
)
|
||||
expect(runCommand).toHaveBeenCalledWith('ps', ['-p', '42', '-o', 'lstart='], 2_000)
|
||||
})
|
||||
|
||||
it.each([0, -1, 1.5, Number.MAX_SAFE_INTEGER + 1, Number.NaN])(
|
||||
'rejects unsafe Windows pid %s before command interpolation',
|
||||
async (pid) => {
|
||||
|
||||
@@ -63,6 +63,9 @@ export async function readProcessCommandLine(
|
||||
}
|
||||
}
|
||||
|
||||
// Why: Get-CimInstance errors (Winmgmt down, corrupt WMI repository, access denied) are
|
||||
// non-terminating and exit 0 with an empty $p, which is indistinguishable from "no such
|
||||
// process" — so the script reports query failure explicitly instead of asserting absence.
|
||||
export async function queryWindowsProcess(
|
||||
pid: number,
|
||||
dependencies: DaemonProcessInspectionDependencies = {}
|
||||
@@ -78,23 +81,26 @@ export async function queryWindowsProcess(
|
||||
'-NoProfile',
|
||||
'-NonInteractive',
|
||||
'-Command',
|
||||
`$p = Get-CimInstance Win32_Process -Filter "ProcessId = ${pid}"; ` +
|
||||
`if (!$p) { @{ exists = $false } | ConvertTo-Json -Compress } else { ` +
|
||||
`$ErrorActionPreference = 'Stop'; ` +
|
||||
`try { $p = Get-CimInstance Win32_Process -Filter "ProcessId = ${pid}" } ` +
|
||||
`catch { @{ status = 'query_failed' } | ConvertTo-Json -Compress; exit 0 }; ` +
|
||||
`if (!$p) { @{ status = 'missing' } | ConvertTo-Json -Compress; exit 0 }; ` +
|
||||
`$start = $null; if ($p.CreationDate) { ` +
|
||||
`$start = [long]([DateTimeOffset]$p.CreationDate).ToUnixTimeMilliseconds() }; ` +
|
||||
`@{ exists = $true; cmd = $p.CommandLine; start = $start } | ConvertTo-Json -Compress }`
|
||||
`@{ status = 'present'; cmd = $p.CommandLine; start = $start } | ConvertTo-Json -Compress`
|
||||
],
|
||||
3_000
|
||||
)
|
||||
const parsed = JSON.parse(stdout.trim()) as {
|
||||
exists?: unknown
|
||||
status?: unknown
|
||||
cmd?: unknown
|
||||
start?: unknown
|
||||
}
|
||||
if (parsed.exists === false) {
|
||||
// Only a query that ran and found nothing proves absence; anything else stays indeterminate.
|
||||
if (parsed.status === 'missing') {
|
||||
return { status: 'missing' }
|
||||
}
|
||||
if (parsed.exists !== true) {
|
||||
if (parsed.status !== 'present') {
|
||||
return { status: 'unavailable' }
|
||||
}
|
||||
return {
|
||||
@@ -108,6 +114,22 @@ export async function queryWindowsProcess(
|
||||
}
|
||||
}
|
||||
|
||||
// Why: `ps` is darwin's only start-time source and this audit-only probe runs in the
|
||||
// Electron main process, where the sync spawn blocks every IPC/UI turn for its duration.
|
||||
export async function readMacosProcessStartedAtMs(
|
||||
pid: number,
|
||||
dependencies: DaemonProcessInspectionDependencies = {}
|
||||
): Promise<number | null> {
|
||||
const runCommand = dependencies.runCommand ?? runInspectionCommand
|
||||
try {
|
||||
const stdout = await runCommand('ps', ['-p', String(pid), '-o', 'lstart='], 2_000)
|
||||
const startedAtMs = Date.parse(stdout.trim())
|
||||
return Number.isFinite(startedAtMs) ? startedAtMs : null
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
async function runInspectionCommand(
|
||||
file: string,
|
||||
args: string[],
|
||||
|
||||
@@ -77,7 +77,7 @@ import {
|
||||
type DaemonAuditTrigger
|
||||
} from './daemon-audit-classifier'
|
||||
import type { DaemonEvidenceSource, ExactDaemonIncarnation } from './daemon-incarnation-evidence'
|
||||
import { trackDaemonAuditEligibility } from './daemon-audit-eligibility-event'
|
||||
import { createDaemonAuditEligibilityTracker } from './daemon-audit-eligibility-event'
|
||||
|
||||
type PendingDaemonSpawnOperation = {
|
||||
exitsBySessionId: Map<string, { incarnationId?: string }[]>
|
||||
@@ -159,6 +159,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
|
||||
private lastAuthenticatedIdentity: DaemonEndpointIdentity | null = null
|
||||
private exactDaemonIncarnation: ExactDaemonIncarnation | null = null
|
||||
private lastAuditObservation: DaemonAuditObservation | null = null
|
||||
// Why: every listProcesses call republishes the same observation; unthrottled it drains the shared per-session telemetry ceiling.
|
||||
private readonly trackAuditEligibility = createDaemonAuditEligibilityTracker()
|
||||
private auditObservationListeners: ((observation: DaemonAuditObservation) => void)[] = []
|
||||
private identityChangeListeners: ((event: DaemonIdentityChangeEvent) => void)[] = []
|
||||
private historyManager: HistoryManager | null
|
||||
@@ -1723,7 +1725,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
|
||||
|
||||
private publishAuditObservation(observation: DaemonAuditObservation): void {
|
||||
this.lastAuditObservation = observation
|
||||
trackDaemonAuditEligibility(observation)
|
||||
this.trackAuditEligibility(observation)
|
||||
notifyAuditListeners(this.auditObservationListeners, observation)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user