bin/daemon/server.ts
Rama main · réplica de lectura del repositorio original.
/**
* The `cotal daemon serve` listener: binds the local endpoint, answers `status` once it is ready,
* and on shutdown removes only what it published.
*
* Readiness is real. On Linux and macOS `status` can only be answered by a listener that has taken
* the lifetime lock, passed the directory check, cleared the public path, bound a private name,
* verified it is a 0600 socket owned by this uid, and published it with link(2), which yields
* exactly one winner among concurrent starters. On Windows it is a pipe whose first instance was
* created first-instance-only, so a name anything else already serves is refused. A peer whose
* kernel-reported user differs is dropped before one byte of its input is decoded.
*
* A restarted daemon comes back by itself. The lifetime lock is kernel-held, so a predecessor
* killed with SIGKILL has already released it, and the record it left names the socket it published
* by `(dev, ino)`. Holding the lock, this daemon removes that exact socket and no other (a Windows
* pipe name dies with its holder), then resumes the service connections its predecessor held (see
* lifetime.ts, service.ts).
*/
import { chmodSync, closeSync, linkSync, unlinkSync, lstatSync } from "node:fs";
import { randomBytes } from "node:crypto";
import { join } from "node:path";
import { createServer, type Server, type Socket } from "node:net";
import { listenPipe, windowsUserSid, type PipeListener, type PipePeer, type PipeRefused } from "@cotal-ai/seat";
import { DAEMON_PROTOCOL_VERSION, MAX_LINE_BYTES, EndpointRefusal, adoptPipe, assertSocketPathFits, assertSupportedPlatform, checkSocket, defaultPeerUid, ensureDir, ensureStateDir, type DaemonPaths, type PeerUid } from "./endpoint.js";
import { claimSocketPath, generationOf, holdLifetime, type HeldLifetime } from "./lifetime.js";
import { encode, parseRequest, type ConnectCheckResult, type DaemonResponse, type DaemonStatus, type HostedManager, type RegisterResult, type ServiceConnectionResult } from "./protocol.js";
import { createLoginCoordinator, type LoginCoordinator } from "./login.js";
import { connectCheck } from "./connect.js";
import { createServiceConnections } from "./service.js";
import { homeCotalDir, isCommandFailure } from "@cotal-ai/workspace";
import { desktopSpaces } from "./desktop-state.js";
import { idpTimeoutMs, withIdpSessionDir } from "@cotal-ai/auth";
import { createHostedManagers, HostedManagerRefusal } from "./manager.js";
import { createAccountReconciler } from "./account-reconcile.js";
import { createAccountComputer } from "./account.js";
/** Requests one connection may send. A sign-in needs two (start, cancel); status polling a few. */
export const MAX_REQUESTS_PER_CONNECTION = 64;
/** Counters a test or `status` can read. Every security decision increments exactly one. */
export interface ServerCounters {
accepted: number;
peerRejected: number;
framesDecoded: number;
requestsRefused: number;
statusAnswered: number;
spacesAnswered: number;
loginsStarted: number;
loginsRefused: number;
cancelsAnswered: number;
connectChecksAnswered: number;
registersAnswered: number;
serviceAnswered: number;
/** Sockets a dead predecessor left that this start recovered: 0 or 1. */
socketsRecovered: number;
/** Service connections this start resumed from a recorded intent. */
connectionsResumed: number;
}
export interface RunningDaemon {
readonly paths: DaemonPaths;
readonly counters: Readonly<ServerCounters>;
/** Whether the socket directory was created by this start; `none` for a Windows pipe. */
readonly dirOutcome: "created" | "existing" | "none";
status(): DaemonStatus;
generationSeats(): Promise<{ space: string; seats: number }[]>;
generationPending(installed: string | undefined, waiting?: string): void;
prepareGenerationRestart(): Promise<void>;
abortGenerationRestart(): Promise<void>;
close(): Promise<void>;
readonly stopped: Promise<void>;
}
export interface ServeOptions {
readonly paths: DaemonPaths;
readonly version: string;
/** Test hook (Linux, macOS): read a peer's kernel uid. Defaults to the native reader. Windows
* reads the peer inside the native pipe accept and does not consult it. */
readonly peerUid?: PeerUid;
readonly uid?: number;
readonly log?: (line: string) => void;
/** Where the daemon keeps its sign-in session and its own per-space state. Defaults to the
* machine-home Cotal directory, where `cotal login` keeps the session. An explicit `--daemon-home`
* passes its directory here together with {@link DaemonPaths.lock}. */
readonly sessionDir?: string;
readonly updateStatus?: () => NonNullable<DaemonStatus["update"]>;
/** Test hook: the stock sign-in to drive. */
readonly login?: LoginCoordinator;
/** Shipped serve entry's receipt-verified native launch provenance. Foreground servers refuse stop. */
readonly stopAllowed?: () => Promise<boolean | { state: "owned" | "not-owned" | "unknown" }>;
/** Native startup admission, checked under the lifetime lock before publishing an endpoint. */
readonly admit?: () => void;
}
/** A published endpoint: closing it stops new connections and removes only what it published. */
interface Endpoint {
close(): Promise<void>;
}
function listen(server: Server, path: string): Promise<void> {
return new Promise((resolve, reject) => {
const onError = (e: Error) => {
server.off("listening", onListening);
reject(e);
};
const onListening = () => {
server.off("error", onError);
resolve();
};
server.once("error", onError);
server.once("listening", onListening);
server.listen(path);
});
}
export async function serveDaemon(opts: ServeOptions): Promise<RunningDaemon> {
assertSupportedPlatform();
// Refuse local IdP configuration before publishing readiness or resuming any connection.
idpTimeoutMs();
const uid = process.platform === "win32" ? undefined : (opts.uid ?? process.geteuid!());
const { paths } = opts;
// The lifetime lock comes before any endpoint path, for both daemon forms. A second serve for the
// same daemon refuses here and never reaches the socket; and because publishing the socket needs
// this lock, holding it is what makes removing a dead predecessor's socket safe below.
ensureStateDir(paths, uid);
const lock = holdLifetime(paths.lock);
try {
opts.admit?.();
return await serveLocked(opts, uid, lock);
} catch (e) {
// A failed start keeps any predecessor socket evidence it has not consumed or rejected.
// A started daemon releases the lock only in close(), clearing its own record.
lock.release({ preservePrevious: true });
throw e;
}
}
async function serveLocked(opts: ServeOptions, uid: number | undefined, lock: HeldLifetime): Promise<RunningDaemon> {
let resolveStopped!: () => void;
const stopped = new Promise<void>(resolve => { resolveStopped = resolve; });
const log = opts.log ?? (() => {});
const { paths } = opts;
let dirOutcome: RunningDaemon["dirOutcome"] = "none";
if (paths.dir !== undefined) {
if (paths.parent !== undefined) ensureDir(paths.parent, uid);
dirOutcome = ensureDir(paths.dir, uid);
}
const counters: ServerCounters = { accepted: 0, peerRejected: 0, framesDecoded: 0, requestsRefused: 0, statusAnswered: 0, spacesAnswered: 0, loginsStarted: 0, loginsRefused: 0, cancelsAnswered: 0, connectChecksAnswered: 0, registersAnswered: 0, serviceAnswered: 0, socketsRecovered: 0, connectionsResumed: 0 };
const machineHome = opts.sessionDir ?? homeCotalDir();
const login = opts.login ?? createLoginCoordinator({ sessionDir: machineHome });
const services = createServiceConnections(machineHome);
const managers = createHostedManagers(machineHome, opts.version, opts.sessionDir);
const reconciler = createAccountReconciler(machineHome, services, managers);
const account = createAccountComputer(machineHome, opts.version, reconciler);
const startedAt = new Date();
let generation: DaemonStatus["generation"];
const status = (): DaemonStatus => ({
computer: account.name(),
...(account.nameWarning() ? { computerNameWarning: account.nameWarning() } : {}),
state: "ready",
protocol: DAEMON_PROTOCOL_VERSION,
version: opts.version,
pid: process.pid,
startedAt: startedAt.toISOString(),
uptimeMs: Date.now() - startedAt.getTime(),
managers: managers.all(),
account: account.status(),
...(opts.updateStatus ? { update: opts.updateStatus() } : {}),
...(generation ? { generation } : {}),
});
const sockets = new Set<Socket>();
const track = (sock: Socket): void => {
sockets.add(sock);
sock.once("close", () => sockets.delete(sock));
sock.on("error", () => {});
};
// One connection whose peer the kernel already reported as this user: decode its requests.
const serveConnection = (sock: Socket): void => {
counters.accepted++;
let buf = "";
let bufBytes = 0;
let requests = 0;
// Sign-in attempts this connection started that have not settled. Closing the connection
// before the result cancels them through the stock fence, so nothing is persisted for a
// requester that went away. Once the result is delivered the attempt is no longer here, so a
// close after it changes nothing.
const owned = new Set<string>();
const write = (r: DaemonResponse): Promise<void> => new Promise((resolve) => {
if (sock.destroyed || !sock.writable) return resolve();
sock.write(encode(r), () => resolve());
});
const pendingReplies = new Set<Promise<unknown>>();
const replyTask = (task: Promise<unknown>): void => {
pendingReplies.add(task);
void task.finally(() => pendingReplies.delete(task)).catch(() => {});
};
const closeAfterReplies = async (): Promise<void> => {
// Stop parsing immediately, but let already accepted finite operations flush their answers.
// Device sign-in streams are not finite replies and are cancelled when this socket closes.
let timer: ReturnType<typeof setTimeout> | undefined;
try {
await Promise.race([
Promise.allSettled([...pendingReplies]),
new Promise<void>((resolve) => { timer = setTimeout(resolve, 5000); }),
]);
} finally {
if (timer) clearTimeout(timer);
}
sock.end();
sock.destroySoon();
};
const refuse = (id: string | null, code: "LOGIN_IN_PROGRESS" | "TOO_MANY_REQUESTS", message: string) => {
counters.requestsRefused++;
return write({ v: DAEMON_PROTOCOL_VERSION, id, ok: false, error: { code, message } });
};
sock.once("close", () => {
for (const attempt of owned) void login.cancel(attempt);
owned.clear();
});
let stopPrepared = false;
const handle = (line: string): void => {
counters.framesDecoded++;
const parsed = parseRequest(line);
if (!parsed.ok) {
counters.requestsRefused++;
void write({ v: DAEMON_PROTOCOL_VERSION, id: parsed.id, ok: false, error: { code: parsed.code, message: parsed.message } });
return;
}
const req = parsed.request;
if (req.op === "daemon-stop") {
// The reply acknowledges the request, not completed shutdown. The uninstaller
// independently waits for this process and endpoint to disappear.
void (async () => {
const observation = req.commit ? stopPrepared : await opts.stopAllowed?.();
const allowed = typeof observation === "object" ? observation.state === "owned" : observation === true;
if (!allowed) {
const unknown = typeof observation === "object" && observation.state === "unknown";
await write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: false, error: { code: "DEPENDENCY_UNAVAILABLE", message: unknown
? "the daemon native launch provenance is still unknown; it was not stopped"
: "the daemon has no verified native launch provenance; it was not stopped" } });
return;
}
// Only an explicit commit on this prepared connection follows native removal.
// A dropped connection or failed native step must leave the daemon running.
stopPrepared = true;
await write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, result: status() });
if (req.commit) await close();
})().catch(error => log(`daemon: shutdown failed: ${String(error)}`));
return;
}
const managerResult = (run: () => DaemonStatus | HostedManager | ServiceConnectionResult | Promise<DaemonStatus | HostedManager | ServiceConnectionResult>) => {
replyTask(Promise.resolve().then(run).then((result) => write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, result }), (error: unknown) => {
counters.requestsRefused++;
const cause = String(error instanceof Error ? error.message : error).replace(/\u001b\[[0-9;]*m/g, "").replace(/eyJ[A-Za-z0-9_.-]*/g, "<redacted>").slice(0, 1000);
log(`daemon: ${req.op} refused: ${cause}`);
if (error instanceof HostedManagerRefusal) return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: false, error: { code: error.code, message: error.message.slice(0, 1500), remedy: error.remedy.slice(0, 1500) } });
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: false, error: { code: "MANAGER_UNAVAILABLE", message: `the daemon could not complete ${req.op}: ${cause}`, remedy: "read \`cotal daemon status\`, restore the Manager connection if needed, then run the same command again" } });
}));
};
if (req.op === "manager-start" || req.op === "manager-status") {
managerResult(() => req.op === "manager-start" ? managers.start(req.space) : managers.refresh(req.space));
return;
}
if (req.op === "managers-stop") {
managerResult(async () => { await Promise.all(managers.all().map((manager) => managers.disconnect(manager.space))); return status(); });
return;
}
if (req.op === "managers-prepare-restart" || req.op === "managers-abort-restart") {
managerResult(async () => {
if (req.op === "managers-prepare-restart") await managers.prepareGenerationRestart();
else await managers.abortGenerationRestart();
return status();
});
return;
}
if (req.op === "status") {
counters.statusAnswered++;
managerResult(async () => {
await Promise.all(managers.all().map((manager) => managers.refresh(manager.space)));
const result = status();
if (!result.update) return result;
// A hosted Manager that is starting or stopping cannot give restart evidence. Status still
// answers, and an updater that needs the evidence refuses without it.
let restartEvidence: Awaited<ReturnType<typeof managers.preparedGenerationEvidence>> | undefined;
try { restartEvidence = await managers.preparedGenerationEvidence(); } catch { restartEvidence = undefined; }
return { ...result, update: { ...result.update, ...(restartEvidence ? { restartEvidence } : {}) } };
});
return;
}
if (req.op === "name") {
managerResult(async () => { if (req.name !== undefined) await account.rename(req.name); return status(); });
return;
}
if (req.op === "spaces") {
replyTask(desktopSpaces(machineHome, req.refresh === true).then((result) => {
counters.spacesAnswered++;
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, result });
}, (error: unknown) => {
counters.requestsRefused++;
if (isCommandFailure(error) && error.code === "DEPENDENCY_UNAVAILABLE")
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: false, error: { code: error.code, message: error.message, remedy: error.remedy } });
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: false, error: { code: "SPACES_UNAVAILABLE", message: "the daemon could not read its spaces" } });
}));
return;
}
if (req.op === "space-intent") {
managerResult(async () => { await account.intent(req.space, req.enabled); return status(); });
return;
}
if (req.op === "connect-check") {
replyTask(connectCheck(req.space, machineHome).catch((): ConnectCheckResult => ({ outcome: "refused", space: req.space, message: "the daemon could not complete the check" }))
.then((result) => {
counters.connectChecksAnswered++;
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, result });
}));
return;
}
if (req.op === "register") {
replyTask(withIdpSessionDir(machineHome, () => services.register(req.space)).catch((): RegisterResult => ({ outcome: "refused", space: req.space, message: "the daemon could not complete the registration request; nothing is connected" }))
.then((result) => {
counters.registersAnswered++;
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, result });
}));
return;
}
if (req.op === "service-connect" || req.op === "service-disconnect" || req.op === "service-status") {
const run = req.op === "service-status" ? Promise.resolve(services.status(req.space))
: (async () => {
if (await account.intent(req.space, req.op === "service-connect")) return services.status(req.space);
const disconnect = req.op === "service-disconnect" ? services.disconnect(req.space) : undefined;
void disconnect?.catch(() => {});
return reconciler.serial(() => withIdpSessionDir(machineHome, async () => {
if (req.op === "service-connect") {
reconciler.localAddition(req.space);
return services.connect(req.space);
}
reconciler.localRemoval(req.space);
let disconnected: ServiceConnectionResult;
try { await managers.disconnect(req.space); }
finally { disconnected = await disconnect!; }
return disconnected;
}));
})().catch((error: unknown): ServiceConnectionResult => ({ space: req.space, state: "unavailable", message: error instanceof Error ? error.message : "The connection could not be changed. Check daemon status and try again." }));
managerResult(async () => {
const result = await run;
counters.serviceAnswered++;
return { ...result, manager: req.op === "service-status" ? await managers.refresh(req.space) : managers.status(req.space) };
});
return;
}
if (req.op === "login-cancel") {
replyTask(login.cancel(req.attempt).then((result) => {
counters.cancelsAnswered++;
return write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, result });
}));
return;
}
const started = login.start({
idp: req.idp,
...(req.clientId !== undefined ? { clientId: req.clientId } : {}),
onPrompt: (attempt, prompt) => void write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, event: "login-prompt", attempt, prompt }),
onResult: async (attempt, result) => {
owned.delete(attempt);
if (result.outcome === "signed-in") account.signedIn();
await write({ v: DAEMON_PROTOCOL_VERSION, id: req.id, ok: true, event: "login-result", attempt, result });
},
});
if (!started) {
counters.loginsRefused++;
void refuse(req.id, "LOGIN_IN_PROGRESS", "a sign-in is already in progress on this daemon; cancel it or wait for its result");
return;
}
counters.loginsStarted++;
owned.add(started.attempt);
};
sock.setEncoding("utf8");
sock.on("data", (chunk: string) => {
let offset = 0;
while (offset < chunk.length) {
const nl = chunk.indexOf("\n", offset);
const part = chunk.slice(offset, nl === -1 ? chunk.length : nl);
const bytes = Buffer.byteLength(part, "utf8");
// Check this line before appending, reserving its newline even while it is incomplete.
if (bufBytes + bytes + 1 > MAX_LINE_BYTES) {
counters.requestsRefused++;
const r: DaemonResponse = { v: DAEMON_PROTOCOL_VERSION, id: null, ok: false, error: { code: "LINE_TOO_LONG", message: `request exceeds ${MAX_LINE_BYTES} bytes` } };
sock.removeAllListeners("data");
replyTask(write(r));
void closeAfterReplies();
buf = "";
bufBytes = 0;
return;
}
buf += part;
bufBytes += bytes;
if (nl === -1) return;
const line = buf;
buf = "";
bufBytes = 0;
offset = nl + 1;
if (++requests > MAX_REQUESTS_PER_CONNECTION) {
// Bounded per connection: one refusal, then the connection closes and its unread input is dropped.
sock.removeAllListeners("data");
buf = "";
bufBytes = 0;
replyTask(refuse(null, "TOO_MANY_REQUESTS", `a connection may send at most ${MAX_REQUESTS_PER_CONNECTION} requests`));
void closeAfterReplies();
return;
}
handle(line);
}
});
sock.resume();
};
let endpoint: Endpoint;
if (paths.dir === undefined) {
endpoint = publishPipe(paths, counters, log, (sock) => { track(sock); serveConnection(sock); });
} else {
const peerUid = opts.peerUid ?? defaultPeerUid;
endpoint = await publishSocket(paths, uid!, lock, counters, log, (sock) => {
track(sock);
let peer: number;
try {
peer = peerUid(sock);
} catch (e) {
counters.peerRejected++;
log(`daemon: peer credentials unreadable (${(e as Error).message}); dropped`);
sock.destroy();
return;
}
if (peer !== uid) {
counters.peerRejected++;
log(`daemon: peer uid ${peer} is not ${uid}; dropped before any frame was read`);
sock.destroy();
return;
}
serveConnection(sock);
});
}
let closed: Promise<void> | undefined;
// Read current account intent before resuming saved hosting. A successful reconciliation
// removes withdrawn intents; an unavailable account leaves them intact for offline recovery.
// The endpoint is already published, so account or space latency never delays local readiness.
const resumed = account.start().then(() => reconciler.serial(async () => {
if (closed) return;
await Promise.all([
managers.resume().catch(() => { log("daemon: hosted Manager records could not be resumed; inspect daemon status and the hosting records"); }),
withIdpSessionDir(machineHome, () => services.resume()).then((spaces) => {
counters.connectionsResumed = spaces.length;
if (spaces.length) log(`daemon: resumed ${spaces.length} recorded service connection(s)`);
}, () => {}),
]);
}));
const close = (): Promise<void> => {
closed ??= (async () => {
const servicesClosed = services.close();
const managersClosed = managers.close();
await account.close();
// In-flight account work observes service cancellation before shutdown waits for it.
// An in-flight sign-in is cancelled through the stock fence before the listener goes away.
await login.close();
await managersClosed;
// Every service connection, its endpoint and its renewal source close before the socket does.
await servicesClosed;
await resumed;
for (const s of sockets) s.destroy();
await endpoint.close();
// The lock is released only once this daemon's own endpoint is gone. If its socket is still
// published (its removal threw), the lock stays held until this process exits, so no other
// serve can start beside a live endpoint.
lock.release();
resolveStopped();
})();
return closed;
};
return { paths, counters, dirOutcome, status, close, stopped,
generationSeats: () => managers.generationSeats(),
generationPending(installed, waiting = "preparing activation") { generation = installed ? { installed, running: opts.version, waiting } : undefined; },
async prepareGenerationRestart() { await resumed; await managers.prepareGenerationRestart(); }, abortGenerationRestart: () => managers.abortGenerationRestart() };
}
/** Linux and macOS: publish the unix socket, and on close remove it only if it is still ours. */
async function publishSocket(paths: DaemonPaths, uid: number, lock: HeldLifetime, counters: ServerCounters, log: (line: string) => void, accept: (sock: Socket) => void): Promise<Endpoint> {
const server = createServer({ allowHalfOpen: false, pauseOnConnect: true }, accept);
// Clear the public path. The lifetime lock is held, so nobody else can be publishing there, and
// the only socket this may remove is the exact inode the dead predecessor's record names, once a
// connect to it is refused.
let claim: "clear" | "recover";
try { claim = await claimSocketPath(paths.socket, lock.previous?.socket); }
catch (error) {
lock.discardPrevious();
throw error;
}
if (claim === "recover") unlinkSync(paths.socket);
lock.discardPrevious();
if (claim === "recover") {
counters.socketsRecovered++;
log(`daemon: recovered the socket its predecessor left at ${paths.socket} (its lifetime lock was already released)`);
}
// Bind on a private name inside the verified 0700 directory, then publish it at the fixed path
// with link(2), which refuses to replace anything already there. Node's server.close() unlinks
// whatever sits at the path it bound (measured on Node 24), so binding the private name means
// close() can only ever remove that name, never a socket another process later put at the
// public path. The private name is removed as soon as the link exists.
const bindPath = join(paths.dir!, `.bind-${process.pid}-${randomBytes(6).toString("hex")}`);
assertSocketPathFits(bindPath);
// umask is process-wide; it is narrowed only around listen() and restored in `finally`. The
// directory is already verified 0700 and the bound socket is chmod-ed and re-checked below, so
// the umask is defense in depth, not the guarantee.
const previous = process.umask(0o177);
try {
await listen(server, bindPath);
} finally {
process.umask(previous);
}
const abort = async (e: unknown): Promise<never> => {
await new Promise<void>((r) => server.close(() => r()));
throw e;
};
try {
chmodSync(bindPath, 0o600);
checkSocket(bindPath, uid);
// link(2) keeps the inode, so the bound name's (dev, ino) is the published socket's. Recording
// it before the link means a crash anywhere after this point leaves a record that names the
// socket actually at the path, which is what the next start correlates against.
const bound = lstatSync(bindPath);
lock.publishing({ path: paths.socket, dev: bound.dev, ino: bound.ino });
linkSync(bindPath, paths.socket);
} catch (e) {
if ((e as NodeJS.ErrnoException).code === "EEXIST") return abort(new EndpointRefusal("ALREADY_RUNNING", `another process published ${paths.socket} first; nothing was replaced`));
return abort(e);
}
unlinkSync(bindPath);
try {
checkSocket(paths.socket, uid);
} catch (e) {
return abort(e);
}
const { dev: boundDev, ino: boundIno } = generationOf(paths.socket);
// Resume what this daemon home was connected to before it stopped. This runs after the endpoint
// is published, so a space that is slow or unreachable never delays readiness; `service-status`
return {
async close() {
await new Promise<void>((r) => server.close(() => r()));
// Remove the public socket only if it is still the exact file (dev + ino) this process
// created; a replacement of any kind is left in place.
try {
const st = lstatSync(paths.socket);
if (st.isSocket() && st.uid === uid && st.dev === boundDev && st.ino === boundIno) unlinkSync(paths.socket);
} catch (e) {
if ((e as NodeJS.ErrnoException).code !== "ENOENT") throw e;
}
},
};
}
/**
* Windows: serve the pipe through the native listener, which hands over only connections whose
* client process token it read; this keeps those whose user is this one. The name needs no cleanup:
* it ends when the last instance closes.
*/
function publishPipe(paths: DaemonPaths, counters: ServerCounters, log: (line: string) => void, accept: (sock: Socket) => void): Endpoint {
let listener: PipeListener;
try {
listener = listenPipe(paths.socket);
} catch (e) {
if ((e as NodeJS.ErrnoException).code === "EADDRINUSE") {
throw new EndpointRefusal("ALREADY_RUNNING", `${paths.socket} is already served by another process; nothing was replaced`,
"find the process holding that pipe (Sysinternals `handle -a cotal-daemon` names it) and stop it, then run `cotal daemon serve` again");
}
throw e;
}
const own = windowsUserSid();
let closing = false;
const accepting = (async () => {
for (;;) {
let peer: PipePeer | PipeRefused;
try {
peer = await listener.accept();
} catch (e) {
if (!closing) log(`daemon: the pipe listener stopped (${(e as Error).message}); it accepts no more connections`);
listener.close();
return;
}
if ("refused" in peer) {
counters.peerRejected++;
log(`daemon: pipe peer identity unreadable (${peer.refused}); dropped before any frame was read`);
continue;
}
if (peer.user !== own) {
counters.peerRejected++;
log(`daemon: pipe peer user ${peer.user} is not ${own}; dropped before any frame was read`);
closeSync(peer.fd);
continue;
}
let sock: Socket;
try {
sock = adoptPipe(peer.fd);
} catch (e) {
log(`daemon: could not adopt an authenticated pipe connection (${(e as Error).message}); dropped`);
continue;
}
accept(sock);
}
})();
return {
async close() {
closing = true;
listener.close();
await accepting;
},
};
}
export { EndpointRefusal };