feat(api): add terminal WebSocket gateway with PTY session management #515
@@ -66,6 +66,7 @@
|
||||
"marked-gfm-heading-id": "^4.1.3",
|
||||
"marked-highlight": "^2.2.3",
|
||||
"matrix-bot-sdk": "^0.8.0",
|
||||
"node-pty": "^1.0.0",
|
||||
"ollama": "^0.6.3",
|
||||
"openai": "^6.17.0",
|
||||
"reflect-metadata": "^0.2.2",
|
||||
|
||||
@@ -40,6 +40,7 @@ import { CredentialsModule } from "./credentials/credentials.module";
|
||||
import { MosaicTelemetryModule } from "./mosaic-telemetry";
|
||||
import { SpeechModule } from "./speech/speech.module";
|
||||
import { DashboardModule } from "./dashboard/dashboard.module";
|
||||
import { TerminalModule } from "./terminal/terminal.module";
|
||||
import { RlsContextInterceptor } from "./common/interceptors/rls-context.interceptor";
|
||||
|
||||
@Module({
|
||||
@@ -103,6 +104,7 @@ import { RlsContextInterceptor } from "./common/interceptors/rls-context.interce
|
||||
MosaicTelemetryModule,
|
||||
SpeechModule,
|
||||
DashboardModule,
|
||||
TerminalModule,
|
||||
],
|
||||
controllers: [AppController, CsrfController],
|
||||
providers: [
|
||||
|
||||
89
apps/api/src/terminal/terminal.dto.ts
Normal file
89
apps/api/src/terminal/terminal.dto.ts
Normal file
@@ -0,0 +1,89 @@
|
||||
/**
|
||||
* Terminal DTOs
|
||||
*
|
||||
* Data Transfer Objects for terminal WebSocket events.
|
||||
* Validated using class-validator decorators.
|
||||
*/
|
||||
|
||||
import {
|
||||
IsString,
|
||||
IsOptional,
|
||||
IsNumber,
|
||||
IsInt,
|
||||
Min,
|
||||
Max,
|
||||
MinLength,
|
||||
MaxLength,
|
||||
} from "class-validator";
|
||||
|
||||
/**
|
||||
* DTO for creating a new terminal PTY session.
|
||||
*/
|
||||
export class CreateTerminalDto {
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(128)
|
||||
name?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsInt()
|
||||
@Min(1)
|
||||
@Max(500)
|
||||
cols?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsInt()
|
||||
@Min(1)
|
||||
@Max(200)
|
||||
rows?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(4096)
|
||||
cwd?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* DTO for sending input data to a terminal PTY session.
|
||||
*/
|
||||
export class TerminalInputDto {
|
||||
@IsString()
|
||||
@MinLength(1)
|
||||
@MaxLength(64)
|
||||
sessionId!: string;
|
||||
|
||||
@IsString()
|
||||
data!: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* DTO for resizing a terminal PTY session.
|
||||
*/
|
||||
export class TerminalResizeDto {
|
||||
@IsString()
|
||||
@MinLength(1)
|
||||
@MaxLength(64)
|
||||
sessionId!: string;
|
||||
|
||||
@IsNumber()
|
||||
@IsInt()
|
||||
@Min(1)
|
||||
@Max(500)
|
||||
cols!: number;
|
||||
|
||||
@IsNumber()
|
||||
@IsInt()
|
||||
@Min(1)
|
||||
@Max(200)
|
||||
rows!: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* DTO for closing a terminal PTY session.
|
||||
*/
|
||||
export class CloseTerminalDto {
|
||||
@IsString()
|
||||
@MinLength(1)
|
||||
@MaxLength(64)
|
||||
sessionId!: string;
|
||||
}
|
||||
501
apps/api/src/terminal/terminal.gateway.spec.ts
Normal file
501
apps/api/src/terminal/terminal.gateway.spec.ts
Normal file
@@ -0,0 +1,501 @@
|
||||
/**
|
||||
* TerminalGateway Tests
|
||||
*
|
||||
* Unit tests for WebSocket terminal gateway:
|
||||
* - Authentication on connection
|
||||
* - terminal:create event handling
|
||||
* - terminal:input event handling
|
||||
* - terminal:resize event handling
|
||||
* - terminal:close event handling
|
||||
* - disconnect cleanup
|
||||
* - Error paths
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, vi, afterEach } from "vitest";
|
||||
import type { Socket } from "socket.io";
|
||||
import { TerminalGateway } from "./terminal.gateway";
|
||||
import { TerminalService } from "./terminal.service";
|
||||
import { AuthService } from "../auth/auth.service";
|
||||
import { PrismaService } from "../prisma/prisma.service";
|
||||
|
||||
// ==========================================
|
||||
// Mocks
|
||||
// ==========================================
|
||||
|
||||
// Mock node-pty globally so TerminalService doesn't fail to import
|
||||
vi.mock("node-pty", () => ({
|
||||
spawn: vi.fn(() => ({
|
||||
onData: vi.fn(),
|
||||
onExit: vi.fn(),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
kill: vi.fn(),
|
||||
pid: 1000,
|
||||
})),
|
||||
}));
|
||||
|
||||
interface AuthenticatedSocket extends Socket {
|
||||
data: {
|
||||
userId?: string;
|
||||
workspaceId?: string;
|
||||
};
|
||||
}
|
||||
|
||||
function createMockSocket(id = "test-socket-id"): AuthenticatedSocket {
|
||||
return {
|
||||
id,
|
||||
emit: vi.fn(),
|
||||
join: vi.fn(),
|
||||
leave: vi.fn(),
|
||||
disconnect: vi.fn(),
|
||||
data: {},
|
||||
handshake: {
|
||||
auth: { token: "valid-token" },
|
||||
query: {},
|
||||
headers: {},
|
||||
},
|
||||
} as unknown as AuthenticatedSocket;
|
||||
}
|
||||
|
||||
function createMockAuthService() {
|
||||
return {
|
||||
verifySession: vi.fn().mockResolvedValue({
|
||||
user: { id: "user-123" },
|
||||
session: { id: "session-123" },
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
function createMockPrismaService() {
|
||||
return {
|
||||
workspaceMember: {
|
||||
findFirst: vi.fn().mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function createMockTerminalService() {
|
||||
return {
|
||||
createSession: vi.fn().mockReturnValue({
|
||||
sessionId: "session-uuid-1",
|
||||
name: undefined,
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
}),
|
||||
writeToSession: vi.fn(),
|
||||
resizeSession: vi.fn(),
|
||||
closeSession: vi.fn().mockReturnValue(true),
|
||||
closeWorkspaceSessions: vi.fn(),
|
||||
sessionBelongsToWorkspace: vi.fn().mockReturnValue(true),
|
||||
getWorkspaceSessionCount: vi.fn().mockReturnValue(0),
|
||||
};
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
// Tests
|
||||
// ==========================================
|
||||
|
||||
describe("TerminalGateway", () => {
|
||||
let gateway: TerminalGateway;
|
||||
let mockAuthService: ReturnType<typeof createMockAuthService>;
|
||||
let mockPrismaService: ReturnType<typeof createMockPrismaService>;
|
||||
let mockTerminalService: ReturnType<typeof createMockTerminalService>;
|
||||
let mockClient: AuthenticatedSocket;
|
||||
|
||||
beforeEach(() => {
|
||||
mockAuthService = createMockAuthService();
|
||||
mockPrismaService = createMockPrismaService();
|
||||
mockTerminalService = createMockTerminalService();
|
||||
mockClient = createMockSocket();
|
||||
|
||||
gateway = new TerminalGateway(
|
||||
mockAuthService as unknown as AuthService,
|
||||
mockPrismaService as unknown as PrismaService,
|
||||
mockTerminalService as unknown as TerminalService
|
||||
);
|
||||
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// handleConnection (authentication)
|
||||
// ==========================================
|
||||
describe("handleConnection", () => {
|
||||
it("should authenticate client and join workspace room on valid token", async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue({
|
||||
user: { id: "user-123" },
|
||||
});
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
|
||||
await gateway.handleConnection(mockClient);
|
||||
|
||||
expect(mockAuthService.verifySession).toHaveBeenCalledWith("valid-token");
|
||||
expect(mockClient.data.userId).toBe("user-123");
|
||||
expect(mockClient.data.workspaceId).toBe("workspace-456");
|
||||
expect(mockClient.join).toHaveBeenCalledWith("terminal:workspace-456");
|
||||
});
|
||||
|
||||
it("should disconnect and emit error if no token provided", async () => {
|
||||
const clientNoToken = createMockSocket("no-token");
|
||||
clientNoToken.handshake = {
|
||||
auth: {},
|
||||
query: {},
|
||||
headers: {},
|
||||
} as typeof clientNoToken.handshake;
|
||||
|
||||
await gateway.handleConnection(clientNoToken);
|
||||
|
||||
expect(clientNoToken.disconnect).toHaveBeenCalled();
|
||||
expect(clientNoToken.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("no token") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should disconnect and emit error if token is invalid", async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue(null);
|
||||
|
||||
await gateway.handleConnection(mockClient);
|
||||
|
||||
expect(mockClient.disconnect).toHaveBeenCalled();
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("invalid") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should disconnect and emit error if no workspace access", async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue(null);
|
||||
|
||||
await gateway.handleConnection(mockClient);
|
||||
|
||||
expect(mockClient.disconnect).toHaveBeenCalled();
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("workspace") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should disconnect and emit error if auth throws", async () => {
|
||||
mockAuthService.verifySession.mockRejectedValue(new Error("Auth service down"));
|
||||
|
||||
await gateway.handleConnection(mockClient);
|
||||
|
||||
expect(mockClient.disconnect).toHaveBeenCalled();
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.any(String) })
|
||||
);
|
||||
});
|
||||
|
||||
it("should extract token from handshake.query as fallback", async () => {
|
||||
const clientQueryToken = createMockSocket("query-token-client");
|
||||
clientQueryToken.handshake = {
|
||||
auth: {},
|
||||
query: { token: "query-token" },
|
||||
headers: {},
|
||||
} as typeof clientQueryToken.handshake;
|
||||
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
|
||||
await gateway.handleConnection(clientQueryToken);
|
||||
|
||||
expect(mockAuthService.verifySession).toHaveBeenCalledWith("query-token");
|
||||
});
|
||||
|
||||
it("should extract token from Authorization header as last fallback", async () => {
|
||||
const clientHeaderToken = createMockSocket("header-token-client");
|
||||
clientHeaderToken.handshake = {
|
||||
auth: {},
|
||||
query: {},
|
||||
headers: { authorization: "Bearer header-token" },
|
||||
} as typeof clientHeaderToken.handshake;
|
||||
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
|
||||
await gateway.handleConnection(clientHeaderToken);
|
||||
|
||||
expect(mockAuthService.verifySession).toHaveBeenCalledWith("header-token");
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// handleDisconnect
|
||||
// ==========================================
|
||||
describe("handleDisconnect", () => {
|
||||
it("should close all workspace sessions on disconnect", async () => {
|
||||
await gateway.handleConnection(mockClient);
|
||||
vi.clearAllMocks();
|
||||
|
||||
gateway.handleDisconnect(mockClient);
|
||||
|
||||
expect(mockTerminalService.closeWorkspaceSessions).toHaveBeenCalledWith("workspace-456");
|
||||
});
|
||||
|
||||
it("should not throw for unauthenticated client disconnect", () => {
|
||||
const unauthClient = createMockSocket("unauth-disconnect");
|
||||
|
||||
expect(() => gateway.handleDisconnect(unauthClient)).not.toThrow();
|
||||
expect(mockTerminalService.closeWorkspaceSessions).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// handleCreate (terminal:create)
|
||||
// ==========================================
|
||||
describe("handleCreate", () => {
|
||||
beforeEach(async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
await gateway.handleConnection(mockClient);
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it("should create a PTY session and emit terminal:created", async () => {
|
||||
mockTerminalService.createSession.mockReturnValue({
|
||||
sessionId: "new-session-id",
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
});
|
||||
|
||||
await gateway.handleCreate(mockClient, {});
|
||||
|
||||
expect(mockTerminalService.createSession).toHaveBeenCalled();
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:created",
|
||||
expect.objectContaining({ sessionId: "new-session-id" })
|
||||
);
|
||||
});
|
||||
|
||||
it("should pass cols, rows, cwd, name to service", async () => {
|
||||
await gateway.handleCreate(mockClient, {
|
||||
cols: 132,
|
||||
rows: 50,
|
||||
cwd: "/home/user",
|
||||
name: "my-shell",
|
||||
});
|
||||
|
||||
expect(mockTerminalService.createSession).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
expect.objectContaining({ cols: 132, rows: 50, cwd: "/home/user", name: "my-shell" })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error if not authenticated", async () => {
|
||||
const unauthClient = createMockSocket("unauth");
|
||||
|
||||
await gateway.handleCreate(unauthClient, {});
|
||||
|
||||
expect(unauthClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("authenticated") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error if service throws (session limit)", async () => {
|
||||
mockTerminalService.createSession.mockImplementation(() => {
|
||||
throw new Error("Workspace has reached the maximum of 10 concurrent terminal sessions");
|
||||
});
|
||||
|
||||
await gateway.handleCreate(mockClient, {});
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("maximum") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error for invalid payload (negative cols)", async () => {
|
||||
await gateway.handleCreate(mockClient, { cols: -1 });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("Invalid payload") })
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// handleInput (terminal:input)
|
||||
// ==========================================
|
||||
describe("handleInput", () => {
|
||||
beforeEach(async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
await gateway.handleConnection(mockClient);
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it("should write data to the PTY session", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(true);
|
||||
|
||||
await gateway.handleInput(mockClient, { sessionId: "sess-1", data: "ls\n" });
|
||||
|
||||
expect(mockTerminalService.writeToSession).toHaveBeenCalledWith("sess-1", "ls\n");
|
||||
});
|
||||
|
||||
it("should emit terminal:error if session does not belong to workspace", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(false);
|
||||
|
||||
await gateway.handleInput(mockClient, { sessionId: "alien-sess", data: "data" });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("not found") })
|
||||
);
|
||||
expect(mockTerminalService.writeToSession).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should emit terminal:error if not authenticated", async () => {
|
||||
const unauthClient = createMockSocket("unauth");
|
||||
|
||||
await gateway.handleInput(unauthClient, { sessionId: "sess-1", data: "x" });
|
||||
|
||||
expect(unauthClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("authenticated") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error for invalid payload (missing sessionId)", async () => {
|
||||
await gateway.handleInput(mockClient, { data: "some input" });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("Invalid payload") })
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// handleResize (terminal:resize)
|
||||
// ==========================================
|
||||
describe("handleResize", () => {
|
||||
beforeEach(async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
await gateway.handleConnection(mockClient);
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it("should resize the PTY session", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(true);
|
||||
|
||||
await gateway.handleResize(mockClient, { sessionId: "sess-1", cols: 120, rows: 40 });
|
||||
|
||||
expect(mockTerminalService.resizeSession).toHaveBeenCalledWith("sess-1", 120, 40);
|
||||
});
|
||||
|
||||
it("should emit terminal:error if session does not belong to workspace", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(false);
|
||||
|
||||
await gateway.handleResize(mockClient, { sessionId: "alien-sess", cols: 80, rows: 24 });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("not found") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error for invalid payload (cols too large)", async () => {
|
||||
await gateway.handleResize(mockClient, { sessionId: "sess-1", cols: 9999, rows: 24 });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("Invalid payload") })
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// handleClose (terminal:close)
|
||||
// ==========================================
|
||||
describe("handleClose", () => {
|
||||
beforeEach(async () => {
|
||||
mockAuthService.verifySession.mockResolvedValue({ user: { id: "user-123" } });
|
||||
mockPrismaService.workspaceMember.findFirst.mockResolvedValue({
|
||||
userId: "user-123",
|
||||
workspaceId: "workspace-456",
|
||||
role: "MEMBER",
|
||||
});
|
||||
await gateway.handleConnection(mockClient);
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it("should close an existing PTY session", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(true);
|
||||
mockTerminalService.closeSession.mockReturnValue(true);
|
||||
|
||||
await gateway.handleClose(mockClient, { sessionId: "sess-1" });
|
||||
|
||||
expect(mockTerminalService.closeSession).toHaveBeenCalledWith("sess-1");
|
||||
});
|
||||
|
||||
it("should emit terminal:error if session does not belong to workspace", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(false);
|
||||
|
||||
await gateway.handleClose(mockClient, { sessionId: "alien-sess" });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("not found") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error if closeSession returns false (session gone)", async () => {
|
||||
mockTerminalService.sessionBelongsToWorkspace.mockReturnValue(true);
|
||||
mockTerminalService.closeSession.mockReturnValue(false);
|
||||
|
||||
await gateway.handleClose(mockClient, { sessionId: "gone-sess" });
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("not found") })
|
||||
);
|
||||
});
|
||||
|
||||
it("should emit terminal:error for invalid payload (missing sessionId)", async () => {
|
||||
await gateway.handleClose(mockClient, {});
|
||||
|
||||
expect(mockClient.emit).toHaveBeenCalledWith(
|
||||
"terminal:error",
|
||||
expect.objectContaining({ message: expect.stringContaining("Invalid payload") })
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
423
apps/api/src/terminal/terminal.gateway.ts
Normal file
423
apps/api/src/terminal/terminal.gateway.ts
Normal file
@@ -0,0 +1,423 @@
|
||||
/**
|
||||
* TerminalGateway
|
||||
*
|
||||
* WebSocket gateway for real-time PTY terminal sessions.
|
||||
* Uses the `/terminal` namespace to keep terminal traffic separate
|
||||
* from the main WebSocket gateway.
|
||||
*
|
||||
* Protocol:
|
||||
* 1. Client connects with auth token in handshake
|
||||
* 2. Client emits `terminal:create` to spawn a new PTY session
|
||||
* 3. Server emits `terminal:created` with { sessionId }
|
||||
* 4. Client emits `terminal:input` with { sessionId, data } to send keystrokes
|
||||
* 5. Server emits `terminal:output` with { sessionId, data } for stdout/stderr
|
||||
* 6. Client emits `terminal:resize` with { sessionId, cols, rows } on window resize
|
||||
* 7. Client emits `terminal:close` with { sessionId } to terminate the PTY
|
||||
* 8. Server emits `terminal:exit` with { sessionId, exitCode, signal } on PTY exit
|
||||
*
|
||||
* Authentication:
|
||||
* - Same pattern as websocket.gateway.ts and speech.gateway.ts
|
||||
* - Token extracted from handshake.auth.token / query.token / Authorization header
|
||||
*
|
||||
* Workspace isolation:
|
||||
* - Clients join room `terminal:{workspaceId}` on connect
|
||||
* - Sessions are scoped to workspace; cross-workspace access is denied
|
||||
*/
|
||||
|
||||
import {
|
||||
WebSocketGateway as WSGateway,
|
||||
WebSocketServer,
|
||||
SubscribeMessage,
|
||||
OnGatewayConnection,
|
||||
OnGatewayDisconnect,
|
||||
} from "@nestjs/websockets";
|
||||
import { Logger } from "@nestjs/common";
|
||||
import { Server, Socket } from "socket.io";
|
||||
import { AuthService } from "../auth/auth.service";
|
||||
import { PrismaService } from "../prisma/prisma.service";
|
||||
import { TerminalService } from "./terminal.service";
|
||||
import {
|
||||
CreateTerminalDto,
|
||||
TerminalInputDto,
|
||||
TerminalResizeDto,
|
||||
CloseTerminalDto,
|
||||
} from "./terminal.dto";
|
||||
import { validate } from "class-validator";
|
||||
import { plainToInstance } from "class-transformer";
|
||||
|
||||
// ==========================================
|
||||
// Types
|
||||
// ==========================================
|
||||
|
||||
interface AuthenticatedSocket extends Socket {
|
||||
data: {
|
||||
userId?: string;
|
||||
workspaceId?: string;
|
||||
};
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
// Gateway
|
||||
// ==========================================
|
||||
|
||||
@WSGateway({
|
||||
namespace: "/terminal",
|
||||
cors: {
|
||||
origin: process.env.WEB_URL ?? "http://localhost:3000",
|
||||
credentials: true,
|
||||
},
|
||||
})
|
||||
export class TerminalGateway implements OnGatewayConnection, OnGatewayDisconnect {
|
||||
@WebSocketServer()
|
||||
server!: Server;
|
||||
|
||||
private readonly logger = new Logger(TerminalGateway.name);
|
||||
private readonly CONNECTION_TIMEOUT_MS = 5000;
|
||||
|
||||
constructor(
|
||||
private readonly authService: AuthService,
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly terminalService: TerminalService
|
||||
) {}
|
||||
|
||||
// ==========================================
|
||||
// Connection lifecycle
|
||||
// ==========================================
|
||||
|
||||
/**
|
||||
* Authenticate client on connection using handshake token.
|
||||
* Validates workspace membership and joins the workspace-scoped room.
|
||||
*/
|
||||
async handleConnection(client: Socket): Promise<void> {
|
||||
const authenticatedClient = client as AuthenticatedSocket;
|
||||
|
||||
const timeoutId = setTimeout(() => {
|
||||
if (!authenticatedClient.data.userId) {
|
||||
this.logger.warn(
|
||||
`Terminal client ${authenticatedClient.id} timed out during authentication`
|
||||
);
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Authentication timed out.",
|
||||
});
|
||||
authenticatedClient.disconnect();
|
||||
}
|
||||
}, this.CONNECTION_TIMEOUT_MS);
|
||||
|
||||
try {
|
||||
const token = this.extractTokenFromHandshake(authenticatedClient);
|
||||
|
||||
if (!token) {
|
||||
this.logger.warn(`Terminal client ${authenticatedClient.id} connected without token`);
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Authentication failed: no token provided.",
|
||||
});
|
||||
authenticatedClient.disconnect();
|
||||
clearTimeout(timeoutId);
|
||||
return;
|
||||
}
|
||||
|
||||
const sessionData = await this.authService.verifySession(token);
|
||||
|
||||
if (!sessionData) {
|
||||
this.logger.warn(`Terminal client ${authenticatedClient.id} has invalid token`);
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Authentication failed: invalid or expired token.",
|
||||
});
|
||||
authenticatedClient.disconnect();
|
||||
clearTimeout(timeoutId);
|
||||
return;
|
||||
}
|
||||
|
||||
const user = sessionData.user as { id: string };
|
||||
const userId = user.id;
|
||||
|
||||
const workspaceMembership = await this.prisma.workspaceMember.findFirst({
|
||||
where: { userId },
|
||||
select: { workspaceId: true, userId: true, role: true },
|
||||
});
|
||||
|
||||
if (!workspaceMembership) {
|
||||
this.logger.warn(`Terminal user ${userId} has no workspace access`);
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Authentication failed: no workspace access.",
|
||||
});
|
||||
authenticatedClient.disconnect();
|
||||
clearTimeout(timeoutId);
|
||||
return;
|
||||
}
|
||||
|
||||
authenticatedClient.data.userId = userId;
|
||||
authenticatedClient.data.workspaceId = workspaceMembership.workspaceId;
|
||||
|
||||
// Join workspace-scoped terminal room
|
||||
const room = this.getWorkspaceRoom(workspaceMembership.workspaceId);
|
||||
await authenticatedClient.join(room);
|
||||
|
||||
clearTimeout(timeoutId);
|
||||
this.logger.log(
|
||||
`Terminal client ${authenticatedClient.id} connected (user: ${userId}, workspace: ${workspaceMembership.workspaceId})`
|
||||
);
|
||||
} catch (error) {
|
||||
clearTimeout(timeoutId);
|
||||
this.logger.error(
|
||||
`Authentication failed for terminal client ${authenticatedClient.id}:`,
|
||||
error instanceof Error ? error.message : "Unknown error"
|
||||
);
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Authentication failed: an unexpected error occurred.",
|
||||
});
|
||||
authenticatedClient.disconnect();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up all PTY sessions for this client's workspace on disconnect.
|
||||
*/
|
||||
handleDisconnect(client: Socket): void {
|
||||
const authenticatedClient = client as AuthenticatedSocket;
|
||||
const { workspaceId, userId } = authenticatedClient.data;
|
||||
|
||||
if (workspaceId) {
|
||||
this.terminalService.closeWorkspaceSessions(workspaceId);
|
||||
|
||||
const room = this.getWorkspaceRoom(workspaceId);
|
||||
void authenticatedClient.leave(room);
|
||||
this.logger.log(
|
||||
`Terminal client ${authenticatedClient.id} disconnected (user: ${userId ?? "unknown"}, workspace: ${workspaceId})`
|
||||
);
|
||||
} else {
|
||||
this.logger.debug(`Terminal client ${authenticatedClient.id} disconnected (unauthenticated)`);
|
||||
}
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
// Terminal events
|
||||
// ==========================================
|
||||
|
||||
/**
|
||||
* Spawn a new PTY session for the connected client.
|
||||
*
|
||||
* Emits `terminal:created` with { sessionId, name, cols, rows } on success.
|
||||
* Emits `terminal:error` on failure.
|
||||
*/
|
||||
@SubscribeMessage("terminal:create")
|
||||
async handleCreate(client: Socket, payload: unknown): Promise<void> {
|
||||
const authenticatedClient = client as AuthenticatedSocket;
|
||||
const { userId, workspaceId } = authenticatedClient.data;
|
||||
|
||||
if (!userId || !workspaceId) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Not authenticated. Connect with a valid token.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Validate DTO
|
||||
const dto = plainToInstance(CreateTerminalDto, payload ?? {});
|
||||
const errors = await validate(dto);
|
||||
if (errors.length > 0) {
|
||||
const messages = errors.map((e) => Object.values(e.constraints ?? {}).join(", ")).join("; ");
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Invalid payload: ${messages}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const result = this.terminalService.createSession(authenticatedClient, {
|
||||
workspaceId,
|
||||
socketId: authenticatedClient.id,
|
||||
...(dto.name !== undefined ? { name: dto.name } : {}),
|
||||
...(dto.cols !== undefined ? { cols: dto.cols } : {}),
|
||||
...(dto.rows !== undefined ? { rows: dto.rows } : {}),
|
||||
...(dto.cwd !== undefined ? { cwd: dto.cwd } : {}),
|
||||
});
|
||||
|
||||
authenticatedClient.emit("terminal:created", {
|
||||
sessionId: result.sessionId,
|
||||
name: result.name,
|
||||
cols: result.cols,
|
||||
rows: result.rows,
|
||||
});
|
||||
|
||||
this.logger.log(
|
||||
`Terminal session ${result.sessionId} created for client ${authenticatedClient.id} (workspace: ${workspaceId})`
|
||||
);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error(
|
||||
`Failed to create terminal session for client ${authenticatedClient.id}: ${message}`
|
||||
);
|
||||
authenticatedClient.emit("terminal:error", { message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Write input data to an existing PTY session.
|
||||
*
|
||||
* Emits `terminal:error` if the session is not found or unauthorized.
|
||||
*/
|
||||
@SubscribeMessage("terminal:input")
|
||||
async handleInput(client: Socket, payload: unknown): Promise<void> {
|
||||
const authenticatedClient = client as AuthenticatedSocket;
|
||||
const { userId, workspaceId } = authenticatedClient.data;
|
||||
|
||||
if (!userId || !workspaceId) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Not authenticated. Connect with a valid token.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const dto = plainToInstance(TerminalInputDto, payload ?? {});
|
||||
const errors = await validate(dto);
|
||||
if (errors.length > 0) {
|
||||
const messages = errors.map((e) => Object.values(e.constraints ?? {}).join(", ")).join("; ");
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Invalid payload: ${messages}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (!this.terminalService.sessionBelongsToWorkspace(dto.sessionId, workspaceId)) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Terminal session ${dto.sessionId} not found or unauthorized.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
this.terminalService.writeToSession(dto.sessionId, dto.data);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.warn(`Failed to write to terminal session ${dto.sessionId}: ${message}`);
|
||||
authenticatedClient.emit("terminal:error", { message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resize an existing PTY session.
|
||||
*
|
||||
* Emits `terminal:error` if the session is not found or unauthorized.
|
||||
*/
|
||||
@SubscribeMessage("terminal:resize")
|
||||
async handleResize(client: Socket, payload: unknown): Promise<void> {
|
||||
const authenticatedClient = client as AuthenticatedSocket;
|
||||
const { userId, workspaceId } = authenticatedClient.data;
|
||||
|
||||
if (!userId || !workspaceId) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Not authenticated. Connect with a valid token.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const dto = plainToInstance(TerminalResizeDto, payload ?? {});
|
||||
const errors = await validate(dto);
|
||||
if (errors.length > 0) {
|
||||
const messages = errors.map((e) => Object.values(e.constraints ?? {}).join(", ")).join("; ");
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Invalid payload: ${messages}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (!this.terminalService.sessionBelongsToWorkspace(dto.sessionId, workspaceId)) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Terminal session ${dto.sessionId} not found or unauthorized.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
this.terminalService.resizeSession(dto.sessionId, dto.cols, dto.rows);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.warn(`Failed to resize terminal session ${dto.sessionId}: ${message}`);
|
||||
authenticatedClient.emit("terminal:error", { message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Kill and close an existing PTY session.
|
||||
*
|
||||
* Emits `terminal:error` if the session is not found or unauthorized.
|
||||
*/
|
||||
@SubscribeMessage("terminal:close")
|
||||
async handleClose(client: Socket, payload: unknown): Promise<void> {
|
||||
const authenticatedClient = client as AuthenticatedSocket;
|
||||
const { userId, workspaceId } = authenticatedClient.data;
|
||||
|
||||
if (!userId || !workspaceId) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: "Not authenticated. Connect with a valid token.",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const dto = plainToInstance(CloseTerminalDto, payload ?? {});
|
||||
const errors = await validate(dto);
|
||||
if (errors.length > 0) {
|
||||
const messages = errors.map((e) => Object.values(e.constraints ?? {}).join(", ")).join("; ");
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Invalid payload: ${messages}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (!this.terminalService.sessionBelongsToWorkspace(dto.sessionId, workspaceId)) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Terminal session ${dto.sessionId} not found or unauthorized.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const closed = this.terminalService.closeSession(dto.sessionId);
|
||||
if (!closed) {
|
||||
authenticatedClient.emit("terminal:error", {
|
||||
message: `Terminal session ${dto.sessionId} not found.`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.log(`Terminal session ${dto.sessionId} closed by client ${authenticatedClient.id}`);
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
// Private helpers
|
||||
// ==========================================
|
||||
|
||||
/**
|
||||
* Extract authentication token from Socket.IO handshake.
|
||||
* Checks auth.token, query.token, and Authorization header (in that order).
|
||||
*/
|
||||
private extractTokenFromHandshake(client: Socket): string | undefined {
|
||||
const authToken = client.handshake.auth.token as unknown;
|
||||
if (typeof authToken === "string" && authToken.length > 0) {
|
||||
return authToken;
|
||||
}
|
||||
|
||||
const queryToken = client.handshake.query.token as unknown;
|
||||
if (typeof queryToken === "string" && queryToken.length > 0) {
|
||||
return queryToken;
|
||||
}
|
||||
|
||||
const authHeader = client.handshake.headers.authorization as unknown;
|
||||
if (typeof authHeader === "string") {
|
||||
const parts = authHeader.split(" ");
|
||||
const [type, token] = parts;
|
||||
if (type === "Bearer" && token) {
|
||||
return token;
|
||||
}
|
||||
}
|
||||
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the workspace-scoped room name for the terminal namespace.
|
||||
*/
|
||||
private getWorkspaceRoom(workspaceId: string): string {
|
||||
return `terminal:${workspaceId}`;
|
||||
}
|
||||
}
|
||||
28
apps/api/src/terminal/terminal.module.ts
Normal file
28
apps/api/src/terminal/terminal.module.ts
Normal file
@@ -0,0 +1,28 @@
|
||||
/**
|
||||
* TerminalModule
|
||||
*
|
||||
* NestJS module for WebSocket-based terminal sessions via node-pty.
|
||||
*
|
||||
* Imports:
|
||||
* - AuthModule for WebSocket authentication (verifySession)
|
||||
* - PrismaModule for workspace membership queries
|
||||
*
|
||||
* Providers:
|
||||
* - TerminalService: manages PTY session lifecycle
|
||||
* - TerminalGateway: WebSocket gateway on /terminal namespace
|
||||
*
|
||||
* The module does not export providers; terminal sessions are
|
||||
* self-contained within this module.
|
||||
*/
|
||||
|
||||
import { Module } from "@nestjs/common";
|
||||
import { TerminalGateway } from "./terminal.gateway";
|
||||
import { TerminalService } from "./terminal.service";
|
||||
import { AuthModule } from "../auth/auth.module";
|
||||
import { PrismaModule } from "../prisma/prisma.module";
|
||||
|
||||
@Module({
|
||||
imports: [AuthModule, PrismaModule],
|
||||
providers: [TerminalGateway, TerminalService],
|
||||
})
|
||||
export class TerminalModule {}
|
||||
337
apps/api/src/terminal/terminal.service.spec.ts
Normal file
337
apps/api/src/terminal/terminal.service.spec.ts
Normal file
@@ -0,0 +1,337 @@
|
||||
/**
|
||||
* TerminalService Tests
|
||||
*
|
||||
* Unit tests for PTY session management: create, write, resize, close,
|
||||
* workspace cleanup, and access control.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, vi, afterEach } from "vitest";
|
||||
import type { Socket } from "socket.io";
|
||||
import { TerminalService, MAX_SESSIONS_PER_WORKSPACE } from "./terminal.service";
|
||||
|
||||
// ==========================================
|
||||
// Mocks
|
||||
// ==========================================
|
||||
|
||||
// Mock node-pty before importing service
|
||||
const mockPtyProcess = {
|
||||
onData: vi.fn(),
|
||||
onExit: vi.fn(),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
kill: vi.fn(),
|
||||
pid: 12345,
|
||||
};
|
||||
|
||||
vi.mock("node-pty", () => ({
|
||||
spawn: vi.fn(() => mockPtyProcess),
|
||||
}));
|
||||
|
||||
function createMockSocket(id = "socket-1"): Socket {
|
||||
return {
|
||||
id,
|
||||
emit: vi.fn(),
|
||||
join: vi.fn(),
|
||||
leave: vi.fn(),
|
||||
disconnect: vi.fn(),
|
||||
data: {},
|
||||
} as unknown as Socket;
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
// Tests
|
||||
// ==========================================
|
||||
|
||||
describe("TerminalService", () => {
|
||||
let service: TerminalService;
|
||||
let mockSocket: Socket;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
// Reset mock implementations
|
||||
mockPtyProcess.onData.mockImplementation((_cb: (data: string) => void) => {});
|
||||
mockPtyProcess.onExit.mockImplementation(
|
||||
(_cb: (e: { exitCode: number; signal?: number }) => void) => {}
|
||||
);
|
||||
service = new TerminalService();
|
||||
mockSocket = createMockSocket();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// createSession
|
||||
// ==========================================
|
||||
describe("createSession", () => {
|
||||
it("should create a PTY session and return sessionId", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(result.sessionId).toBeDefined();
|
||||
expect(typeof result.sessionId).toBe("string");
|
||||
expect(result.cols).toBe(80);
|
||||
expect(result.rows).toBe(24);
|
||||
});
|
||||
|
||||
it("should use provided cols and rows", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
cols: 120,
|
||||
rows: 40,
|
||||
});
|
||||
|
||||
expect(result.cols).toBe(120);
|
||||
expect(result.rows).toBe(40);
|
||||
});
|
||||
|
||||
it("should return the provided session name", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
name: "my-terminal",
|
||||
});
|
||||
|
||||
expect(result.name).toBe("my-terminal");
|
||||
});
|
||||
|
||||
it("should wire PTY onData to emit terminal:output", () => {
|
||||
let dataCallback: ((data: string) => void) | undefined;
|
||||
mockPtyProcess.onData.mockImplementation((cb: (data: string) => void) => {
|
||||
dataCallback = cb;
|
||||
});
|
||||
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(dataCallback).toBeDefined();
|
||||
dataCallback!("hello world");
|
||||
|
||||
expect(mockSocket.emit).toHaveBeenCalledWith("terminal:output", {
|
||||
sessionId: result.sessionId,
|
||||
data: "hello world",
|
||||
});
|
||||
});
|
||||
|
||||
it("should wire PTY onExit to emit terminal:exit and cleanup", () => {
|
||||
let exitCallback: ((e: { exitCode: number; signal?: number }) => void) | undefined;
|
||||
mockPtyProcess.onExit.mockImplementation(
|
||||
(cb: (e: { exitCode: number; signal?: number }) => void) => {
|
||||
exitCallback = cb;
|
||||
}
|
||||
);
|
||||
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(exitCallback).toBeDefined();
|
||||
exitCallback!({ exitCode: 0 });
|
||||
|
||||
expect(mockSocket.emit).toHaveBeenCalledWith("terminal:exit", {
|
||||
sessionId: result.sessionId,
|
||||
exitCode: 0,
|
||||
signal: undefined,
|
||||
});
|
||||
|
||||
// Session should be cleaned up
|
||||
expect(service.sessionBelongsToWorkspace(result.sessionId, "ws-1")).toBe(false);
|
||||
expect(service.getWorkspaceSessionCount("ws-1")).toBe(0);
|
||||
});
|
||||
|
||||
it("should throw when workspace session limit is reached", () => {
|
||||
const limit = MAX_SESSIONS_PER_WORKSPACE;
|
||||
|
||||
for (let i = 0; i < limit; i++) {
|
||||
service.createSession(createMockSocket(`socket-${String(i)}`), {
|
||||
workspaceId: "ws-limit",
|
||||
socketId: `socket-${String(i)}`,
|
||||
});
|
||||
}
|
||||
|
||||
expect(() =>
|
||||
service.createSession(createMockSocket("socket-overflow"), {
|
||||
workspaceId: "ws-limit",
|
||||
socketId: "socket-overflow",
|
||||
})
|
||||
).toThrow(/maximum/i);
|
||||
});
|
||||
|
||||
it("should allow sessions in different workspaces independently", () => {
|
||||
service.createSession(mockSocket, { workspaceId: "ws-a", socketId: "s1" });
|
||||
service.createSession(createMockSocket("s2"), { workspaceId: "ws-b", socketId: "s2" });
|
||||
|
||||
expect(service.getWorkspaceSessionCount("ws-a")).toBe(1);
|
||||
expect(service.getWorkspaceSessionCount("ws-b")).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// writeToSession
|
||||
// ==========================================
|
||||
describe("writeToSession", () => {
|
||||
it("should write data to PTY", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
service.writeToSession(result.sessionId, "ls -la\n");
|
||||
|
||||
expect(mockPtyProcess.write).toHaveBeenCalledWith("ls -la\n");
|
||||
});
|
||||
|
||||
it("should throw for unknown sessionId", () => {
|
||||
expect(() => service.writeToSession("nonexistent-id", "data")).toThrow(/not found/i);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// resizeSession
|
||||
// ==========================================
|
||||
describe("resizeSession", () => {
|
||||
it("should resize PTY dimensions", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
service.resizeSession(result.sessionId, 132, 50);
|
||||
|
||||
expect(mockPtyProcess.resize).toHaveBeenCalledWith(132, 50);
|
||||
});
|
||||
|
||||
it("should throw for unknown sessionId", () => {
|
||||
expect(() => service.resizeSession("nonexistent-id", 80, 24)).toThrow(/not found/i);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// closeSession
|
||||
// ==========================================
|
||||
describe("closeSession", () => {
|
||||
it("should kill PTY and return true for existing session", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
const closed = service.closeSession(result.sessionId);
|
||||
|
||||
expect(closed).toBe(true);
|
||||
expect(mockPtyProcess.kill).toHaveBeenCalled();
|
||||
expect(service.sessionBelongsToWorkspace(result.sessionId, "ws-1")).toBe(false);
|
||||
});
|
||||
|
||||
it("should return false for nonexistent sessionId", () => {
|
||||
const closed = service.closeSession("does-not-exist");
|
||||
expect(closed).toBe(false);
|
||||
});
|
||||
|
||||
it("should clean up workspace tracking after close", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(service.getWorkspaceSessionCount("ws-1")).toBe(1);
|
||||
service.closeSession(result.sessionId);
|
||||
expect(service.getWorkspaceSessionCount("ws-1")).toBe(0);
|
||||
});
|
||||
|
||||
it("should not throw if PTY kill throws", () => {
|
||||
mockPtyProcess.kill.mockImplementationOnce(() => {
|
||||
throw new Error("PTY already dead");
|
||||
});
|
||||
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(() => service.closeSession(result.sessionId)).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// closeWorkspaceSessions
|
||||
// ==========================================
|
||||
describe("closeWorkspaceSessions", () => {
|
||||
it("should kill all sessions for a workspace", () => {
|
||||
service.createSession(mockSocket, { workspaceId: "ws-1", socketId: "s1" });
|
||||
service.createSession(createMockSocket("s2"), { workspaceId: "ws-1", socketId: "s2" });
|
||||
|
||||
expect(service.getWorkspaceSessionCount("ws-1")).toBe(2);
|
||||
|
||||
service.closeWorkspaceSessions("ws-1");
|
||||
|
||||
expect(service.getWorkspaceSessionCount("ws-1")).toBe(0);
|
||||
expect(mockPtyProcess.kill).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("should not affect sessions in other workspaces", () => {
|
||||
service.createSession(mockSocket, { workspaceId: "ws-1", socketId: "s1" });
|
||||
service.createSession(createMockSocket("s2"), { workspaceId: "ws-2", socketId: "s2" });
|
||||
|
||||
service.closeWorkspaceSessions("ws-1");
|
||||
|
||||
expect(service.getWorkspaceSessionCount("ws-1")).toBe(0);
|
||||
expect(service.getWorkspaceSessionCount("ws-2")).toBe(1);
|
||||
});
|
||||
|
||||
it("should not throw for workspaces with no sessions", () => {
|
||||
expect(() => service.closeWorkspaceSessions("ws-nonexistent")).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// sessionBelongsToWorkspace
|
||||
// ==========================================
|
||||
describe("sessionBelongsToWorkspace", () => {
|
||||
it("should return true for a session belonging to the workspace", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(service.sessionBelongsToWorkspace(result.sessionId, "ws-1")).toBe(true);
|
||||
});
|
||||
|
||||
it("should return false for a session in a different workspace", () => {
|
||||
const result = service.createSession(mockSocket, {
|
||||
workspaceId: "ws-1",
|
||||
socketId: "socket-1",
|
||||
});
|
||||
|
||||
expect(service.sessionBelongsToWorkspace(result.sessionId, "ws-2")).toBe(false);
|
||||
});
|
||||
|
||||
it("should return false for a nonexistent sessionId", () => {
|
||||
expect(service.sessionBelongsToWorkspace("no-such-id", "ws-1")).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
// ==========================================
|
||||
// getWorkspaceSessionCount
|
||||
// ==========================================
|
||||
describe("getWorkspaceSessionCount", () => {
|
||||
it("should return 0 for workspace with no sessions", () => {
|
||||
expect(service.getWorkspaceSessionCount("empty-ws")).toBe(0);
|
||||
});
|
||||
|
||||
it("should track session count accurately", () => {
|
||||
service.createSession(mockSocket, { workspaceId: "ws-count", socketId: "s1" });
|
||||
expect(service.getWorkspaceSessionCount("ws-count")).toBe(1);
|
||||
|
||||
service.createSession(createMockSocket("s2"), { workspaceId: "ws-count", socketId: "s2" });
|
||||
expect(service.getWorkspaceSessionCount("ws-count")).toBe(2);
|
||||
});
|
||||
});
|
||||
});
|
||||
251
apps/api/src/terminal/terminal.service.ts
Normal file
251
apps/api/src/terminal/terminal.service.ts
Normal file
@@ -0,0 +1,251 @@
|
||||
/**
|
||||
* TerminalService
|
||||
*
|
||||
* Manages PTY (pseudo-terminal) sessions for workspace users.
|
||||
* Spawns real shell processes via node-pty, streams I/O to connected sockets,
|
||||
* and enforces per-workspace session limits.
|
||||
*
|
||||
* Session lifecycle:
|
||||
* - createSession: spawn a new PTY, wire onData/onExit, return sessionId
|
||||
* - writeToSession: send input data to PTY stdin
|
||||
* - resizeSession: resize PTY dimensions (cols x rows)
|
||||
* - closeSession: kill PTY process, emit terminal:exit, cleanup
|
||||
* - closeWorkspaceSessions: kill all sessions for a workspace (on disconnect)
|
||||
*/
|
||||
|
||||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import * as pty from "node-pty";
|
||||
import type { Socket } from "socket.io";
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
||||
/** Maximum concurrent PTY sessions per workspace */
|
||||
export const MAX_SESSIONS_PER_WORKSPACE = parseInt(
|
||||
process.env.TERMINAL_MAX_SESSIONS_PER_WORKSPACE ?? "10",
|
||||
10
|
||||
);
|
||||
|
||||
/** Default PTY dimensions */
|
||||
const DEFAULT_COLS = 80;
|
||||
const DEFAULT_ROWS = 24;
|
||||
|
||||
export interface TerminalSession {
|
||||
sessionId: string;
|
||||
workspaceId: string;
|
||||
pty: pty.IPty;
|
||||
name?: string;
|
||||
createdAt: Date;
|
||||
}
|
||||
|
||||
export interface CreateSessionOptions {
|
||||
name?: string;
|
||||
cols?: number;
|
||||
rows?: number;
|
||||
cwd?: string;
|
||||
workspaceId: string;
|
||||
socketId: string;
|
||||
}
|
||||
|
||||
export interface SessionCreatedResult {
|
||||
sessionId: string;
|
||||
name?: string;
|
||||
cols: number;
|
||||
rows: number;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class TerminalService {
|
||||
private readonly logger = new Logger(TerminalService.name);
|
||||
|
||||
/**
|
||||
* Map of sessionId -> TerminalSession
|
||||
*/
|
||||
private readonly sessions = new Map<string, TerminalSession>();
|
||||
|
||||
/**
|
||||
* Map of workspaceId -> Set<sessionId> for fast per-workspace lookups
|
||||
*/
|
||||
private readonly workspaceSessions = new Map<string, Set<string>>();
|
||||
|
||||
/**
|
||||
* Create a new PTY session for the given workspace and socket.
|
||||
* Wires PTY onData -> emit terminal:output and onExit -> emit terminal:exit.
|
||||
*
|
||||
* @throws Error if workspace session limit is exceeded
|
||||
*/
|
||||
createSession(socket: Socket, options: CreateSessionOptions): SessionCreatedResult {
|
||||
const { workspaceId, name, cwd, socketId } = options;
|
||||
const cols = options.cols ?? DEFAULT_COLS;
|
||||
const rows = options.rows ?? DEFAULT_ROWS;
|
||||
|
||||
// Enforce per-workspace session limit
|
||||
const workspaceSessionIds = this.workspaceSessions.get(workspaceId) ?? new Set<string>();
|
||||
if (workspaceSessionIds.size >= MAX_SESSIONS_PER_WORKSPACE) {
|
||||
throw new Error(
|
||||
`Workspace ${workspaceId} has reached the maximum of ${String(MAX_SESSIONS_PER_WORKSPACE)} concurrent terminal sessions`
|
||||
);
|
||||
}
|
||||
|
||||
const sessionId = randomUUID();
|
||||
const shell = process.env.SHELL ?? "/bin/bash";
|
||||
|
||||
this.logger.log(
|
||||
`Spawning PTY session ${sessionId} for workspace ${workspaceId} (socket: ${socketId}, shell: ${shell}, ${String(cols)}x${String(rows)})`
|
||||
);
|
||||
|
||||
const ptyProcess = pty.spawn(shell, [], {
|
||||
name: "xterm-256color",
|
||||
cols,
|
||||
rows,
|
||||
cwd: cwd ?? process.cwd(),
|
||||
env: process.env as Record<string, string>,
|
||||
});
|
||||
|
||||
const session: TerminalSession = {
|
||||
sessionId,
|
||||
workspaceId,
|
||||
pty: ptyProcess,
|
||||
...(name !== undefined ? { name } : {}),
|
||||
createdAt: new Date(),
|
||||
};
|
||||
|
||||
this.sessions.set(sessionId, session);
|
||||
|
||||
// Track by workspace
|
||||
if (!this.workspaceSessions.has(workspaceId)) {
|
||||
this.workspaceSessions.set(workspaceId, new Set());
|
||||
}
|
||||
const wsSet = this.workspaceSessions.get(workspaceId);
|
||||
if (wsSet) {
|
||||
wsSet.add(sessionId);
|
||||
}
|
||||
|
||||
// Wire PTY stdout/stderr -> terminal:output
|
||||
ptyProcess.onData((data: string) => {
|
||||
socket.emit("terminal:output", { sessionId, data });
|
||||
});
|
||||
|
||||
// Wire PTY exit -> terminal:exit, cleanup
|
||||
ptyProcess.onExit(({ exitCode, signal }) => {
|
||||
this.logger.log(
|
||||
`PTY session ${sessionId} exited (exitCode: ${String(exitCode)}, signal: ${String(signal ?? "none")})`
|
||||
);
|
||||
socket.emit("terminal:exit", { sessionId, exitCode, signal });
|
||||
this.cleanupSession(sessionId, workspaceId);
|
||||
});
|
||||
|
||||
return { sessionId, ...(name !== undefined ? { name } : {}), cols, rows };
|
||||
}
|
||||
|
||||
/**
|
||||
* Write input data to a PTY session's stdin.
|
||||
*
|
||||
* @throws Error if session not found
|
||||
*/
|
||||
writeToSession(sessionId: string, data: string): void {
|
||||
const session = this.sessions.get(sessionId);
|
||||
if (!session) {
|
||||
throw new Error(`Terminal session ${sessionId} not found`);
|
||||
}
|
||||
session.pty.write(data);
|
||||
}
|
||||
|
||||
/**
|
||||
* Resize a PTY session's terminal dimensions.
|
||||
*
|
||||
* @throws Error if session not found
|
||||
*/
|
||||
resizeSession(sessionId: string, cols: number, rows: number): void {
|
||||
const session = this.sessions.get(sessionId);
|
||||
if (!session) {
|
||||
throw new Error(`Terminal session ${sessionId} not found`);
|
||||
}
|
||||
session.pty.resize(cols, rows);
|
||||
this.logger.debug(`Resized PTY session ${sessionId} to ${String(cols)}x${String(rows)}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Kill and clean up a specific PTY session.
|
||||
* Returns true if the session existed, false if it was already gone.
|
||||
*/
|
||||
closeSession(sessionId: string): boolean {
|
||||
const session = this.sessions.get(sessionId);
|
||||
if (!session) {
|
||||
return false;
|
||||
}
|
||||
|
||||
this.logger.log(`Closing PTY session ${sessionId} for workspace ${session.workspaceId}`);
|
||||
|
||||
try {
|
||||
session.pty.kill();
|
||||
} catch (error) {
|
||||
this.logger.warn(
|
||||
`Error killing PTY session ${sessionId}: ${error instanceof Error ? error.message : String(error)}`
|
||||
);
|
||||
}
|
||||
|
||||
this.cleanupSession(sessionId, session.workspaceId);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Close all PTY sessions for a workspace (called on client disconnect).
|
||||
*/
|
||||
closeWorkspaceSessions(workspaceId: string): void {
|
||||
const sessionIds = this.workspaceSessions.get(workspaceId);
|
||||
if (!sessionIds || sessionIds.size === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Closing ${String(sessionIds.size)} PTY session(s) for workspace ${workspaceId} (disconnect)`
|
||||
);
|
||||
|
||||
// Copy to array to avoid mutation during iteration
|
||||
const ids = Array.from(sessionIds);
|
||||
for (const sessionId of ids) {
|
||||
const session = this.sessions.get(sessionId);
|
||||
if (session) {
|
||||
try {
|
||||
session.pty.kill();
|
||||
} catch (error) {
|
||||
this.logger.warn(
|
||||
`Error killing PTY session ${sessionId} on disconnect: ${error instanceof Error ? error.message : String(error)}`
|
||||
);
|
||||
}
|
||||
this.cleanupSession(sessionId, workspaceId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of active sessions for a workspace.
|
||||
*/
|
||||
getWorkspaceSessionCount(workspaceId: string): number {
|
||||
return this.workspaceSessions.get(workspaceId)?.size ?? 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a session belongs to a given workspace.
|
||||
* Used for access control in the gateway.
|
||||
*/
|
||||
sessionBelongsToWorkspace(sessionId: string, workspaceId: string): boolean {
|
||||
const session = this.sessions.get(sessionId);
|
||||
return session?.workspaceId === workspaceId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal cleanup: remove session from tracking maps.
|
||||
* Does NOT kill the PTY (caller is responsible).
|
||||
*/
|
||||
private cleanupSession(sessionId: string, workspaceId: string): void {
|
||||
this.sessions.delete(sessionId);
|
||||
|
||||
const workspaceSessionIds = this.workspaceSessions.get(workspaceId);
|
||||
if (workspaceSessionIds) {
|
||||
workspaceSessionIds.delete(sessionId);
|
||||
if (workspaceSessionIds.size === 0) {
|
||||
this.workspaceSessions.delete(workspaceId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3,13 +3,14 @@ packages:
|
||||
- packages/*
|
||||
|
||||
ignoredBuiltDependencies:
|
||||
- '@nestjs/core'
|
||||
- '@swc/core'
|
||||
- "@nestjs/core"
|
||||
- "@swc/core"
|
||||
- better-sqlite3
|
||||
- esbuild
|
||||
- sharp
|
||||
|
||||
onlyBuiltDependencies:
|
||||
- '@prisma/client'
|
||||
- '@prisma/engines'
|
||||
- "@prisma/client"
|
||||
- "@prisma/engines"
|
||||
- prisma
|
||||
- node-pty
|
||||
|
||||
Reference in New Issue
Block a user