refactor(relay): fix type-aware lint errors in WebSocket and crypto handling (#698)
This commit is contained in:
@@ -49,6 +49,52 @@ interface WebSocketWithAttachment extends WebSocket {
|
|||||||
deserializeAttachment(): unknown;
|
deserializeAttachment(): unknown;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function hasAttachmentMethods(ws: WebSocket): ws is WebSocketWithAttachment {
|
||||||
|
// Type-safe check for attachment methods - required for Cloudflare WebSocket hibernation API
|
||||||
|
// Use Reflect to check for methods without type assertions
|
||||||
|
return (
|
||||||
|
"serializeAttachment" in ws &&
|
||||||
|
"deserializeAttachment" in ws &&
|
||||||
|
typeof Reflect.get(ws, "serializeAttachment") === "function" &&
|
||||||
|
typeof Reflect.get(ws, "deserializeAttachment") === "function"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
function deserializeAttachment(ws: WebSocket): unknown {
|
||||||
|
if (!hasAttachmentMethods(ws)) return null;
|
||||||
|
try {
|
||||||
|
return ws.deserializeAttachment();
|
||||||
|
} catch {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function serializeAttachment(ws: WebSocket, value: unknown): void {
|
||||||
|
if (!hasAttachmentMethods(ws)) {
|
||||||
|
throw new Error("WebSocket does not support attachments");
|
||||||
|
}
|
||||||
|
ws.serializeAttachment(value);
|
||||||
|
}
|
||||||
|
|
||||||
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||||
|
return typeof value === "object" && value !== null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function getString(record: Record<string, unknown>, key: string): string | undefined {
|
||||||
|
const value = record[key];
|
||||||
|
return typeof value === "string" ? value : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
function getGlobalWebSocketPair(): (new () => WebSocketPair) | undefined {
|
||||||
|
// Access WebSocketPair from global scope (Cloudflare Workers runtime)
|
||||||
|
// Use Reflect to access global property without type assertions
|
||||||
|
const WebSocketPair = Reflect.get(globalThis, "WebSocketPair") as unknown;
|
||||||
|
if (typeof WebSocketPair === "function") {
|
||||||
|
return WebSocketPair as new () => WebSocketPair;
|
||||||
|
}
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
interface Env {
|
interface Env {
|
||||||
RELAY: DurableObjectNamespace;
|
RELAY: DurableObjectNamespace;
|
||||||
}
|
}
|
||||||
@@ -91,9 +137,11 @@ export class RelayDurableObject {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private createWebSocketPair(): [WebSocket, WebSocket] {
|
private createWebSocketPair(): [WebSocket, WebSocket] {
|
||||||
const pair = new (
|
const WebSocketPairCtor = getGlobalWebSocketPair();
|
||||||
globalThis as unknown as { WebSocketPair: new () => WebSocketPair }
|
if (!WebSocketPairCtor) {
|
||||||
).WebSocketPair();
|
throw new Error("WebSocketPair not available in global scope");
|
||||||
|
}
|
||||||
|
const pair: WebSocketPair = new WebSocketPairCtor();
|
||||||
return [pair[0], pair[1]];
|
return [pair[0], pair[1]];
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -146,8 +194,9 @@ export class RelayDurableObject {
|
|||||||
|
|
||||||
private handleControlKeepalive(ws: WebSocket, message: string): void {
|
private handleControlKeepalive(ws: WebSocket, message: string): void {
|
||||||
try {
|
try {
|
||||||
const parsed = JSON.parse(message) as unknown as { type?: unknown };
|
const parsed: unknown = JSON.parse(message);
|
||||||
if (parsed?.type !== "ping") return;
|
const parsedRecord = isRecord(parsed) ? parsed : null;
|
||||||
|
if (parsedRecord?.type !== "ping") return;
|
||||||
try {
|
try {
|
||||||
ws.send(JSON.stringify({ type: "pong", ts: Date.now() }));
|
ws.send(JSON.stringify({ type: "pong", ts: Date.now() }));
|
||||||
} catch {
|
} catch {
|
||||||
@@ -220,9 +269,8 @@ export class RelayDurableObject {
|
|||||||
const out = new Set<string>();
|
const out = new Set<string>();
|
||||||
for (const ws of this.state.getWebSockets("client")) {
|
for (const ws of this.state.getWebSockets("client")) {
|
||||||
try {
|
try {
|
||||||
const attachment = (
|
const attachmentRaw = deserializeAttachment(ws);
|
||||||
ws as WebSocketWithAttachment
|
const attachment = isRecord(attachmentRaw) ? attachmentRaw : null;
|
||||||
).deserializeAttachment() as RelaySessionAttachment | null;
|
|
||||||
if (
|
if (
|
||||||
attachment?.role === "client" &&
|
attachment?.role === "client" &&
|
||||||
typeof attachment.connectionId === "string" &&
|
typeof attachment.connectionId === "string" &&
|
||||||
@@ -271,7 +319,7 @@ export class RelayDurableObject {
|
|||||||
connectionId: null,
|
connectionId: null,
|
||||||
createdAt: Date.now(),
|
createdAt: Date.now(),
|
||||||
};
|
};
|
||||||
(server as WebSocketWithAttachment).serializeAttachment(attachment);
|
serializeAttachment(server, attachment);
|
||||||
|
|
||||||
console.log(`[Relay DO] v1:${role} connected to session ${serverId}`);
|
console.log(`[Relay DO] v1:${role} connected to session ${serverId}`);
|
||||||
|
|
||||||
@@ -322,7 +370,7 @@ export class RelayDurableObject {
|
|||||||
connectionId: resolvedConnectionId || null,
|
connectionId: resolvedConnectionId || null,
|
||||||
createdAt: Date.now(),
|
createdAt: Date.now(),
|
||||||
};
|
};
|
||||||
(server as WebSocketWithAttachment).serializeAttachment(attachment);
|
serializeAttachment(server, attachment);
|
||||||
|
|
||||||
let roleSuffix = "";
|
let roleSuffix = "";
|
||||||
if (isServerControl) {
|
if (isServerControl) {
|
||||||
@@ -359,7 +407,8 @@ export class RelayDurableObject {
|
|||||||
|
|
||||||
async fetch(request: Request): Promise<Response> {
|
async fetch(request: Request): Promise<Response> {
|
||||||
const url = new URL(request.url);
|
const url = new URL(request.url);
|
||||||
const role = url.searchParams.get("role") as ConnectionRole | null;
|
const roleRaw = url.searchParams.get("role");
|
||||||
|
const role = roleRaw === "server" || roleRaw === "client" ? roleRaw : null;
|
||||||
const serverId = url.searchParams.get("serverId");
|
const serverId = url.searchParams.get("serverId");
|
||||||
const connectionIdRaw = url.searchParams.get("connectionId");
|
const connectionIdRaw = url.searchParams.get("connectionId");
|
||||||
const connectionId = typeof connectionIdRaw === "string" ? connectionIdRaw.trim() : "";
|
const connectionId = typeof connectionIdRaw === "string" ? connectionIdRaw.trim() : "";
|
||||||
@@ -388,15 +437,14 @@ export class RelayDurableObject {
|
|||||||
* Called when a WebSocket message is received (wakes from hibernation).
|
* Called when a WebSocket message is received (wakes from hibernation).
|
||||||
*/
|
*/
|
||||||
webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): void {
|
webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): void {
|
||||||
const attachment = (
|
const attachmentRaw = deserializeAttachment(ws);
|
||||||
ws as WebSocketWithAttachment
|
if (!isRecord(attachmentRaw)) {
|
||||||
).deserializeAttachment() as RelaySessionAttachment | null;
|
|
||||||
if (!attachment) {
|
|
||||||
console.error("[Relay DO] Message from WebSocket without attachment");
|
console.error("[Relay DO] Message from WebSocket without attachment");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
const attachment = attachmentRaw;
|
||||||
|
|
||||||
const version = attachment.version ?? LEGACY_RELAY_VERSION;
|
const version = getString(attachment, "version") ?? LEGACY_RELAY_VERSION;
|
||||||
|
|
||||||
if (version === LEGACY_RELAY_VERSION) {
|
if (version === LEGACY_RELAY_VERSION) {
|
||||||
const targetRole = attachment.role === "server" ? "client" : "server";
|
const targetRole = attachment.role === "server" ? "client" : "server";
|
||||||
@@ -411,7 +459,8 @@ export class RelayDurableObject {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
const { role, connectionId } = attachment;
|
const role = getString(attachment, "role");
|
||||||
|
const connectionId = getString(attachment, "connectionId");
|
||||||
if (!connectionId) {
|
if (!connectionId) {
|
||||||
// Control channel: support simple app-level keepalive.
|
// Control channel: support simple app-level keepalive.
|
||||||
if (typeof message === "string") {
|
if (typeof message === "string") {
|
||||||
@@ -451,44 +500,46 @@ export class RelayDurableObject {
|
|||||||
* Called when a WebSocket closes (wakes from hibernation).
|
* Called when a WebSocket closes (wakes from hibernation).
|
||||||
*/
|
*/
|
||||||
webSocketClose(ws: WebSocket, code: number, reason: string, _wasClean: boolean): void {
|
webSocketClose(ws: WebSocket, code: number, reason: string, _wasClean: boolean): void {
|
||||||
const attachment = (
|
const attachmentRaw = deserializeAttachment(ws);
|
||||||
ws as WebSocketWithAttachment
|
if (!isRecord(attachmentRaw)) return;
|
||||||
).deserializeAttachment() as RelaySessionAttachment | null;
|
const attachment = attachmentRaw;
|
||||||
if (!attachment) return;
|
|
||||||
|
|
||||||
const version = attachment.version ?? LEGACY_RELAY_VERSION;
|
const version = getString(attachment, "version") ?? LEGACY_RELAY_VERSION;
|
||||||
|
const role = getString(attachment, "role");
|
||||||
|
const connectionId = getString(attachment, "connectionId");
|
||||||
|
const serverId = getString(attachment, "serverId");
|
||||||
console.log(
|
console.log(
|
||||||
`[Relay DO] v${version}:${attachment.role}${attachment.connectionId ? `(${attachment.connectionId})` : ""} disconnected from session ${attachment.serverId} (${code}: ${reason})`,
|
`[Relay DO] v${version}:${role ?? "unknown"}${connectionId ? `(${connectionId})` : ""} disconnected from session ${serverId ?? "unknown"} (${code}: ${reason})`,
|
||||||
);
|
);
|
||||||
|
|
||||||
if (version === LEGACY_RELAY_VERSION) {
|
if (version === LEGACY_RELAY_VERSION) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (attachment.role === "client" && attachment.connectionId) {
|
if (role === "client" && connectionId) {
|
||||||
const remainingClientSockets = this.state
|
const remainingClientSockets = this.state
|
||||||
.getWebSockets(`client:${attachment.connectionId}`)
|
.getWebSockets(`client:${connectionId}`)
|
||||||
.some((socket) => socket !== ws);
|
.some((socket) => socket !== ws);
|
||||||
if (remainingClientSockets) {
|
if (remainingClientSockets) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
this.pendingFrames.delete(attachment.connectionId);
|
this.pendingFrames.delete(connectionId);
|
||||||
// Last socket for this session closed: now clean up matching server-data socket.
|
// Last socket for this session closed: now clean up matching server-data socket.
|
||||||
for (const serverWs of this.state.getWebSockets(`server:${attachment.connectionId}`)) {
|
for (const serverWs of this.state.getWebSockets(`server:${connectionId}`)) {
|
||||||
try {
|
try {
|
||||||
serverWs.close(1001, "Client disconnected");
|
serverWs.close(1001, "Client disconnected");
|
||||||
} catch {
|
} catch {
|
||||||
// ignore
|
// ignore
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
this.notifyControls({ type: "disconnected", connectionId: attachment.connectionId });
|
this.notifyControls({ type: "disconnected", connectionId });
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (attachment.role === "server" && attachment.connectionId) {
|
if (role === "server" && connectionId) {
|
||||||
// Force the client to reconnect and re-handshake when the daemon side drops.
|
// Force the client to reconnect and re-handshake when the daemon side drops.
|
||||||
for (const clientWs of this.state.getWebSockets(`client:${attachment.connectionId}`)) {
|
for (const clientWs of this.state.getWebSockets(`client:${connectionId}`)) {
|
||||||
try {
|
try {
|
||||||
clientWs.close(1012, "Server disconnected");
|
clientWs.close(1012, "Server disconnected");
|
||||||
} catch {
|
} catch {
|
||||||
@@ -502,10 +553,10 @@ export class RelayDurableObject {
|
|||||||
* Called on WebSocket error.
|
* Called on WebSocket error.
|
||||||
*/
|
*/
|
||||||
webSocketError(ws: WebSocket, error: unknown): void {
|
webSocketError(ws: WebSocket, error: unknown): void {
|
||||||
const attachment = (
|
const attachmentRaw = deserializeAttachment(ws);
|
||||||
ws as WebSocketWithAttachment
|
const attachment = isRecord(attachmentRaw) ? attachmentRaw : null;
|
||||||
).deserializeAttachment() as RelaySessionAttachment | null;
|
const role = attachment ? getString(attachment, "role") : undefined;
|
||||||
console.error(`[Relay DO] WebSocket error for ${attachment?.role ?? "unknown"}:`, error);
|
console.error(`[Relay DO] WebSocket error for ${role ?? "unknown"}:`, error);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -26,6 +26,15 @@ const NONCE_LENGTH = nacl.box.nonceLength; // 24
|
|||||||
|
|
||||||
let prngReady = false;
|
let prngReady = false;
|
||||||
|
|
||||||
|
interface GlobalWithCrypto {
|
||||||
|
crypto?: Crypto;
|
||||||
|
}
|
||||||
|
|
||||||
|
function getGlobalCrypto(): Crypto | undefined {
|
||||||
|
const g = globalThis as GlobalWithCrypto;
|
||||||
|
return g.crypto;
|
||||||
|
}
|
||||||
|
|
||||||
function ensurePrng(): void {
|
function ensurePrng(): void {
|
||||||
if (prngReady) return;
|
if (prngReady) return;
|
||||||
|
|
||||||
@@ -37,7 +46,7 @@ function ensurePrng(): void {
|
|||||||
// fallthrough
|
// fallthrough
|
||||||
}
|
}
|
||||||
|
|
||||||
const cryptoObj = (globalThis as unknown as { crypto?: Crypto }).crypto;
|
const cryptoObj = getGlobalCrypto();
|
||||||
if (cryptoObj?.getRandomValues) {
|
if (cryptoObj?.getRandomValues) {
|
||||||
nacl.setPRNG((x, n) => {
|
nacl.setPRNG((x, n) => {
|
||||||
const buf = new Uint8Array(n);
|
const buf = new Uint8Array(n);
|
||||||
|
|||||||
@@ -56,9 +56,25 @@ interface E2EEReadyMessage {
|
|||||||
type: "e2ee_ready";
|
type: "e2ee_ready";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||||
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||||
|
}
|
||||||
|
|
||||||
|
function isE2EEHelloMessage(value: unknown): value is E2EEHelloMessage {
|
||||||
|
return (
|
||||||
|
isRecord(value) &&
|
||||||
|
value.type === "e2ee_hello" &&
|
||||||
|
typeof value.key === "string" &&
|
||||||
|
value.key.trim().length > 0
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
function isE2EEReadyMessage(value: unknown): value is E2EEReadyMessage {
|
||||||
|
return isRecord(value) && value.type === "e2ee_ready";
|
||||||
|
}
|
||||||
|
|
||||||
function buildInvalidHelloError(rawText: string, parsed?: unknown): Error {
|
function buildInvalidHelloError(rawText: string, parsed?: unknown): Error {
|
||||||
const parsedRecord =
|
const parsedRecord = isRecord(parsed) ? parsed : null;
|
||||||
parsed && typeof parsed === "object" ? (parsed as Record<string, unknown>) : null;
|
|
||||||
const rawType = parsedRecord?.type;
|
const rawType = parsedRecord?.type;
|
||||||
function describeType(value: unknown): string {
|
function describeType(value: unknown): string {
|
||||||
if (typeof value === "string") return value;
|
if (typeof value === "string") return value;
|
||||||
@@ -77,6 +93,19 @@ function buildInvalidHelloError(rawText: string, parsed?: unknown): Error {
|
|||||||
const HANDSHAKE_RETRY_MS = 1000;
|
const HANDSHAKE_RETRY_MS = 1000;
|
||||||
const MAX_PENDING_SENDS = 200;
|
const MAX_PENDING_SENDS = 200;
|
||||||
|
|
||||||
|
interface TimeoutWithUnref {
|
||||||
|
unref(): void;
|
||||||
|
}
|
||||||
|
|
||||||
|
function hasUnref(timeout: unknown): timeout is TimeoutWithUnref {
|
||||||
|
return (
|
||||||
|
typeof timeout === "object" &&
|
||||||
|
timeout !== null &&
|
||||||
|
"unref" in timeout &&
|
||||||
|
typeof (timeout as Record<string, unknown>).unref === "function"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates an encrypted channel as the initiator (client).
|
* Creates an encrypted channel as the initiator (client).
|
||||||
*
|
*
|
||||||
@@ -137,7 +166,9 @@ export async function createClientChannel(
|
|||||||
sendHello();
|
sendHello();
|
||||||
}, HANDSHAKE_RETRY_MS);
|
}, HANDSHAKE_RETRY_MS);
|
||||||
// Avoid keeping Node processes alive (e.g. tests) if the handshake is stuck.
|
// Avoid keeping Node processes alive (e.g. tests) if the handshake is stuck.
|
||||||
(retry as unknown as { unref?: () => void }).unref?.();
|
if (hasUnref(retry)) {
|
||||||
|
retry.unref();
|
||||||
|
}
|
||||||
|
|
||||||
return channel;
|
return channel;
|
||||||
}
|
}
|
||||||
@@ -160,8 +191,8 @@ export async function createDaemonChannel(
|
|||||||
const shouldIgnorePostHelloPlaintext = (data: string | ArrayBuffer): boolean => {
|
const shouldIgnorePostHelloPlaintext = (data: string | ArrayBuffer): boolean => {
|
||||||
try {
|
try {
|
||||||
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
||||||
const parsed = JSON.parse(text) as Partial<E2EEHelloMessage | E2EEReadyMessage>;
|
const parsed: unknown = JSON.parse(text);
|
||||||
return parsed.type === "e2ee_hello" || parsed.type === "e2ee_ready";
|
return isE2EEHelloMessage(parsed) || isE2EEReadyMessage(parsed);
|
||||||
} catch {
|
} catch {
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
@@ -178,11 +209,12 @@ export async function createDaemonChannel(
|
|||||||
throw buildInvalidHelloError(helloText);
|
throw buildInvalidHelloError(helloText);
|
||||||
}
|
}
|
||||||
|
|
||||||
const msg = parsed as Partial<E2EEHelloMessage>;
|
if (!isE2EEHelloMessage(parsed)) {
|
||||||
if (msg.type !== "e2ee_hello" || typeof msg.key !== "string" || !msg.key.trim()) {
|
|
||||||
throw buildInvalidHelloError(helloText, parsed);
|
throw buildInvalidHelloError(helloText, parsed);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const msg = parsed;
|
||||||
|
|
||||||
// Buffer any subsequent messages that arrive while we're doing async
|
// Buffer any subsequent messages that arrive while we're doing async
|
||||||
// WebCrypto work to derive the shared key. Without this, it's possible
|
// WebCrypto work to derive the shared key. Without this, it's possible
|
||||||
// for the next message (already encrypted) to be misinterpreted as a
|
// for the next message (already encrypted) to be misinterpreted as a
|
||||||
@@ -269,8 +301,8 @@ export class EncryptedChannel {
|
|||||||
if (this.state === "handshaking") {
|
if (this.state === "handshaking") {
|
||||||
try {
|
try {
|
||||||
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
||||||
const msg = JSON.parse(text) as Partial<E2EEReadyMessage>;
|
const parsed: unknown = JSON.parse(text);
|
||||||
if (msg.type === "e2ee_ready") {
|
if (isE2EEReadyMessage(parsed)) {
|
||||||
this.state = "open";
|
this.state = "open";
|
||||||
this.events.onopen?.();
|
this.events.onopen?.();
|
||||||
for (const cb of this.onOpenCallbacks) cb();
|
for (const cb of this.onOpenCallbacks) cb();
|
||||||
@@ -290,16 +322,16 @@ export class EncryptedChannel {
|
|||||||
try {
|
try {
|
||||||
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
||||||
if (text.trim().startsWith("{")) {
|
if (text.trim().startsWith("{")) {
|
||||||
const parsed = JSON.parse(text) as Partial<E2EEHelloMessage | E2EEReadyMessage>;
|
const parsed: unknown = JSON.parse(text);
|
||||||
|
|
||||||
if (parsed.type === "e2ee_hello" && typeof parsed.key === "string") {
|
if (isE2EEHelloMessage(parsed)) {
|
||||||
if (this.options.daemonKeyPair) {
|
if (this.options.daemonKeyPair) {
|
||||||
await this.handleDaemonRehello(parsed.key);
|
await this.handleDaemonRehello(parsed.key);
|
||||||
}
|
}
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (parsed.type === "e2ee_ready") {
|
if (isE2EEReadyMessage(parsed)) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user