feat: wire daemon console/files/config/players and improve runtime fallbacks

2026-02-22 12:09:07 +00:00
parent 614d25c189
commit 44c439e2f9
12 changed files with 1793 additions and 96 deletions
@@ -0,0 +1,259 @@
+import fp from 'fastify-plugin';
+import type { FastifyInstance } from 'fastify';
+import { and, eq } from 'drizzle-orm';
+import { Server as SocketIOServer } from 'socket.io';
+import { nodes, organizationMembers, servers } from '@source/database';
+import { ROLES } from '@source/shared';
+import type { Role } from '@source/shared';
+import type { AccessTokenPayload } from '../lib/jwt.js';
+import {
+  daemonOpenConsoleStream,
+  daemonSendCommand,
+  type DaemonConsoleStreamHandle,
+  type DaemonNodeConnection,
+} from '../lib/daemon.js';
+
+declare module 'fastify' {
+  interface FastifyInstance {
+    io: SocketIOServer;
+  }
+}
+
+type ConsolePermission = 'console.read' | 'console.write';
+
+export default fp(async (app: FastifyInstance) => {
+  const io = new SocketIOServer(app.server, {
+    path: '/socket.io',
+    cors: {
+      origin: process.env.CORS_ORIGIN || 'http://localhost:5173',
+      credentials: true,
+    },
+  });
+
+  app.decorate('io', io);
+
+  const activeStreams = new Map<string, DaemonConsoleStreamHandle>();
+
+  io.use((socket, next) => {
+    const token = typeof socket.handshake.auth?.token === 'string'
+      ? socket.handshake.auth.token
+      : null;
+
+    if (!token) {
+      next(new Error('Unauthorized'));
+      return;
+    }
+
+    const verifier = (app as any).jwt?.verify;
+    if (typeof verifier !== 'function') {
+      next(new Error('Authentication is not configured'));
+      return;
+    }
+
+    try {
+      const payload = verifier(token) as AccessTokenPayload;
+      (socket.data as { user?: AccessTokenPayload }).user = payload;
+      next();
+    } catch {
+      next(new Error('Unauthorized'));
+    }
+  });
+
+  io.on('connection', (socket) => {
+    const cleanupSocketStream = () => {
+      const current = activeStreams.get(socket.id);
+      if (!current) return;
+      current.close();
+      activeStreams.delete(socket.id);
+    };
+
+    socket.on('server:console:join', async (payload: unknown) => {
+      const serverId = typeof (payload as { serverId?: unknown })?.serverId === 'string'
+        ? ((payload as { serverId: string }).serverId)
+        : '';
+      if (!serverId) {
+        socket.emit('server:console:output', { line: '[error] Invalid server id' });
+        return;
+      }
+
+      const user = (socket.data as { user?: AccessTokenPayload }).user;
+      if (!user) {
+        socket.emit('server:console:output', { line: '[error] Unauthorized' });
+        return;
+      }
+
+      const server = await getServerContext(app, serverId);
+      if (!server) {
+        socket.emit('server:console:output', { line: '[error] Server not found' });
+        return;
+      }
+
+      const allowed = await hasConsolePermission(app, user, server.organizationId, 'console.read');
+      if (!allowed) {
+        socket.emit('server:console:output', { line: '[error] Missing permission: console.read' });
+        return;
+      }
+
+      cleanupSocketStream();
+
+      try {
+        const streamHandle = await daemonOpenConsoleStream(server.node, server.serverUuid);
+        streamHandle.stream.on('data', (output) => {
+          socket.emit('server:console:output', { line: output.line });
+        });
+        streamHandle.stream.on('end', () => {
+          activeStreams.delete(socket.id);
+          socket.emit('server:console:output', { line: '[console] Stream ended' });
+        });
+        streamHandle.stream.on('error', (error) => {
+          activeStreams.delete(socket.id);
+          app.log.warn(
+            { error, serverId, serverUuid: server.serverUuid, socketId: socket.id },
+            'Console stream failed',
+          );
+          socket.emit('server:console:output', { line: '[error] Console stream failed' });
+        });
+
+        activeStreams.set(socket.id, streamHandle);
+      } catch (error) {
+        app.log.warn(
+          { error, serverId, serverUuid: server.serverUuid, socketId: socket.id },
+          'Failed to open console stream',
+        );
+        socket.emit('server:console:output', { line: '[error] Failed to open console stream' });
+      }
+    });
+
+    socket.on('server:console:leave', () => {
+      cleanupSocketStream();
+    });
+
+    socket.on('server:console:command', async (payload: unknown) => {
+      const body = payload as {
+        serverId?: unknown;
+        orgId?: unknown;
+        command?: unknown;
+      };
+
+      const serverId = typeof body.serverId === 'string' ? body.serverId : '';
+      const orgId = typeof body.orgId === 'string' ? body.orgId : '';
+      const command = typeof body.command === 'string' ? body.command.trim() : '';
+
+      if (!serverId || !orgId || !command) {
+        socket.emit('server:console:output', { line: '[error] Invalid command payload' });
+        return;
+      }
+
+      const user = (socket.data as { user?: AccessTokenPayload }).user;
+      if (!user) {
+        socket.emit('server:console:output', { line: '[error] Unauthorized' });
+        return;
+      }
+
+      const server = await getServerContext(app, serverId, orgId);
+      if (!server) {
+        socket.emit('server:console:output', { line: '[error] Server not found' });
+        return;
+      }
+
+      const allowed = await hasConsolePermission(app, user, orgId, 'console.write');
+      if (!allowed) {
+        socket.emit('server:console:output', { line: '[error] Missing permission: console.write' });
+        return;
+      }
+
+      try {
+        await daemonSendCommand(server.node, server.serverUuid, command);
+      } catch (error) {
+        app.log.warn(
+          { error, serverId, serverUuid: server.serverUuid, socketId: socket.id },
+          'Failed to send console command',
+        );
+        socket.emit('server:console:output', { line: '[error] Failed to send command' });
+      }
+    });
+
+    socket.on('disconnect', () => {
+      cleanupSocketStream();
+    });
+  });
+
+  app.addHook('onClose', async () => {
+    for (const handle of activeStreams.values()) {
+      handle.close();
+    }
+    activeStreams.clear();
+
+    await new Promise<void>((resolve) => {
+      io.close(() => resolve());
+    });
+  });
+});
+
+async function hasConsolePermission(
+  app: FastifyInstance,
+  user: AccessTokenPayload,
+  orgId: string,
+  permission: ConsolePermission,
+): Promise<boolean> {
+  if (user.isSuperAdmin) return true;
+
+  const member = await app.db.query.organizationMembers.findFirst({
+    where: and(
+      eq(organizationMembers.organizationId, orgId),
+      eq(organizationMembers.userId, user.sub),
+    ),
+    columns: {
+      role: true,
+      customPermissions: true,
+    },
+  });
+
+  if (!member) return false;
+
+  const custom = (member.customPermissions ?? {}) as Record<string, boolean>;
+  if (permission in custom) {
+    return Boolean(custom[permission]);
+  }
+
+  const rolePerms = ROLES[member.role as Role]?.permissions ?? [];
+  return (rolePerms as readonly string[]).includes(permission);
+}
+
+async function getServerContext(
+  app: FastifyInstance,
+  serverId: string,
+  orgId?: string,
+): Promise<{
+  organizationId: string;
+  serverUuid: string;
+  node: DaemonNodeConnection;
+} | null> {
+  const whereClause = orgId
+    ? and(eq(servers.id, serverId), eq(servers.organizationId, orgId))
+    : eq(servers.id, serverId);
+
+  const [row] = await app.db
+    .select({
+      organizationId: servers.organizationId,
+      serverUuid: servers.uuid,
+      nodeFqdn: nodes.fqdn,
+      nodeGrpcPort: nodes.grpcPort,
+      nodeDaemonToken: nodes.daemonToken,
+    })
+    .from(servers)
+    .innerJoin(nodes, eq(servers.nodeId, nodes.id))
+    .where(whereClause);
+
+  if (!row) return null;
+
+  return {
+    organizationId: row.organizationId,
+    serverUuid: row.serverUuid,
+    node: {
+      fqdn: row.nodeFqdn,
+      grpcPort: row.nodeGrpcPort,
+      daemonToken: row.nodeDaemonToken,
+    },
+  };
+}