fix(cli): recover macOS vault daemons without a connection race

.changeset/macos-vault-listener-recovery.md

@@ -0,0 +1,6 @@
+---
+'@inflowpayai/inflow': patch
+---
+
+Recover from incompatible macOS vault daemons by identifying their listening process without opening a connection that
+the daemon can reject before identification completes.

packages/core/native/vault_peer_darwin.c

@@ -3,8 +3,11 @@
 #include <libproc.h>
 #include <node_api.h>
 #include <stdint.h>
+#include <stdlib.h>
 #include <string.h>
+#include <sys/proc_info.h>
 #include <sys/socket.h>
+#include <sys/stat.h>
 #include <sys/types.h>
 #include <sys/un.h>
 #include <unistd.h>
@@ -75,10 +78,114 @@ static napi_value peer_info(napi_env env, napi_callback_info info) {
   return result;
 }
 
+static napi_value listener_info(napi_env env, napi_callback_info info) {
+  size_t argc = 1;
+  napi_value argv[1];
+  char socket_path[sizeof(((struct sockaddr_un *)0)->sun_path)];
+  size_t length = 0;
+  struct stat before, after, parent;
+  pid_t *pids = NULL;
+  struct proc_fdinfo *fds = NULL;
+  pid_t owner = 0;
+  int owner_fd = -1;
+  uint64_t owner_socket = 0;
+  struct proc_bsdinfo owner_process = {0};
+  if (napi_get_cb_info(env, info, &argc, argv, NULL, NULL) != napi_ok || argc != 1 ||
+      napi_get_value_string_utf8(env, argv[0], NULL, 0, &length) != napi_ok ||
+      length == 0 || length >= sizeof(socket_path) ||
+      napi_get_value_string_utf8(env, argv[0], socket_path, sizeof(socket_path), &length) != napi_ok ||
+      strlen(socket_path) != length || socket_path[0] != '/') goto failure;
+  if (lstat(socket_path, &before) != 0 || !S_ISSOCK(before.st_mode) || before.st_uid != getuid()) goto failure;
+  char parent_path[sizeof(socket_path)];
+  memcpy(parent_path, socket_path, length + 1);
+  *strrchr(parent_path, '/') = '\0';
+  if (lstat(parent_path, &parent) != 0 || !S_ISDIR(parent.st_mode) || parent.st_uid != getuid() ||
+      (parent.st_mode & 0022) != 0) goto failure;
+
+  int pid_bytes = proc_listpids(PROC_UID_ONLY, getuid(), NULL, 0);
+  if (pid_bytes <= 0 || pid_bytes > 1024 * 1024) goto failure;
+  pid_bytes += 4096;
+  pids = malloc((size_t)pid_bytes);
+  if (pids == NULL) goto failure;
+  int pid_used = proc_listpids(PROC_UID_ONLY, getuid(), pids, pid_bytes);
+  if (pid_used <= 0 || pid_used >= pid_bytes || pid_used % sizeof(pid_t) != 0) goto failure;
+  for (size_t i = 0; i < (size_t)pid_used / sizeof(pid_t); i++) {
+    if (pids[i] <= 0) continue;
+    int fd_bytes = proc_pidinfo(pids[i], PROC_PIDLISTFDS, 0, NULL, 0);
+    if (fd_bytes <= 0) {
+      if (errno == EPERM || errno == EACCES) goto failure;
+      continue;
+    }
+    if (fd_bytes > 16 * 1024 * 1024) goto failure;
+    fd_bytes += (int)(64 * sizeof(struct proc_fdinfo));
+    fds = malloc((size_t)fd_bytes);
+    if (fds == NULL) goto failure;
+    int fd_used = proc_pidinfo(pids[i], PROC_PIDLISTFDS, 0, fds, fd_bytes);
+    if (fd_used <= 0 && (errno == EPERM || errno == EACCES)) goto failure;
+    if (fd_used >= fd_bytes || (fd_used > 0 && fd_used % sizeof(struct proc_fdinfo) != 0)) goto failure;
+    for (size_t j = 0; fd_used > 0 && j < (size_t)fd_used / sizeof(struct proc_fdinfo); j++) {
+      if (fds[j].proc_fdtype != PROX_FDTYPE_SOCKET) continue;
+      struct socket_fdinfo socket;
+      if (proc_pidfdinfo(pids[i], fds[j].proc_fd, PROC_PIDFDSOCKETINFO, &socket, sizeof(socket)) != sizeof(socket)) {
+        if (errno == EPERM || errno == EACCES) goto failure;
+        continue;
+      }
+      if (socket.psi.soi_family != AF_UNIX || socket.psi.soi_kind != SOCKINFO_UN ||
+          !(socket.psi.soi_options & SO_ACCEPTCONN) ||
+          strncmp(socket.psi.soi_proto.pri_un.unsi_addr.ua_sun.sun_path, socket_path, sizeof(socket_path)) != 0)
+        continue;
+      if (owner != 0) goto failure;
+      owner = pids[i];
+      owner_fd = fds[j].proc_fd;
+      owner_socket = socket.psi.soi_so;
+      if (proc_pidinfo(owner, PROC_PIDTBSDINFO, 0, &owner_process, sizeof(owner_process)) != sizeof(owner_process) ||
+          owner_process.pbi_uid != getuid()) goto failure;
+    }
+    free(fds);
+    fds = NULL;
+  }
+  free(pids);
+  pids = NULL;
+  if (owner == 0) goto failure;
+
+  char path[PROC_PIDPATHINFO_MAXSIZE];
+  int path_len = proc_pidpath(owner, path, sizeof(path));
+  struct proc_bsdinfo current_process;
+  struct socket_fdinfo current_socket;
+  if (path_len <= 0 ||
+      proc_pidinfo(owner, PROC_PIDTBSDINFO, 0, &current_process, sizeof(current_process)) != sizeof(current_process) ||
+      current_process.pbi_uid != owner_process.pbi_uid ||
+      current_process.pbi_start_tvsec != owner_process.pbi_start_tvsec ||
+      current_process.pbi_start_tvusec != owner_process.pbi_start_tvusec ||
+      proc_pidfdinfo(owner, owner_fd, PROC_PIDFDSOCKETINFO, &current_socket, sizeof(current_socket)) != sizeof(current_socket) ||
+      current_socket.psi.soi_so != owner_socket || !(current_socket.psi.soi_options & SO_ACCEPTCONN) ||
+      lstat(socket_path, &after) != 0 || before.st_dev != after.st_dev || before.st_ino != after.st_ino ||
+      before.st_uid != after.st_uid || !S_ISSOCK(after.st_mode)) goto failure;
+
+  napi_value result, value;
+  napi_create_object(env, &result);
+  napi_create_int32(env, owner, &value);
+  napi_set_named_property(env, result, "pid", value);
+  napi_create_uint32(env, owner_process.pbi_uid, &value);
+  napi_set_named_property(env, result, "uid", value);
+  napi_create_string_utf8(env, path, (size_t)path_len, &value);
+  napi_set_named_property(env, result, "path", value);
+  return result;
+
+failure:
+  free(fds);
+  free(pids);
+  napi_throw(env, make_error(env, "EPEERLISTENER", "Vault listener ownership could not be verified."));
+  return NULL;
+}
+
 static napi_value init(napi_env env, napi_value exports) {
   napi_value peer_info_fn;
   napi_create_function(env, "peerInfo", NAPI_AUTO_LENGTH, peer_info, NULL, &peer_info_fn);
   napi_set_named_property(env, exports, "peerInfo", peer_info_fn);
+  napi_value listener_info_fn;
+  napi_create_function(env, "listenerInfo", NAPI_AUTO_LENGTH, listener_info, NULL, &listener_info_fn);
+  napi_set_named_property(env, exports, "listenerInfo", listener_info_fn);
   if (register_vault_secure_memory(env, exports) != napi_ok) {
     return NULL;
   }

packages/core/src/secure-storage/vault-peer-recovery.ts

@@ -5,7 +5,11 @@ import process from 'node:process';
 import { setTimeout as delay } from 'node:timers/promises';
 import { SecureStorageError } from './errors.js';
 import { usesLinuxVaultService, vaultFilePaths } from './vault-files.js';
-import { createSameUserVaultSocketPeerVerifier, type VaultSocketPeer } from './vault-peer-verifier.js';
+import {
+  createSameUserVaultSocketPeerVerifier,
+  inspectSameUserVaultListener,
+  type VaultSocketPeer,
+} from './vault-peer-verifier.js';
 import { inspectVaultSocketPeer, isReachableVaultSocket } from './vault-socket.js';
 
 const SHUTDOWN_TIMEOUT_MILLISECONDS = 2_000;
@@ -40,7 +44,10 @@ const defaultProcessDependencies: VaultDaemonProcessDependencies = {
 
 const defaultDependencies: LocalVaultPeerRecoveryDependencies = {
   currentProcessId: process.pid,
-  inspectPeer: (socketPath) => inspectVaultSocketPeer(socketPath, createSameUserVaultSocketPeerVerifier()),
+  inspectPeer: async (socketPath) =>
+    process.platform === 'darwin'
+      ? inspectSameUserVaultListener(socketPath)
+      : inspectVaultSocketPeer(socketPath, createSameUserVaultSocketPeerVerifier()),
   isVaultDaemonProcess,
   now: Date.now,
   platform: process.platform,

packages/core/src/secure-storage/vault-peer-verifier.ts

@@ -21,6 +21,7 @@ export interface VaultSocketPeer {
 export type VaultSocketPeerVerifier = (socket: Socket) => Promise<VaultSocketPeer> | VaultSocketPeer;
 
 interface NativeVaultPeerModule {
+  listenerInfo?(socketPath: string): VaultSocketPeer;
   peerCredentials(fd: number): Pick<VaultSocketPeer, 'pid' | 'uid'>;
   peerInfo(fd: number): VaultSocketPeer;
 }
@@ -109,6 +110,25 @@ export function createSameUserVaultSocketPeerVerifier(
   };
 }
 
+export function inspectSameUserVaultListener(
+  socketPath: string,
+  options: VaultPeerVerifierOptions = {},
+  dependencies: VaultPeerVerifierDependencies = defaultPeerVerifierDependencies,
+): VaultSocketPeer {
+  if (process.platform !== 'darwin') {
+    throw new SecureStorageError('secure_storage_peer_verification_failed', 'Vault peer verification failed.');
+  }
+  const config = createVaultPeerVerificationConfig(options, dependencies);
+  verifyVaultPeerVerificationConfig(config, dependencies);
+  const native = dependencies.loadNativeModule(config.nativeModulePath);
+  const peer = native.listenerInfo?.(socketPath);
+  const currentUserId = dependencies.currentUserId();
+  if (peer === undefined || currentUserId === undefined || peer.uid !== currentUserId) {
+    throw new SecureStorageError('secure_storage_peer_verification_failed', 'Vault peer verification failed.');
+  }
+  return peer;
+}
+
 export function verifyTransferredVaultSocketPeer(socket: Socket, attestedPeer: VaultSocketPeer): VaultSocketPeer {
   const config = createVaultPeerVerificationConfig();
   verifyVaultPeerVerificationConfig(config);

packages/core/test/unit/secure-storage/vault-listener.test.ts

@@ -0,0 +1,81 @@
+import { once } from 'node:events';
+import { chmodSync, mkdtempSync, rmSync, symlinkSync, unlinkSync, writeFileSync } from 'node:fs';
+import { createConnection, createServer, type Server } from 'node:net';
+import { join } from 'node:path';
+import { describe, expect, it } from 'vitest';
+import { inspectSameUserVaultListener } from '../../../src/secure-storage/vault-peer-verifier.js';
+
+describe.runIf(process.platform === 'darwin')('macOS native vault listener inspection', () => {
+  it('identifies an immediately rejecting listener without making another connection', async () => {
+    const root = mkdtempSync('/tmp/inflow-listener-');
+    const socketPath = join(root, 'vault.sock');
+    let connections = 0;
+    const server = createServer((socket) => {
+      connections++;
+      socket.on('error', () => undefined);
+      socket.end('rejected', () => socket.destroy());
+    });
+    try {
+      server.listen(socketPath);
+      await once(server, 'listening');
+      for (let iteration = 0; iteration < 25; iteration++) {
+        const client = createConnection(socketPath);
+        client.resume();
+        await once(client, 'close');
+        expect(inspectSameUserVaultListener(socketPath)).toMatchObject({ pid: process.pid, uid: process.getuid?.() });
+        expect(connections).toBe(iteration + 1);
+      }
+    } finally {
+      await close(server);
+      rmSync(root, { force: true, recursive: true });
+    }
+  });
+
+  it('refuses ambiguous listeners after the socket pathname has been replaced', async () => {
+    const root = mkdtempSync('/tmp/inflow-listener-');
+    const socketPath = join(root, 'vault.sock');
+    const original = createServer();
+    const replacement = createServer();
+    try {
+      original.listen(socketPath);
+      await once(original, 'listening');
+      unlinkSync(socketPath);
+      replacement.listen(socketPath);
+      await once(replacement, 'listening');
+      expect(() => inspectSameUserVaultListener(socketPath)).toThrow('Vault listener ownership could not be verified.');
+    } finally {
+      await close(replacement);
+      await close(original);
+      rmSync(root, { force: true, recursive: true });
+    }
+  });
+
+  it('refuses missing paths, regular files, symlinks, and writable socket directories', async () => {
+    const root = mkdtempSync('/tmp/inflow-listener-');
+    const socketPath = join(root, 'vault.sock');
+    const server = createServer();
+    try {
+      expect(() => inspectSameUserVaultListener(socketPath)).toThrow();
+      writeFileSync(socketPath, 'not a socket');
+      expect(() => inspectSameUserVaultListener(socketPath)).toThrow();
+      unlinkSync(socketPath);
+      server.listen(socketPath);
+      await once(server, 'listening');
+      const link = join(root, 'link.sock');
+      symlinkSync(socketPath, link);
+      expect(() => inspectSameUserVaultListener(link)).toThrow();
+      expect(() => inspectSameUserVaultListener(`${socketPath}\0extra`)).toThrow();
+      expect(() => inspectSameUserVaultListener('/'.repeat(104))).toThrow();
+      chmodSync(root, 0o777);
+      expect(() => inspectSameUserVaultListener(socketPath)).toThrow();
+    } finally {
+      chmodSync(root, 0o700);
+      await close(server);
+      rmSync(root, { force: true, recursive: true });
+    }
+  });
+});
+
+async function close(server: Server): Promise<void> {
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+}

packages/core/test/unit/secure-storage/vault-peer-recovery.test.ts

@@ -1,11 +1,14 @@
 import { spawn } from 'node:child_process';
 import { once } from 'node:events';
 import { mkdtempSync, rmSync } from 'node:fs';
+import { createConnection } from 'node:net';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { describe, expect, it, vi } from 'vitest';
 import { SecureStorageError } from '../../../src/secure-storage/errors.js';
 import { vaultFilePaths } from '../../../src/secure-storage/vault-files.js';
+import * as peerVerification from '../../../src/secure-storage/vault-peer-verifier.js';
+import * as vaultSocket from '../../../src/secure-storage/vault-socket.js';
 import {
   __testing,
   type LocalVaultPeerRecoveryDependencies,
@@ -29,9 +32,46 @@ function dependencies(overrides: Partial<LocalVaultPeerRecoveryDependencies> = {
 }
 
 describe('vault peer recovery', () => {
-  it.runIf(process.platform === 'darwin' || process.platform === 'linux')(
-    'shuts down a same-user local daemon through the public recovery boundary',
-    async () => {
+  it.runIf(process.platform === 'darwin' || process.platform === 'linux').each(['darwin', 'linux'] as const)(
+    'routes %s recovery to its operating-system inspection method',
+    async (platform) => {
+      const originalPlatform = process.platform;
+      const peer = { path: '/old/inflow', pid: 123, uid: 501 };
+      const verifier = () => peer;
+      const createVerifier = vi
+        .spyOn(peerVerification, 'createSameUserVaultSocketPeerVerifier')
+        .mockReturnValue(verifier);
+      const inspectListener = vi.spyOn(peerVerification, 'inspectSameUserVaultListener').mockReturnValue(peer);
+      const inspectSocket = vi.spyOn(vaultSocket, 'inspectVaultSocketPeer').mockResolvedValue(peer);
+      vi.spyOn(vaultSocket, 'isReachableVaultSocket').mockResolvedValue(false);
+      const signal = vi.spyOn(process, 'kill').mockImplementation(() => {
+        throw Object.assign(new Error('process exited'), { code: 'ESRCH' });
+      });
+      try {
+        Object.defineProperty(process, 'platform', { value: platform });
+        await shutdownUnverifiedLocalVaultDaemon('/isolated-vault');
+        if (platform === 'darwin') {
+          expect(inspectListener).toHaveBeenCalledWith('/isolated-vault/run/vault.sock');
+          expect(createVerifier).not.toHaveBeenCalled();
+          expect(inspectSocket).not.toHaveBeenCalled();
+        } else {
+          expect(createVerifier).toHaveBeenCalledOnce();
+          expect(inspectSocket).toHaveBeenCalledWith('/isolated-vault/run/vault.sock', verifier);
+          expect(inspectListener).not.toHaveBeenCalled();
+        }
+        expect(signal).toHaveBeenCalledWith(peer.pid, 'SIGTERM');
+      } finally {
+        Object.defineProperty(process, 'platform', { value: originalPlatform });
+        vi.restoreAllMocks();
+      }
+    },
+  );
+
+  it
+    .runIf(process.platform === 'darwin' || process.platform === 'linux')
+    .each(process.platform === 'darwin' ? [false, true] : [false])(
+    'shuts down a same-user local daemon through the public recovery boundary (reject connections: %s)',
+    async (rejectConnections) => {
       const rootDirectory = mkdtempSync(join(tmpdir(), 'inflow-vault-recovery-'));
       const socketPath = vaultFilePaths(rootDirectory).socket;
       const child = spawn(
@@ -44,9 +84,17 @@ describe('vault peer recovery', () => {
             "const { createServer } = require('node:net');",
             'const socketPath = process.argv[1];',
             'mkdirSync(dirname(socketPath), { recursive: true });',
-            'const server = createServer(() => {});',
+            'let connections = 0;',
+            `const rejectConnections = ${String(rejectConnections)};`,
+            'const server = createServer(socket => {',
+            'connections++;',
+            'if (rejectConnections) {',
+            "socket.on('error', () => {});",
+            "socket.end('rejected', () => socket.destroy());",
+            '}',
+            '});',
             "server.listen(socketPath, () => process.send('ready'));",
-            "process.on('SIGTERM', () => server.close(() => process.exit(0)));",
+            "process.on('SIGTERM', () => server.close(() => process.exit(rejectConnections && connections !== 1 ? 1 : 0)));",
           ].join(''),
           socketPath,
           '--daemon',
@@ -56,6 +104,11 @@ describe('vault peer recovery', () => {
       );
       try {
         await once(child, 'message');
+        if (rejectConnections) {
+          const client = createConnection(socketPath);
+          client.resume();
+          await once(client, 'close');
+        }
         await expect(shutdownUnverifiedLocalVaultDaemon(rootDirectory)).resolves.toBeUndefined();
         expect(child.exitCode).toBe(0);
       } finally {