Initial commit: WebSocket channel plugin for OpenClaw
Features: - API key authentication (X-API-Key header or query param) - Multi-tenant support with per-account configuration - Configurable capabilities (text, images, files, voice, reactions, threads, typing) - DM security policies (allowlist, open, closed) - Text-based pairing flow - Threading support - Handshake protocol with capabilities advertisement - CLI commands (websocket status, websocket accounts) - Full test suite
This commit is contained in:
root
68
src/auth.ts
Normal file
68
src/auth.ts
Normal file
@@ -0,0 +1,68 @@
|
||||
/**
|
||||
* Authentication utilities for WebSocket connections
|
||||
*/
|
||||
|
||||
import type { WebSocketAccount } from "./config.ts";
|
||||
|
||||
export type AuthResult =
|
||||
| { success: true; account: WebSocketAccount }
|
||||
| { success: false; error: string };
|
||||
|
||||
/**
|
||||
* Extract API key from WebSocket upgrade request
|
||||
* Supports X-API-Key header or ?key= query parameter
|
||||
*/
|
||||
export function extractApiKey(
|
||||
req: { headers: Record<string, string | string[]>; url?: string }
|
||||
): string | null {
|
||||
// Check header first
|
||||
const headerValue = req.headers["x-api-key"];
|
||||
if (headerValue) {
|
||||
return Array.isArray(headerValue) ? headerValue[0] : headerValue;
|
||||
}
|
||||
|
||||
// Check query parameter
|
||||
if (req.url) {
|
||||
const url = new URL(req.url, "http://localhost");
|
||||
const keyParam = url.searchParams.get("key");
|
||||
if (keyParam) {
|
||||
return keyParam;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate API key against configured accounts
|
||||
*/
|
||||
export function validateApiKey(
|
||||
apiKey: string,
|
||||
accounts: WebSocketAccount[]
|
||||
): AuthResult {
|
||||
if (!apiKey) {
|
||||
return { success: false, error: "API key is required" };
|
||||
}
|
||||
|
||||
const account = accounts.find((acc) => acc.apiKey === apiKey);
|
||||
if (!account) {
|
||||
return { success: false, error: "Invalid API key" };
|
||||
}
|
||||
|
||||
return { success: true, account };
|
||||
}
|
||||
|
||||
/**
|
||||
* Authenticate a connection request
|
||||
*/
|
||||
export function authenticateConnection(
|
||||
req: { headers: Record<string, string | string[]>; url?: string },
|
||||
accounts: WebSocketAccount[]
|
||||
): AuthResult {
|
||||
const apiKey = extractApiKey(req);
|
||||
if (!apiKey) {
|
||||
return { success: false, error: "Missing API key. Provide via X-API-Key header or ?key= query parameter" };
|
||||
}
|
||||
|
||||
return validateApiKey(apiKey, accounts);
|
||||
}
|
||||
339
src/channel.test.ts
Normal file
339
src/channel.test.ts
Normal file
@@ -0,0 +1,339 @@
|
||||
/**
|
||||
* WebSocket Channel Plugin Tests
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, vi } from "vitest";
|
||||
import { parseCapabilities, defaultCapabilities, type ChannelCapabilities } from "./config.ts";
|
||||
import { extractApiKey, validateApiKey, authenticateConnection } from "./auth.ts";
|
||||
import { WebSocketServerManager } from "./websocket-server.ts";
|
||||
|
||||
// Mock WebSocket
|
||||
class MockWebSocket {
|
||||
readyState = 1; // WebSocket.OPEN
|
||||
listeners: Record<string, ((...args: any[]) => void)[]> = {};
|
||||
|
||||
on(event: string, handler: (...args: any[]) => void) {
|
||||
if (!this.listeners[event]) this.listeners[event] = [];
|
||||
this.listeners[event].push(handler);
|
||||
}
|
||||
|
||||
send(data: string) {
|
||||
// Mock send
|
||||
}
|
||||
|
||||
close(code?: number, reason?: string) {
|
||||
this.readyState = 3; // WebSocket.CLOSED
|
||||
}
|
||||
|
||||
terminate() {
|
||||
this.readyState = 3;
|
||||
}
|
||||
|
||||
ping() {
|
||||
// Mock ping
|
||||
}
|
||||
|
||||
emit(event: string, ...args: any[]) {
|
||||
this.listeners[event]?.forEach((h) => h(...args));
|
||||
}
|
||||
}
|
||||
|
||||
describe("config", () => {
|
||||
describe("parseCapabilities", () => {
|
||||
it("returns defaults when given undefined", () => {
|
||||
const caps = parseCapabilities(undefined);
|
||||
expect(caps).toEqual(defaultCapabilities);
|
||||
});
|
||||
|
||||
it("merges partial capabilities with defaults", () => {
|
||||
const caps = parseCapabilities({ images: true, voice: true });
|
||||
expect(caps.text).toBe(true); // default
|
||||
expect(caps.images).toBe(true); // overridden
|
||||
expect(caps.voice).toBe(true); // overridden
|
||||
expect(caps.files).toBe(false); // default
|
||||
});
|
||||
|
||||
it("returns all false when explicitly set", () => {
|
||||
const caps: ChannelCapabilities = {
|
||||
text: false,
|
||||
images: false,
|
||||
files: false,
|
||||
voice: false,
|
||||
reactions: false,
|
||||
threads: false,
|
||||
typing: false,
|
||||
};
|
||||
const parsed = parseCapabilities(caps);
|
||||
expect(parsed.text).toBe(false);
|
||||
expect(parsed.images).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("auth", () => {
|
||||
const mockAccounts = [
|
||||
{
|
||||
accountId: "test-account",
|
||||
apiKey: "test-key-123",
|
||||
allowFrom: ["user1", "user2"],
|
||||
dmPolicy: "allowlist" as const,
|
||||
capabilities: defaultCapabilities,
|
||||
},
|
||||
{
|
||||
accountId: "prod-account",
|
||||
apiKey: "prod-key-456",
|
||||
allowFrom: [],
|
||||
dmPolicy: "open" as const,
|
||||
capabilities: { ...defaultCapabilities, images: true },
|
||||
},
|
||||
];
|
||||
|
||||
describe("extractApiKey", () => {
|
||||
it("extracts API key from header", () => {
|
||||
const req = {
|
||||
headers: { "x-api-key": "my-api-key" },
|
||||
};
|
||||
expect(extractApiKey(req)).toBe("my-api-key");
|
||||
});
|
||||
|
||||
it("extracts first API key from array header", () => {
|
||||
const req = {
|
||||
headers: { "x-api-key": ["key1", "key2"] },
|
||||
};
|
||||
expect(extractApiKey(req)).toBe("key1");
|
||||
});
|
||||
|
||||
it("extracts API key from query parameter", () => {
|
||||
const req = {
|
||||
headers: {},
|
||||
url: "/ws?key=query-key",
|
||||
};
|
||||
expect(extractApiKey(req)).toBe("query-key");
|
||||
});
|
||||
|
||||
it("returns null when no API key provided", () => {
|
||||
const req = {
|
||||
headers: {},
|
||||
url: "/ws",
|
||||
};
|
||||
expect(extractApiKey(req)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("validateApiKey", () => {
|
||||
it("returns success for valid API key", () => {
|
||||
const result = validateApiKey("test-key-123", mockAccounts);
|
||||
expect(result.success).toBe(true);
|
||||
if (result.success) {
|
||||
expect(result.account.accountId).toBe("test-account");
|
||||
}
|
||||
});
|
||||
|
||||
it("returns error for invalid API key", () => {
|
||||
const result = validateApiKey("invalid-key", mockAccounts);
|
||||
expect(result.success).toBe(false);
|
||||
if (!result.success) {
|
||||
expect(result.error).toBe("Invalid API key");
|
||||
}
|
||||
});
|
||||
|
||||
it("returns error for empty API key", () => {
|
||||
const result = validateApiKey("", mockAccounts);
|
||||
expect(result.success).toBe(false);
|
||||
if (!result.success) {
|
||||
expect(result.error).toBe("API key is required");
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("authenticateConnection", () => {
|
||||
it("authenticates with valid header", () => {
|
||||
const req = {
|
||||
headers: { "x-api-key": "test-key-123" },
|
||||
};
|
||||
const result = authenticateConnection(req, mockAccounts);
|
||||
expect(result.success).toBe(true);
|
||||
});
|
||||
|
||||
it("fails with missing API key", () => {
|
||||
const req = {
|
||||
headers: {},
|
||||
url: "/ws",
|
||||
};
|
||||
const result = authenticateConnection(req, mockAccounts);
|
||||
expect(result.success).toBe(false);
|
||||
if (!result.success) {
|
||||
expect(result.error).toContain("Missing API key");
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("WebSocketServerManager", () => {
|
||||
let manager: WebSocketServerManager;
|
||||
let mockWs: MockWebSocket;
|
||||
|
||||
const mockAccount = {
|
||||
accountId: "test-account",
|
||||
apiKey: "test-key",
|
||||
allowFrom: ["user1"],
|
||||
dmPolicy: "allowlist" as const,
|
||||
capabilities: defaultCapabilities,
|
||||
metadata: { serverName: "Test Server" },
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
manager = new WebSocketServerManager();
|
||||
mockWs = new MockWebSocket();
|
||||
});
|
||||
|
||||
describe("addClient", () => {
|
||||
it("adds a client and sends handshake", () => {
|
||||
const sentMessages: string[] = [];
|
||||
mockWs.send = (data: string) => sentMessages.push(data);
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
|
||||
expect(sentMessages.length).toBe(1);
|
||||
const handshake = JSON.parse(sentMessages[0]);
|
||||
expect(handshake.type).toBe("handshake");
|
||||
expect(handshake.accountId).toBe("test-account");
|
||||
expect(handshake.capabilities.text).toBe(true);
|
||||
expect(handshake.metadata.serverName).toBe("Test Server");
|
||||
});
|
||||
|
||||
it("tracks connection count", () => {
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
expect(manager.getConnectionCount("test-account")).toBe(1);
|
||||
});
|
||||
|
||||
it("tracks multiple connections", () => {
|
||||
const ws2 = new MockWebSocket();
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
manager.addClient(ws2 as any, mockAccount);
|
||||
expect(manager.getConnectionCount("test-account")).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("sendText", () => {
|
||||
it("sends text message to all connections for account", async () => {
|
||||
const sentMessages: string[] = [];
|
||||
mockWs.send = (data: string) => sentMessages.push(data);
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
await manager.sendText("user1", "Hello!", mockAccount);
|
||||
|
||||
expect(sentMessages.length).toBe(2); // handshake + message
|
||||
const message = JSON.parse(sentMessages[1]);
|
||||
expect(message.type).toBe("message");
|
||||
expect(message.to).toBe("user1");
|
||||
expect(message.content.type).toBe("text");
|
||||
expect(message.content.text).toBe("Hello!");
|
||||
expect(message.messageId).toBeDefined();
|
||||
});
|
||||
|
||||
it("throws when no connections exist", async () => {
|
||||
await expect(
|
||||
manager.sendText("user1", "Hello!", mockAccount)
|
||||
).rejects.toThrow("No active connections");
|
||||
});
|
||||
});
|
||||
|
||||
describe("sendMedia", () => {
|
||||
it("sends file message", async () => {
|
||||
const sentMessages: string[] = [];
|
||||
mockWs.send = (data: string) => sentMessages.push(data);
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
await manager.sendMedia("user1", "/path/to/file.pdf", mockAccount);
|
||||
|
||||
const message = JSON.parse(sentMessages[1]);
|
||||
expect(message.content.type).toBe("file");
|
||||
expect(message.content.name).toBe("file.pdf");
|
||||
});
|
||||
});
|
||||
|
||||
describe("sendTyping", () => {
|
||||
it("sends typing indicator", async () => {
|
||||
const sentMessages: string[] = [];
|
||||
mockWs.send = (data: string) => sentMessages.push(data);
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
await manager.sendTyping("user1", true, mockAccount);
|
||||
|
||||
const message = JSON.parse(sentMessages[1]);
|
||||
expect(message.type).toBe("typing");
|
||||
expect(message.to).toBe("user1");
|
||||
expect(message.isTyping).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("sendPairingCode", () => {
|
||||
it("sends pairing code message", async () => {
|
||||
const sentMessages: string[] = [];
|
||||
mockWs.send = (data: string) => sentMessages.push(data);
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
await manager.sendPairingCode("user1", "ABC123", mockAccount);
|
||||
|
||||
const message = JSON.parse(sentMessages[1]);
|
||||
expect(message.type).toBe("pairing_code");
|
||||
expect(message.code).toBe("ABC123");
|
||||
expect(message.target).toBe("user1");
|
||||
});
|
||||
});
|
||||
|
||||
describe("cleanup", () => {
|
||||
it("removes stale connections", () => {
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
expect(manager.getConnectionCount("test-account")).toBe(1);
|
||||
|
||||
// Simulate old ping
|
||||
const connections = manager.getConnections();
|
||||
connections[0].lastPingAt = new Date(Date.now() - 120000); // 2 minutes ago
|
||||
|
||||
manager.cleanup();
|
||||
expect(manager.getConnectionCount("test-account")).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("onMessage", () => {
|
||||
it("receives and processes inbound messages", async () => {
|
||||
const receivedMessages: any[] = [];
|
||||
manager.onMessage(async (msg) => {
|
||||
receivedMessages.push(msg);
|
||||
});
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
|
||||
// Simulate incoming message
|
||||
mockWs.emit("message", Buffer.from(JSON.stringify({
|
||||
type: "message",
|
||||
from: "user1",
|
||||
content: { type: "text", text: "Hello bot!" },
|
||||
})));
|
||||
|
||||
// Wait for async processing
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
|
||||
expect(receivedMessages.length).toBe(1);
|
||||
expect(receivedMessages[0].from).toBe("user1");
|
||||
expect(receivedMessages[0].content.text).toBe("Hello bot!");
|
||||
});
|
||||
|
||||
it("handles invalid JSON gracefully", async () => {
|
||||
const errors: string[] = [];
|
||||
mockWs.send = (data: string) => {
|
||||
const parsed = JSON.parse(data);
|
||||
if (parsed.type === "error") errors.push(parsed.error);
|
||||
};
|
||||
|
||||
manager.addClient(mockWs as any, mockAccount);
|
||||
mockWs.emit("message", Buffer.from("not valid json"));
|
||||
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
expect(errors.length).toBe(1);
|
||||
expect(errors[0]).toBe("Invalid message format");
|
||||
});
|
||||
});
|
||||
});
|
||||
256
src/channel.ts
Normal file
256
src/channel.ts
Normal file
@@ -0,0 +1,256 @@
|
||||
/**
|
||||
* WebSocket channel plugin implementation
|
||||
*/
|
||||
|
||||
import {
|
||||
createChatChannelPlugin,
|
||||
createChannelPluginBase,
|
||||
} from "openclaw/plugin-sdk/core";
|
||||
import type { OpenClawConfig, OpenClawPluginApi } from "openclaw/plugin-sdk/core";
|
||||
import type { WebSocketAccount, ChannelCapabilities } from "./config.ts";
|
||||
import { parseCapabilities } from "./config.ts";
|
||||
import { wsManager } from "./websocket-server.ts";
|
||||
|
||||
/**
|
||||
* Resolve account from OpenClaw config
|
||||
*/
|
||||
function resolveAccount(
|
||||
cfg: OpenClawConfig,
|
||||
accountId?: string | null
|
||||
): WebSocketAccount {
|
||||
const section = (cfg.channels as Record<string, any>)?.["websocket"];
|
||||
if (!section) {
|
||||
throw new Error("websocket: channel configuration not found");
|
||||
}
|
||||
|
||||
const accounts = section.accounts || [];
|
||||
if (!Array.isArray(accounts) || accounts.length === 0) {
|
||||
throw new Error("websocket: no accounts configured");
|
||||
}
|
||||
|
||||
// Find matching account
|
||||
const targetId = accountId ?? "default";
|
||||
const account = accounts.find((a: any) => a.accountId === targetId);
|
||||
|
||||
if (!account) {
|
||||
throw new Error(`websocket: account ${targetId} not found`);
|
||||
}
|
||||
|
||||
if (!account.apiKey) {
|
||||
throw new Error(`websocket: apiKey is required for account ${targetId}`);
|
||||
}
|
||||
|
||||
return {
|
||||
accountId: account.accountId ?? null,
|
||||
apiKey: account.apiKey,
|
||||
allowFrom: account.allowFrom ?? [],
|
||||
dmPolicy: account.dmPolicy ?? "allowlist",
|
||||
capabilities: parseCapabilities(account.capabilities),
|
||||
metadata: account.metadata,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Inspect account without materializing secrets
|
||||
*/
|
||||
function inspectAccount(
|
||||
cfg: OpenClawConfig,
|
||||
accountId?: string | null
|
||||
): {
|
||||
enabled: boolean;
|
||||
configured: boolean;
|
||||
accountStatus: string;
|
||||
capabilities?: ChannelCapabilities;
|
||||
} {
|
||||
const section = (cfg.channels as Record<string, any>)?.["websocket"];
|
||||
const accounts = section?.accounts;
|
||||
|
||||
if (!Array.isArray(accounts) || accounts.length === 0) {
|
||||
return {
|
||||
enabled: false,
|
||||
configured: false,
|
||||
accountStatus: "no-accounts",
|
||||
};
|
||||
}
|
||||
|
||||
const targetId = accountId ?? "default";
|
||||
const account = accounts.find((a: any) => a.accountId === targetId);
|
||||
|
||||
if (!account) {
|
||||
return {
|
||||
enabled: false,
|
||||
configured: false,
|
||||
accountStatus: "account-not-found",
|
||||
};
|
||||
}
|
||||
|
||||
const hasKey = Boolean(account.apiKey);
|
||||
|
||||
return {
|
||||
enabled: hasKey,
|
||||
configured: hasKey,
|
||||
accountStatus: hasKey ? "configured" : "missing-api-key",
|
||||
capabilities: parseCapabilities(account.capabilities),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all accounts from config
|
||||
*/
|
||||
function getAllAccounts(cfg: OpenClawConfig): WebSocketAccount[] {
|
||||
const section = (cfg.channels as Record<string, any>)?.["websocket"];
|
||||
const accounts = section?.accounts || [];
|
||||
|
||||
return accounts.map((a: any) => ({
|
||||
accountId: a.accountId ?? null,
|
||||
apiKey: a.apiKey,
|
||||
allowFrom: a.allowFrom ?? [],
|
||||
dmPolicy: a.dmPolicy ?? "allowlist",
|
||||
capabilities: parseCapabilities(a.capabilities),
|
||||
metadata: a.metadata,
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the WebSocket channel plugin
|
||||
*/
|
||||
export const websocketChannelPlugin = createChatChannelPlugin<WebSocketAccount>({
|
||||
base: createChannelPluginBase({
|
||||
id: "websocket",
|
||||
setup: {
|
||||
resolveAccount,
|
||||
inspectAccount,
|
||||
},
|
||||
meta: {
|
||||
name: "WebSocket Channel",
|
||||
description: "Custom WebSocket-based messaging channel",
|
||||
},
|
||||
capabilities: {
|
||||
supportsTyping: true,
|
||||
supportsReactions: true,
|
||||
supportsThreads: true,
|
||||
supportsMedia: true,
|
||||
},
|
||||
}),
|
||||
|
||||
// DM security configuration
|
||||
security: {
|
||||
dm: {
|
||||
channelKey: "websocket",
|
||||
resolvePolicy: (account) => account.dmPolicy,
|
||||
resolveAllowFrom: (account) => account.allowFrom,
|
||||
defaultPolicy: "allowlist",
|
||||
},
|
||||
},
|
||||
|
||||
// Pairing configuration
|
||||
pairing: {
|
||||
text: {
|
||||
idLabel: "WebSocket user ID",
|
||||
message: "Send this code to verify your identity:",
|
||||
notify: async ({ target, code, account }) => {
|
||||
// Resolve the account if it's a config object
|
||||
const resolvedAccount = (account as any)?.apiKey
|
||||
? (account as WebSocketAccount)
|
||||
: null;
|
||||
|
||||
if (resolvedAccount) {
|
||||
await wsManager.sendPairingCode(target, code, resolvedAccount);
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
// Threading configuration
|
||||
threading: {
|
||||
topLevelReplyToMode: "reply",
|
||||
resolveThreadId: (message) => message.threadId,
|
||||
},
|
||||
|
||||
// Outbound messaging
|
||||
outbound: {
|
||||
attachedResults: {
|
||||
sendText: async (params) => {
|
||||
const account = resolveAccount(params.cfg, params.accountId);
|
||||
const result = await wsManager.sendText(
|
||||
params.to,
|
||||
params.text,
|
||||
account,
|
||||
params.threadId
|
||||
);
|
||||
return { messageId: result.messageId };
|
||||
},
|
||||
},
|
||||
base: {
|
||||
sendMedia: async (params) => {
|
||||
const account = resolveAccount(params.cfg, params.accountId);
|
||||
await wsManager.sendMedia(
|
||||
params.to,
|
||||
params.filePath,
|
||||
account,
|
||||
params.threadId
|
||||
);
|
||||
},
|
||||
async sendTyping(params) {
|
||||
const account = resolveAccount(params.cfg, params.accountId);
|
||||
await wsManager.sendTyping(params.to, true, account);
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* Register WebSocket HTTP route for handling upgrades
|
||||
*/
|
||||
export function registerWebSocketRoute(
|
||||
api: OpenClawPluginApi,
|
||||
config: OpenClawConfig
|
||||
): void {
|
||||
const path = ((config.channels as any)?.websocket?.path as string) || "/ws";
|
||||
|
||||
api.registerHttpRoute({
|
||||
path,
|
||||
auth: "plugin",
|
||||
handler: async (req, res, _ctx, upgrade) => {
|
||||
// Handle WebSocket upgrade
|
||||
if (upgrade?.upgrade) {
|
||||
const accounts = getAllAccounts(config);
|
||||
|
||||
if (accounts.length === 0) {
|
||||
res.statusCode = 503;
|
||||
res.end("No WebSocket accounts configured");
|
||||
return true;
|
||||
}
|
||||
|
||||
// Perform upgrade
|
||||
const ws = upgrade.upgrade(req, req.socket, Buffer.alloc(0));
|
||||
|
||||
// Handle the connection
|
||||
const { handleWebSocketUpgrade } = await import("./client.ts");
|
||||
handleWebSocketUpgrade(ws, req, { accounts, api });
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
// Regular HTTP request - return info
|
||||
res.statusCode = 200;
|
||||
res.setHeader("Content-Type", "application/json");
|
||||
res.end(JSON.stringify({
|
||||
service: "OpenClaw WebSocket Channel",
|
||||
version: "1.0.0",
|
||||
websocket: true,
|
||||
path,
|
||||
}));
|
||||
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
// Start cleanup interval
|
||||
const interval = wsManager.startCleanupInterval();
|
||||
|
||||
// Clean up on plugin unload
|
||||
api.on?.("plugin_unload", () => {
|
||||
clearInterval(interval);
|
||||
});
|
||||
}
|
||||
147
src/client.ts
Normal file
147
src/client.ts
Normal file
@@ -0,0 +1,147 @@
|
||||
/**
|
||||
* WebSocket client connection handler
|
||||
*/
|
||||
|
||||
import type { WebSocket as WebSocketType } from "ws";
|
||||
import type { OpenClawPluginApi } from "openclaw/plugin-sdk/core";
|
||||
import type { WebSocketAccount } from "./config.ts";
|
||||
import { authenticateConnection } from "./auth.ts";
|
||||
import { wsManager, type InboundMessage, type ConnectedClient } from "./websocket-server.ts";
|
||||
|
||||
export type ClientHandlerOptions = {
|
||||
accounts: WebSocketAccount[];
|
||||
api: OpenClawPluginApi;
|
||||
};
|
||||
|
||||
/**
|
||||
* Handle WebSocket upgrade request
|
||||
*/
|
||||
export function handleWebSocketUpgrade(
|
||||
ws: WebSocketType,
|
||||
req: { headers: Record<string, string | string[]>; url?: string },
|
||||
options: ClientHandlerOptions
|
||||
): boolean {
|
||||
const { accounts, api } = options;
|
||||
|
||||
// Authenticate the connection
|
||||
const authResult = authenticateConnection(req, accounts);
|
||||
if (!authResult.success) {
|
||||
ws.close(1008, authResult.error);
|
||||
api.logger?.warn(`WebSocket authentication failed: ${authResult.error}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Add client to manager
|
||||
const client = wsManager.addClient(ws, authResult.account);
|
||||
api.logger?.info(
|
||||
`WebSocket client connected: account=${authResult.account.accountId}, connections=${wsManager.getConnectionCount(authResult.account.accountId)}`
|
||||
);
|
||||
|
||||
// Set up message handler for this client
|
||||
wsManager.onMessage(async (message, senderClient) => {
|
||||
await handleInboundMessage(message, senderClient, api);
|
||||
});
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Process incoming message from WebSocket client
|
||||
*/
|
||||
async function handleInboundMessage(
|
||||
message: InboundMessage,
|
||||
client: ConnectedClient,
|
||||
api: OpenClawPluginApi
|
||||
): Promise<void> {
|
||||
const account = client.account;
|
||||
|
||||
// Validate message type
|
||||
if (message.type !== "message") {
|
||||
api.logger?.warn(`Unsupported message type: ${message.type}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Validate capabilities
|
||||
const content = message.content;
|
||||
if (content.type === "image" && !account.capabilities.images) {
|
||||
api.logger?.warn(`Account ${account.accountId} tried to send image but capability is disabled`);
|
||||
return;
|
||||
}
|
||||
if (content.type === "file" && !account.capabilities.files) {
|
||||
api.logger?.warn(`Account ${account.accountId} tried to send file but capability is disabled`);
|
||||
return;
|
||||
}
|
||||
if (content.type === "voice" && !account.capabilities.voice) {
|
||||
api.logger?.warn(`Account ${account.accountId} tried to send voice but capability is disabled`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Check DM security - user must be in allowFrom list (unless policy is "open")
|
||||
if (account.dmPolicy === "closed") {
|
||||
api.logger?.debug(`Message rejected: DM policy is closed for account ${account.accountId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
if (account.dmPolicy === "allowlist" && !account.allowFrom.includes(message.from)) {
|
||||
api.logger?.debug(`Message rejected: ${message.from} not in allowlist for account ${account.accountId}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Dispatch to OpenClaw
|
||||
try {
|
||||
// Find the channel's inbound handler
|
||||
const channelApi = api as OpenClawPluginApi & {
|
||||
channelInbound?: {
|
||||
dispatch: (params: {
|
||||
channel: string;
|
||||
accountId: string | null;
|
||||
from: string;
|
||||
content: string;
|
||||
threadId?: string;
|
||||
}) => Promise<void>;
|
||||
};
|
||||
};
|
||||
|
||||
if (channelApi.channelInbound) {
|
||||
await channelApi.channelInbound.dispatch({
|
||||
channel: "websocket",
|
||||
accountId: account.accountId,
|
||||
from: message.from,
|
||||
content: content.type === "text" ? content.text : JSON.stringify(content),
|
||||
threadId: message.threadId,
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
api.logger?.error("Failed to dispatch inbound message:", err);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get active accounts from config
|
||||
*/
|
||||
export function getActiveAccounts(config: { websocket?: { accounts?: unknown[] } }): WebSocketAccount[] {
|
||||
const accounts: WebSocketAccount[] = [];
|
||||
const rawAccounts = config.websocket?.accounts;
|
||||
|
||||
if (Array.isArray(rawAccounts)) {
|
||||
for (const raw of rawAccounts) {
|
||||
if (isValidAccount(raw)) {
|
||||
accounts.push(raw);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return accounts;
|
||||
}
|
||||
|
||||
function isValidAccount(raw: unknown): raw is WebSocketAccount {
|
||||
if (!raw || typeof raw !== "object") return false;
|
||||
const obj = raw as Record<string, unknown>;
|
||||
return (
|
||||
typeof obj.accountId === "string" &&
|
||||
typeof obj.apiKey === "string" &&
|
||||
Array.isArray(obj.allowFrom) &&
|
||||
typeof obj.dmPolicy === "string" &&
|
||||
obj.capabilities !== null && typeof obj.capabilities === "object"
|
||||
);
|
||||
}
|
||||
50
src/config.ts
Normal file
50
src/config.ts
Normal file
@@ -0,0 +1,50 @@
|
||||
/**
|
||||
* Configuration types for WebSocket channel plugin
|
||||
*/
|
||||
|
||||
export type ChannelCapabilities = {
|
||||
text: boolean;
|
||||
images: boolean;
|
||||
files: boolean;
|
||||
voice: boolean;
|
||||
reactions: boolean;
|
||||
threads: boolean;
|
||||
typing: boolean;
|
||||
};
|
||||
|
||||
export type WebSocketAccount = {
|
||||
accountId: string | null;
|
||||
apiKey: string;
|
||||
allowFrom: string[];
|
||||
dmPolicy: "allowlist" | "open" | "closed";
|
||||
capabilities: ChannelCapabilities;
|
||||
metadata?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
export type WebSocketConfig = {
|
||||
port: number;
|
||||
path: string;
|
||||
accounts: WebSocketAccount[];
|
||||
};
|
||||
|
||||
export const defaultCapabilities: ChannelCapabilities = {
|
||||
text: true,
|
||||
images: false,
|
||||
files: false,
|
||||
voice: false,
|
||||
reactions: false,
|
||||
threads: false,
|
||||
typing: false,
|
||||
};
|
||||
|
||||
export function parseCapabilities(raw?: Partial<ChannelCapabilities>): ChannelCapabilities {
|
||||
return {
|
||||
text: raw?.text ?? defaultCapabilities.text,
|
||||
images: raw?.images ?? defaultCapabilities.images,
|
||||
files: raw?.files ?? defaultCapabilities.files,
|
||||
voice: raw?.voice ?? defaultCapabilities.voice,
|
||||
reactions: raw?.reactions ?? defaultCapabilities.reactions,
|
||||
threads: raw?.threads ?? defaultCapabilities.threads,
|
||||
typing: raw?.typing ?? defaultCapabilities.typing,
|
||||
};
|
||||
}
|
||||
301
src/websocket-server.ts
Normal file
301
src/websocket-server.ts
Normal file
@@ -0,0 +1,301 @@
|
||||
/**
|
||||
* WebSocket server implementation
|
||||
*/
|
||||
|
||||
import type { WebSocket as WebSocketType } from "ws";
|
||||
import type { WebSocketAccount, ChannelCapabilities } from "./config.ts";
|
||||
|
||||
// Message types for WebSocket protocol
|
||||
export type InboundMessage = {
|
||||
type: "message";
|
||||
accountId?: string;
|
||||
from: string;
|
||||
content: MessageContent;
|
||||
timestamp?: number;
|
||||
threadId?: string;
|
||||
};
|
||||
|
||||
export type OutboundMessage = {
|
||||
type: "message";
|
||||
to: string;
|
||||
content: MessageContent;
|
||||
timestamp: number;
|
||||
threadId?: string;
|
||||
messageId?: string;
|
||||
};
|
||||
|
||||
export type HandshakeMessage = {
|
||||
type: "handshake";
|
||||
accountId: string | null;
|
||||
capabilities: ChannelCapabilities;
|
||||
metadata: Record<string, unknown>;
|
||||
timestamp: number;
|
||||
};
|
||||
|
||||
export type TypingIndicator = {
|
||||
type: "typing";
|
||||
to: string;
|
||||
isTyping: boolean;
|
||||
};
|
||||
|
||||
export type PairingCodeMessage = {
|
||||
type: "pairing_code";
|
||||
code: string;
|
||||
target: string;
|
||||
message: string;
|
||||
};
|
||||
|
||||
export type WebSocketMessage =
|
||||
| InboundMessage
|
||||
| OutboundMessage
|
||||
| HandshakeMessage
|
||||
| TypingIndicator
|
||||
| PairingCodeMessage;
|
||||
|
||||
export type MessageContent =
|
||||
| { type: "text"; text: string }
|
||||
| { type: "image"; url: string; alt?: string }
|
||||
| { type: "file"; url: string; name: string; size?: number }
|
||||
| { type: "voice"; url: string; duration?: number };
|
||||
|
||||
// Connected client representation
|
||||
export type ConnectedClient = {
|
||||
ws: WebSocketType;
|
||||
account: WebSocketAccount;
|
||||
userId?: string;
|
||||
connectedAt: Date;
|
||||
lastPingAt: Date;
|
||||
};
|
||||
|
||||
export type MessageHandler = (message: InboundMessage, client: ConnectedClient) => Promise<void> | void;
|
||||
|
||||
export class WebSocketServerManager {
|
||||
private clients = new Map<WebSocketType, ConnectedClient>();
|
||||
private messageHandler?: MessageHandler;
|
||||
|
||||
/**
|
||||
* Register a handler for incoming messages
|
||||
*/
|
||||
onMessage(handler: MessageHandler): void {
|
||||
this.messageHandler = handler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a new connected client
|
||||
*/
|
||||
addClient(ws: WebSocketType, account: WebSocketAccount): ConnectedClient {
|
||||
const client: ConnectedClient = {
|
||||
ws,
|
||||
account,
|
||||
connectedAt: new Date(),
|
||||
lastPingAt: new Date(),
|
||||
};
|
||||
this.clients.set(ws, client);
|
||||
|
||||
// Set up message handler
|
||||
ws.on("message", async (data) => {
|
||||
try {
|
||||
const message = JSON.parse(data.toString()) as InboundMessage;
|
||||
if (this.messageHandler) {
|
||||
await this.messageHandler(message, client);
|
||||
}
|
||||
} catch (err) {
|
||||
console.error("Failed to parse WebSocket message:", err);
|
||||
this.sendError(ws, "Invalid message format");
|
||||
}
|
||||
});
|
||||
|
||||
// Handle client disconnect
|
||||
ws.on("close", () => {
|
||||
this.clients.delete(ws);
|
||||
});
|
||||
|
||||
// Handle pong (keep-alive)
|
||||
ws.on("pong", () => {
|
||||
const c = this.clients.get(ws);
|
||||
if (c) {
|
||||
c.lastPingAt = new Date();
|
||||
}
|
||||
});
|
||||
|
||||
// Send handshake
|
||||
this.sendHandshake(client);
|
||||
|
||||
return client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send handshake message to client
|
||||
*/
|
||||
private sendHandshake(client: ConnectedClient): void {
|
||||
const handshake: HandshakeMessage = {
|
||||
type: "handshake",
|
||||
accountId: client.account.accountId,
|
||||
capabilities: client.account.capabilities,
|
||||
metadata: client.account.metadata || {},
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
this.sendToClient(client.ws, handshake);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a message to a specific client
|
||||
*/
|
||||
private sendToClient(ws: WebSocketType, message: WebSocketMessage): boolean {
|
||||
if (ws.readyState === 1) { // WebSocket.OPEN
|
||||
ws.send(JSON.stringify(message));
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send error message to client
|
||||
*/
|
||||
sendError(ws: WebSocketType, error: string): void {
|
||||
if (ws.readyState === 1) {
|
||||
ws.send(JSON.stringify({ type: "error", error }));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a text message to a user
|
||||
*/
|
||||
async sendText(to: string, text: string, account: WebSocketAccount, threadId?: string): Promise<{ messageId: string }> {
|
||||
const messageId = `ws-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
|
||||
const outbound: OutboundMessage = {
|
||||
type: "message",
|
||||
to,
|
||||
content: { type: "text", text },
|
||||
timestamp: Date.now(),
|
||||
threadId,
|
||||
messageId,
|
||||
};
|
||||
|
||||
// Send to all connections for this account
|
||||
let sent = false;
|
||||
for (const [ws, client] of this.clients) {
|
||||
if (client.account.accountId === account.accountId) {
|
||||
if (this.sendToClient(ws, outbound)) {
|
||||
sent = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!sent) {
|
||||
throw new Error(`No active connections for account ${account.accountId}`);
|
||||
}
|
||||
|
||||
return { messageId };
|
||||
}
|
||||
|
||||
/**
|
||||
* Send media (file) to a user
|
||||
*/
|
||||
async sendMedia(to: string, filePath: string, account: WebSocketAccount, threadId?: string): Promise<void> {
|
||||
const outbound: OutboundMessage = {
|
||||
type: "message",
|
||||
to,
|
||||
content: { type: "file", url: filePath, name: filePath.split("/").pop() || "file" },
|
||||
timestamp: Date.now(),
|
||||
threadId,
|
||||
};
|
||||
|
||||
// Send to all connections for this account
|
||||
for (const [ws, client] of this.clients) {
|
||||
if (client.account.accountId === account.accountId) {
|
||||
this.sendToClient(ws, outbound);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send typing indicator
|
||||
*/
|
||||
async sendTyping(to: string, isTyping: boolean, account: WebSocketAccount): Promise<void> {
|
||||
const typing: TypingIndicator = {
|
||||
type: "typing",
|
||||
to,
|
||||
isTyping,
|
||||
};
|
||||
|
||||
for (const [ws, client] of this.clients) {
|
||||
if (client.account.accountId === account.accountId) {
|
||||
this.sendToClient(ws, typing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send pairing code to a user
|
||||
*/
|
||||
async sendPairingCode(target: string, code: string, account: WebSocketAccount): Promise<void> {
|
||||
const pairingMsg: PairingCodeMessage = {
|
||||
type: "pairing_code",
|
||||
code,
|
||||
target,
|
||||
message: `Send this code to verify your identity: ${code}`,
|
||||
};
|
||||
|
||||
for (const [ws, client] of this.clients) {
|
||||
if (client.account.accountId === account.accountId) {
|
||||
this.sendToClient(ws, pairingMsg);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get connection count for an account
|
||||
*/
|
||||
getConnectionCount(accountId: string | null): number {
|
||||
let count = 0;
|
||||
for (const client of this.clients.values()) {
|
||||
if (client.account.accountId === accountId) {
|
||||
count++;
|
||||
}
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all active connections
|
||||
*/
|
||||
getConnections(): ConnectedClient[] {
|
||||
return Array.from(this.clients.values());
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up disconnected clients
|
||||
*/
|
||||
cleanup(): void {
|
||||
const now = Date.now();
|
||||
for (const [ws, client] of this.clients) {
|
||||
// Remove clients that haven't responded to ping in 60 seconds
|
||||
if (now - client.lastPingAt.getTime() > 60000) {
|
||||
ws.terminate();
|
||||
this.clients.delete(ws);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start periodic cleanup interval
|
||||
*/
|
||||
startCleanupInterval(intervalMs = 30000): ReturnType<typeof setInterval> {
|
||||
return setInterval(() => this.cleanup(), intervalMs);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send ping to all clients
|
||||
*/
|
||||
pingAll(): void {
|
||||
for (const ws of this.clients.keys()) {
|
||||
if (ws.readyState === 1) {
|
||||
ws.ping();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Singleton instance
|
||||
export const wsManager = new WebSocketServerManager();
|
||||
Reference in New Issue
Block a user