Repository Public
1/** `buildServer(config, deps)`: the whole server as a testable factory (no listen, no process state). */2import fs from 'node:fs';3import Fastify, { type FastifyError, type FastifyInstance, type FastifyRequest } from 'fastify';4import cookie from '@fastify/cookie';5import helmet from '@fastify/helmet';6import rateLimit from '@fastify/rate-limit';7import websocket from '@fastify/websocket';8import type { DestinationStream } from 'pino';9import type { ServerConfig } from './config.js';10import { createDb, type Db } from './db.js';11import { Audit } from './audit.js';12import { EventHub } from './events-hub.js';13import { HttpError, type Ctx, type Plugin, type RateClass, type RouteEntry } from './http.js';14import { startJanitor } from './janitor.js';15import { LiveRegistry } from './live.js';16import { RuntimeClient, RuntimeError } from './runtime-client.js';17import { sha256Hex } from './security/crypto.js';18import { registerGuard } from './security/guard.js';19import { findSessionForRequest } from './security/sessions.js';20import { ensureSetupToken } from './security/setup-token.js';21import { registerSecurityHeaders, registerStatic } from './static.js';22import authPlugin from './modules/auth/index.js';23import clusterPlugin from './modules/cluster/index.js';24import { AgentdSource, ClusterService, type ClusterSource } from './modules/cluster/service.js';25import filesPlugin from './modules/files/index.js';26import notificationsPlugin from './modules/notifications/index.js';27import { Notifier } from './modules/notifications/service.js';28import openapiPlugin from './modules/openapi.js';29import projectsPlugin from './modules/projects/index.js';30import sessionsPlugin from './modules/sessions/index.js';31import { SessionService } from './modules/sessions/service.js';32import settingsPlugin from './modules/settings/index.js';33import skillsMcpPlugin from './modules/skills-mcp/index.js';34import { HttpSkillsRegistry, type SkillsRegistry } from './modules/skills-mcp/registry.js';35import systemPlugin from './modules/system/index.js';36import tasksPlugin from './modules/tasks/index.js';37import { TaskService } from './modules/tasks/service.js';38import terminalWsPlugin from './modules/terminal-ws.js';3940export const JSON_BODY_LIMIT = 6 * 1024 * 1024;41export const WS_MAX_FRAME = 1024 * 1024;4243/** Never logged: credentials, tokens, and any body field that could carry one. */44export const REDACT_PATHS = [45 'req.headers.cookie', 'req.headers.authorization', 'req.headers["x-csrf-token"]', 'headers.cookie', 'headers.authorization', 'headers["x-csrf-token"]',46 ...['password', 'currentPassword', 'newPassword', 'token', 'setupToken', 'csrfToken', 'secret', 'totp', 'code', 'command', 'content', 'env', 'headers.*'].flatMap((k) => [k, `*.${k}`, `body.${k}`, `*.body.${k}`]),47];4849export interface BuildDeps {50 db?: Db;51 fetch?: typeof fetch;52 /** undefined ⇒ AgentdSource when CR_AGENTD_URL is set; null ⇒ none. */53 clusterSource?: ClusterSource | null;54 skillsRegistry?: SkillsRegistry | null;55 now?: () => number;56 logStream?: DestinationStream;57}58export interface Server {59 app: FastifyInstance;60 ctx: Ctx;61 /** Runtime `/events` subscription (+ periodic reconciler, cluster poll, janitor when `timers`). */62 startBackground(opts?: { timers?: boolean }): void;63 /** One reconciler pass over sessions and tasks; false when the runtime is unreachable. */64 reconcile(): Promise<boolean>;65 close(): Promise<void>;66}6768export async function buildServer(config: ServerConfig, deps: BuildDeps = {}): Promise<Server> {69 const app = Fastify({70 logger: { level: config.logLevel, redact: { paths: REDACT_PATHS, censor: '[redacted]' }, ...(deps.logStream ? { stream: deps.logStream } : {}) },71 trustProxy: config.trustedProxies, // X-Forwarded-For is honoured ONLY from these peers72 bodyLimit: JSON_BODY_LIMIT,73 routerOptions: { maxParamLength: 200 },74 });75 const db = deps.db ?? createDb(config.databaseUrl);76 const rt = new RuntimeClient(config.runtimeSocket, config.runtimeToken);77 const hub = new EventHub();78 const fetchImpl = deps.fetch ?? fetch;79 const now = deps.now ?? Date.now;80 const notifier = new Notifier(db, hub);81 const log = app.log;82 const source = deps.clusterSource !== undefined ? deps.clusterSource : config.agentdUrl ? new AgentdSource(config.agentdUrl, config.agentdToken, fetchImpl) : null;83 const ctx: Ctx = {84 config, db, rt, log, hub, notifier, now,85 live: new LiveRegistry(),86 audit: new Audit(db, log),87 sessions: new SessionService({ db, rt, hub, notifier, log, agent: config.agent }),88 tasks: new TaskService({ db, rt, hub, notifier, log }),89 cluster: new ClusterService({ config, db, rt, log, source, now }),90 skillsRegistry: deps.skillsRegistry !== undefined ? deps.skillsRegistry : config.skillsRegistryUrl ? new HttpSkillsRegistry(config.skillsRegistryUrl, fetchImpl) : null,91 routes: [],92 fetch: fetchImpl,93 };9495 app.addHook('onRoute', (r) => {96 const methods = Array.isArray(r.method) ? r.method : [r.method];97 const cfg = (r.config ?? {}) as { auth?: RouteEntry['auth']; doc?: RouteEntry['doc'] };98 for (const method of methods) {99 ctx.routes.push({ method, url: r.url, auth: cfg.auth ?? 'full', doc: cfg.doc ?? null, websocket: (r as { websocket?: boolean }).websocket === true });100 }101 });102103 app.setErrorHandler((error: FastifyError | HttpError | RuntimeError, req, reply) => {104 let status = 500;105 let code = 'internal';106 let message = 'internal error';107 let details: unknown;108 if (error instanceof HttpError) {109 ({ status, code, message, details } = error);110 } else if (error instanceof RuntimeError) {111 status = error.unreachable ? 503 : 502;112 code = error.unreachable ? 'runtime_unavailable' : 'runtime_error';113 message = error.unreachable ? 'the runtime daemon is not reachable' : 'the runtime daemon reported an error';114 } else if (typeof error.statusCode === 'number' && error.statusCode >= 400 && error.statusCode < 500) {115 status = error.statusCode;116 code = status === 429 ? 'rate_limited' : status === 413 ? 'payload_too_large' : status === 415 ? 'unsupported_media_type' : 'bad_request';117 message = status === 429 ? 'too many requests' : error.message;118 }119 if (status >= 500) req.log.error({ err: error }, 'request failed');120 const retryAfterSec = (details as { retryAfterSec?: unknown } | undefined)?.retryAfterSec;121 return reply.code(status).send({122 error: { code, message, ...(details !== undefined ? { details } : {}) },123 // Login lockouts also expose the wait in the `LoginStep` shape the web client reads.124 ...(typeof retryAfterSec === 'number' ? { ok: false, next: 'done', retryAfterSec } : {}),125 });126 });127128 await app.register(cookie);129 await app.register(helmet, {130 contentSecurityPolicy: false, // set by static.ts (per-document script hashes)131 hsts: { maxAge: 63072000, includeSubDomains: true, preload: false },132 referrerPolicy: { policy: 'no-referrer' },133 frameguard: { action: 'deny' },134 crossOriginEmbedderPolicy: false,135 crossOriginResourcePolicy: { policy: 'same-origin' },136 });137 registerSecurityHeaders(app, config);138 // Buckets (see `RateClass`): credential checks are strict per client IP; the read-only139 // `GET /auth/status` and `GET /health` have their own generous per-IP buckets, so page loads140 // and health checks can neither exhaust nor be starved by the login budget; everything else141 // is limited per session cookie.142 const rateClassOf = (req: FastifyRequest): RateClass => {143 const cfg = req.routeOptions.config;144 return cfg.rate ?? (cfg.auth === 'preauth' ? 'credentials' : cfg.auth === 'public' ? 'health' : 'api');145 };146 await app.register(rateLimit, {147 global: true,148 timeWindow: 60_000,149 allowList: (req) => !req.url.startsWith('/api/'),150 keyGenerator: async (req) => {151 const cls = rateClassOf(req);152 if (cls === 'credentials') return `ip:pre:${req.ip}`;153 if (cls === 'health') return `ip:health:${req.ip}`;154 const sid = req.cookies[config.cookieName];155 if (cls === 'status') {156 // Only a real session moves the caller to its own bucket: a made-up cookie stays per IP.157 const row = sid ? await findSessionForRequest(db, req, sid).catch(() => null) : null;158 return row && sid ? `sid:${sha256Hex(sid)}` : `ip:status:${req.ip}`;159 }160 return sid ? `sid:${sha256Hex(sid)}` : `ip:api:${req.ip}`;161 },162 max: (req, key) => {163 const cls = rateClassOf(req);164 if (cls === 'credentials') return config.rateLimit.preAuth;165 if (cls === 'health') return config.rateLimit.health;166 if (cls === 'status' && !key.startsWith('sid:')) return config.rateLimit.status;167 return config.rateLimit.api;168 },169 });170 await app.register(websocket, { options: { maxPayload: WS_MAX_FRAME, perMessageDeflate: false } });171 registerGuard(app, ctx);172 await registerStatic(app, config);173174 const plugins: Plugin[] = [175 systemPlugin, authPlugin, projectsPlugin, sessionsPlugin, terminalWsPlugin, filesPlugin, tasksPlugin,176 clusterPlugin, skillsMcpPlugin, notificationsPlugin, settingsPlugin, openapiPlugin,177 ];178 await app.register(async (api) => {179 for (const plugin of plugins) await plugin(api, ctx);180 }, { prefix: '/api/v1' });181182 fs.mkdirSync(config.dataDir, { recursive: true, mode: 0o700 });183 await ensureSetupToken(db, config.dataDir);184 await app.ready();185186 let events: { close(): void } | null = null;187 const stops: (() => void)[] = [];188 const reconcile = async (): Promise<boolean> => {189 const up = await ctx.sessions.reconcile();190 if (up) await ctx.tasks.reconcile();191 return up;192 };193 const safeReconcile = (): void => void reconcile().catch((err: unknown) => log.error({ err }, 'reconcile failed'));194195 return {196 app,197 ctx,198 reconcile,199 startBackground(opts = {}) {200 if (events) return;201 events = rt.subscribeEvents(202 (ev) => {203 const run = ev.type === 'task.updated' ? ctx.tasks.onRuntimeEvent(ev) : ctx.sessions.onRuntimeEvent(ev);204 run.catch((err: unknown) => log.error({ err, type: ev.type }, 'runtime event handling failed'));205 },206 (connected) => {207 if (connected) safeReconcile(); // catch up on whatever happened while the link was down208 },209 );210 if (opts.timers !== false) {211 safeReconcile();212 const t = setInterval(safeReconcile, config.reconcileIntervalMs);213 t.unref();214 stops.push(() => clearInterval(t), startJanitor(db, log));215 ctx.cluster.start();216 }217 },218 async close() {219 events?.close();220 for (const stop of stops) stop();221 ctx.cluster.stop();222 ctx.sessions.stop();223 ctx.live.closeAll(1001, 'server shutting down');224 await app.close();225 await rt.close();226 if (!deps.db) await db.end();227 },228 };229}230