signal-desktop/ts/textsecure/WebsocketResources.ts

1245 lines
35 KiB
TypeScript
Raw Normal View History

2023-01-03 19:55:46 +00:00
// Copyright 2020 Signal Messenger, LLC
2020-10-30 20:34:04 +00:00
// SPDX-License-Identifier: AGPL-3.0-only
2021-08-26 18:18:00 +00:00
/* eslint-disable max-classes-per-file */
/*
* WebSocket-Resources
*
* Create a request-response interface over websockets using the
* WebSocket-Resources sub-protocol[1].
*
* var client = new WebSocketResource(socket, function(request) {
* request.respond(200, 'OK');
* });
*
* const { response, status } = await client.sendRequest({
* verb: 'PUT',
* path: '/v1/messages',
* headers: ['content-type:application/json'],
* body: Buffer.from('{ some: "json" }'),
* });
*
* 1. https://github.com/signalapp/WebSocket-Resources
*
*/
/* eslint-disable @typescript-eslint/no-namespace */
/* eslint-disable @typescript-eslint/brace-style */
import type { connection as WebSocket, IMessage } from 'websocket';
2022-03-23 20:49:27 +00:00
import Long from 'long';
2023-06-07 00:36:38 +00:00
import pTimeout from 'p-timeout';
import { Response } from 'node-fetch';
import net from 'net';
import { z } from 'zod';
import { clearInterval } from 'timers';
import { random } from 'lodash';
import type { ChatServiceDebugInfo } from '@signalapp/libsignal-client/Native';
2021-06-09 22:28:54 +00:00
2024-09-19 18:32:17 +00:00
import type { LibSignalError, Net } from '@signalapp/libsignal-client';
2024-08-06 21:21:15 +00:00
import { Buffer } from 'node:buffer';
import type {
ChatServerMessageAck,
ChatServiceListener,
ConnectionEventsListener,
} from '@signalapp/libsignal-client/dist/net';
import type { EventHandler } from './EventTarget';
import EventTarget from './EventTarget';
import * as durations from '../util/durations';
2021-07-02 19:21:24 +00:00
import { dropNull } from '../util/dropNull';
import { drop } from '../util/drop';
import { isOlderThan } from '../util/timestamp';
2021-07-02 19:21:24 +00:00
import { strictAssert } from '../util/assert';
import * as Errors from '../types/errors';
2021-07-02 19:21:24 +00:00
import { SignalService as Proto } from '../protobuf';
import * as log from '../logging/log';
import * as Timers from '../Timers';
import type { IResource } from './WebSocket';
import { isProduction } from '../util/version';
import { ToastType } from '../types/Toast';
import { AbortableProcess } from '../util/AbortableProcess';
2024-08-06 21:21:15 +00:00
import type { WebAPICredentials } from './Types';
2024-09-19 18:32:17 +00:00
import { NORMAL_DISCONNECT_CODE } from './SocketManager';
2024-10-02 19:03:10 +00:00
import { parseUnknown } from '../util/schemas';
const THIRTY_SECONDS = 30 * durations.SECOND;
const STATS_UPDATE_INTERVAL = durations.MINUTE;
const MAX_MESSAGE_SIZE = 512 * 1024;
const AGGREGATED_STATS_KEY = 'websocketStats';
export enum IpVersion {
IPv4 = 'ipv4',
IPv6 = 'ipv6',
}
export namespace IpVersion {
export function fromDebugInfoCode(ipType: number): IpVersion | undefined {
switch (ipType) {
case 1:
return IpVersion.IPv4;
case 2:
return IpVersion.IPv6;
default:
return undefined;
}
}
}
const AggregatedStatsSchema = z.object({
connectionFailures: z.number(),
requestsCompared: z.number(),
ipVersionMismatches: z.number(),
healthcheckFailures: z.number(),
healthcheckBadStatus: z.number(),
lastToastTimestamp: z.number(),
});
export type AggregatedStats = z.infer<typeof AggregatedStatsSchema>;
// eslint-disable-next-line @typescript-eslint/no-redeclare
export namespace AggregatedStats {
export function loadOrCreateEmpty(name: string): AggregatedStats {
const key = localStorageKey(name);
try {
const json = localStorage.getItem(key);
return json != null
2024-10-02 19:03:10 +00:00
? parseUnknown(AggregatedStatsSchema, JSON.parse(json) as unknown)
: createEmpty();
} catch (error) {
log.warn(
`Could not load [${key}] from local storage. Possibly, attempting to load for the first time`,
Errors.toLogFormat(error)
);
return createEmpty();
}
}
export function store(stats: AggregatedStats, name: string): void {
const key = localStorageKey(name);
try {
const json = JSON.stringify(stats);
localStorage.setItem(key, json);
} catch (error) {
log.warn(
`Failed to store key [${key}] to the local storage`,
Errors.toLogFormat(error)
);
}
}
export function add(a: AggregatedStats, b: AggregatedStats): AggregatedStats {
return {
requestsCompared: a.requestsCompared + b.requestsCompared,
connectionFailures: a.connectionFailures + b.connectionFailures,
healthcheckFailures: a.healthcheckFailures + b.healthcheckFailures,
ipVersionMismatches: a.ipVersionMismatches + b.ipVersionMismatches,
healthcheckBadStatus: a.healthcheckBadStatus + b.healthcheckBadStatus,
lastToastTimestamp: Math.max(a.lastToastTimestamp, b.lastToastTimestamp),
};
}
export function createEmpty(): AggregatedStats {
return {
requestsCompared: 0,
connectionFailures: 0,
ipVersionMismatches: 0,
healthcheckFailures: 0,
healthcheckBadStatus: 0,
lastToastTimestamp: 0,
};
}
export function shouldReportError(stats: AggregatedStats): boolean {
const timeSinceLastToast = Date.now() - stats.lastToastTimestamp;
if (timeSinceLastToast < durations.DAY || stats.requestsCompared < 1000) {
return false;
}
const totalFailuresSinceLastToast =
stats.healthcheckBadStatus +
stats.healthcheckFailures +
stats.connectionFailures;
return totalFailuresSinceLastToast > 20;
}
export function localStorageKey(name: string): string {
return `${AGGREGATED_STATS_KEY}.${name}`;
}
}
2024-08-06 21:21:15 +00:00
export enum ServerRequestType {
ApiMessage = '/api/v1/message',
ApiEmptyQueue = '/api/v1/queue/empty',
ProvisioningMessage = '/v1/message',
ProvisioningAddress = '/v1/address',
Unknown = 'unknown',
}
export type IncomingWebSocketRequest = {
readonly requestType: ServerRequestType;
readonly body: Uint8Array | undefined;
readonly timestamp: number | undefined;
respond(status: number, message: string): void;
};
export class IncomingWebSocketRequestLibsignal
implements IncomingWebSocketRequest
{
constructor(
readonly requestType: ServerRequestType,
readonly body: Uint8Array | undefined,
readonly timestamp: number | undefined,
private readonly ack: ChatServerMessageAck | undefined
) {}
respond(status: number, _message: string): void {
if (this.ack) {
drop(this.ack.send(status));
}
}
}
2024-08-06 21:21:15 +00:00
export class IncomingWebSocketRequestLegacy
implements IncomingWebSocketRequest
{
private readonly id: Long;
2024-08-06 21:21:15 +00:00
public readonly requestType: ServerRequestType;
2021-07-02 19:21:24 +00:00
public readonly body: Uint8Array | undefined;
2024-08-06 21:21:15 +00:00
public readonly timestamp: number | undefined;
2021-07-02 19:21:24 +00:00
constructor(
request: Proto.IWebSocketRequestMessage,
private readonly sendBytes: (bytes: Buffer) => void
2021-07-02 19:21:24 +00:00
) {
strictAssert(request.id, 'request without id');
strictAssert(request.verb, 'request without verb');
strictAssert(request.path, 'request without path');
2021-07-02 19:21:24 +00:00
this.id = request.id;
2024-08-06 21:21:15 +00:00
this.requestType = resolveType(request.path, request.verb);
2021-07-02 19:21:24 +00:00
this.body = dropNull(request.body);
2024-08-06 21:21:15 +00:00
this.timestamp = resolveTimestamp(request.headers || []);
2021-07-02 19:21:24 +00:00
}
2021-07-02 19:21:24 +00:00
public respond(status: number, message: string): void {
const bytes = Proto.WebSocketMessage.encode({
type: Proto.WebSocketMessage.Type.RESPONSE,
response: { id: this.id, message, status },
}).finish();
this.sendBytes(Buffer.from(bytes));
}
}
2024-08-06 21:21:15 +00:00
function resolveType(path: string, verb: string): ServerRequestType {
if (path === ServerRequestType.ApiMessage) {
return ServerRequestType.ApiMessage;
}
if (path === ServerRequestType.ApiEmptyQueue && verb === 'PUT') {
return ServerRequestType.ApiEmptyQueue;
}
if (path === ServerRequestType.ProvisioningAddress && verb === 'PUT') {
return ServerRequestType.ProvisioningAddress;
}
if (path === ServerRequestType.ProvisioningMessage && verb === 'PUT') {
return ServerRequestType.ProvisioningMessage;
}
return ServerRequestType.Unknown;
}
function resolveTimestamp(headers: ReadonlyArray<string>): number | undefined {
// The 'X-Signal-Timestamp' is usually the last item, so start there.
let it = headers.length;
// eslint-disable-next-line no-plusplus
while (--it >= 0) {
const match = headers[it].match(/^X-Signal-Timestamp:\s*(\d+)\s*$/i);
if (match && match.length === 2) {
return Number(match[1]);
}
}
return undefined;
}
export type SendRequestOptions = Readonly<{
verb: string;
path: string;
2021-07-02 19:21:24 +00:00
body?: Uint8Array;
timeout?: number;
headers?: ReadonlyArray<[string, string]>;
2021-07-02 19:21:24 +00:00
}>;
export type SendRequestResult = Readonly<{
status: number;
message: string;
response?: Uint8Array;
headers: ReadonlyArray<string>;
}>;
export enum TransportOption {
// Only original transport is used
Original = 'original',
// All requests are going through the original transport,
// but for every request that completes sucessfully we're initiating
// a healthcheck request via libsignal transport,
// collecting comparison statistics, and if we see many inconsistencies,
// we're showing a toast asking user to submit a debug log
ShadowingHigh = 'shadowingHigh',
// Similar to `shadowingHigh`, however, only 10% of requests
// will trigger a healthcheck, and toast is never shown.
// Statistics data is still added to the debug logs,
// so it will be available to us with all the debug log uploads.
ShadowingLow = 'shadowingLow',
// Only libsignal transport is used
Libsignal = 'libsignal',
}
2021-07-02 19:21:24 +00:00
export type WebSocketResourceOptions = {
2023-06-07 00:36:38 +00:00
name: string;
2021-07-02 19:21:24 +00:00
handleRequest?: (request: IncomingWebSocketRequest) => void;
2023-06-07 00:36:38 +00:00
keepalive?: KeepAliveOptionsType;
transportOption?: TransportOption;
2021-07-02 19:21:24 +00:00
};
2021-07-09 19:36:10 +00:00
export class CloseEvent extends Event {
2024-07-24 00:31:40 +00:00
constructor(
public readonly code: number,
public readonly reason: string
) {
2021-07-09 19:36:10 +00:00
super('close');
}
}
// eslint-disable-next-line no-restricted-syntax
export interface IWebSocketResource extends IResource {
sendRequest(options: SendRequestOptions): Promise<Response>;
addEventListener(name: 'close', handler: (ev: CloseEvent) => void): void;
forceKeepAlive(timeout?: number): void;
shutdown(): void;
close(code?: number, reason?: string): void;
localPort(): number | undefined;
}
type LibsignalWebSocketResourceHolder = {
resource: LibsignalWebSocketResource | undefined;
};
2024-09-19 18:32:17 +00:00
const UNEXPECTED_DISCONNECT_CODE = 3001;
2024-08-06 21:21:15 +00:00
export function connectUnauthenticatedLibsignal({
libsignalNet,
name,
keepalive,
2024-08-06 21:21:15 +00:00
}: {
libsignalNet: Net.Net;
name: string;
keepalive: KeepAliveOptionsType;
2024-08-06 21:21:15 +00:00
}): AbortableProcess<LibsignalWebSocketResource> {
const logId = `LibsignalWebSocketResource(${name})`;
const listener: LibsignalWebSocketResourceHolder & ConnectionEventsListener =
{
resource: undefined,
2024-09-19 18:32:17 +00:00
onConnectionInterrupted(cause: LibSignalError | null): void {
if (!this.resource) {
logDisconnectedListenerWarn(logId, 'onConnectionInterrupted');
return;
}
2024-09-19 18:32:17 +00:00
this.resource.onConnectionInterrupted(cause);
this.resource = undefined;
},
};
return connectLibsignal(
libsignalNet.newUnauthenticatedChatService(listener),
listener,
logId,
keepalive
);
2024-08-06 21:21:15 +00:00
}
export function connectAuthenticatedLibsignal({
libsignalNet,
name,
credentials,
handler,
receiveStories,
keepalive,
2024-08-06 21:21:15 +00:00
}: {
libsignalNet: Net.Net;
name: string;
credentials: WebAPICredentials;
handler: (request: IncomingWebSocketRequest) => void;
receiveStories: boolean;
keepalive: KeepAliveOptionsType;
2024-08-06 21:21:15 +00:00
}): AbortableProcess<LibsignalWebSocketResource> {
const logId = `LibsignalWebSocketResource(${name})`;
const listener: LibsignalWebSocketResourceHolder & ChatServiceListener = {
resource: undefined,
2024-08-06 21:21:15 +00:00
onIncomingMessage(
envelope: Buffer,
timestamp: number,
ack: ChatServerMessageAck
): void {
// Handle incoming messages even if we've disconnected.
2024-08-06 21:21:15 +00:00
const request = new IncomingWebSocketRequestLibsignal(
ServerRequestType.ApiMessage,
envelope,
timestamp,
ack
);
handler(request);
},
onQueueEmpty(): void {
if (!this.resource) {
logDisconnectedListenerWarn(logId, 'onQueueEmpty');
return;
}
2024-08-06 21:21:15 +00:00
const request = new IncomingWebSocketRequestLibsignal(
ServerRequestType.ApiEmptyQueue,
undefined,
undefined,
undefined
);
handler(request);
},
2024-09-19 18:32:17 +00:00
onConnectionInterrupted(cause): void {
if (!this.resource) {
logDisconnectedListenerWarn(logId, 'onConnectionInterrupted');
return;
}
2024-09-19 18:32:17 +00:00
this.resource.onConnectionInterrupted(cause);
this.resource = undefined;
2024-08-06 21:21:15 +00:00
},
};
return connectLibsignal(
libsignalNet.newAuthenticatedChatService(
credentials.username,
credentials.password,
receiveStories,
listener
),
listener,
logId,
keepalive
2024-08-06 21:21:15 +00:00
);
}
function logDisconnectedListenerWarn(logId: string, method: string): void {
log.warn(`${logId} received ${method}, but listener already disconnected`);
}
2024-08-06 21:21:15 +00:00
function connectLibsignal(
chatService: Net.ChatService,
resourceHolder: LibsignalWebSocketResourceHolder,
logId: string,
keepalive: KeepAliveOptionsType
2024-08-06 21:21:15 +00:00
): AbortableProcess<LibsignalWebSocketResource> {
const connectAsync = async () => {
try {
const debugInfo = await chatService.connect();
log.info(`${logId} connected`, debugInfo);
const resource = new LibsignalWebSocketResource(
2024-08-06 21:21:15 +00:00
chatService,
IpVersion.fromDebugInfoCode(debugInfo.ipType),
logId,
keepalive
2024-08-06 21:21:15 +00:00
);
// eslint-disable-next-line no-param-reassign
resourceHolder.resource = resource;
return resource;
2024-08-06 21:21:15 +00:00
} catch (error) {
// Handle any errors that occur during connection
log.error(`${logId} connection failed`, Errors.toLogFormat(error));
2024-08-06 21:21:15 +00:00
throw error;
}
};
return new AbortableProcess<LibsignalWebSocketResource>(
`${logId}.connect`,
2024-08-06 21:21:15 +00:00
{
abort() {
// if interrupted, trying to disconnect
drop(chatService.disconnect());
},
},
connectAsync()
);
}
export class LibsignalWebSocketResource
extends EventTarget
implements IWebSocketResource
{
closed = false;
// Unlike WebSocketResource, libsignal will automatically attempt to keep the
// socket alive using websocket pings, so we don't need a timer-based
// keepalive mechanism. But we still send one-off keepalive requests when
// things change (see forceKeepAlive()).
private keepalive: KeepAliveSender;
constructor(
private readonly chatService: Net.ChatService,
private readonly socketIpVersion: IpVersion | undefined,
private readonly logId: string,
keepalive: KeepAliveOptionsType
) {
super();
this.keepalive = new KeepAliveSender(this, this.logId, keepalive);
}
public localPort(): number | undefined {
return undefined;
}
public ipVersion(): IpVersion | undefined {
return this.socketIpVersion;
}
public override addEventListener(
name: 'close',
handler: (ev: CloseEvent) => void
): void;
public override addEventListener(name: string, handler: EventHandler): void {
return super.addEventListener(name, handler);
}
2024-09-19 18:32:17 +00:00
public close(code = NORMAL_DISCONNECT_CODE, reason?: string): void {
if (this.closed) {
log.info(`${this.logId}.close: Already closed! ${code}/${reason}`);
return;
}
drop(this.chatService.disconnect());
// On linux the socket can wait a long time to emit its close event if we've
// lost the internet connection. On the order of minutes. This speeds that
// process up.
Timers.setTimeout(
2024-09-19 18:32:17 +00:00
() => this.onConnectionInterrupted(null),
5 * durations.SECOND
);
}
public shutdown(): void {
2024-09-19 18:32:17 +00:00
this.close(NORMAL_DISCONNECT_CODE, 'Shutdown');
}
2024-09-19 18:32:17 +00:00
onConnectionInterrupted(cause: LibSignalError | null): void {
if (this.closed) {
log.warn(
`${this.logId}.onConnectionInterrupted called after resource is closed`
);
return;
}
this.closed = true;
log.warn(`${this.logId}: connection closed`);
2024-09-19 18:32:17 +00:00
let event;
if (cause) {
event = new CloseEvent(UNEXPECTED_DISCONNECT_CODE, cause.message);
} else {
// The cause was an intentional disconnect. Report normal closure.
event = new CloseEvent(NORMAL_DISCONNECT_CODE, 'normal');
}
this.dispatchEvent(event);
}
public forceKeepAlive(timeout?: number): void {
drop(this.keepalive.send(timeout));
}
public async sendRequest(options: SendRequestOptions): Promise<Response> {
const [response] = await this.sendRequestGetDebugInfo(options);
return response;
}
public async sendRequestGetDebugInfo(
options: SendRequestOptions
): Promise<[Response, ChatServiceDebugInfo]> {
2024-08-06 21:21:15 +00:00
const { response, debugInfo } = await this.chatService.fetchAndDebug({
verb: options.verb,
path: options.path,
headers: options.headers ? options.headers : [],
body: options.body,
timeoutMillis: options.timeout,
});
return [
new Response(response.body, {
status: response.status,
statusText: response.message,
headers: [...response.headers],
}),
debugInfo,
];
}
}
export class WebSocketResourceWithShadowing implements IWebSocketResource {
private shadowing: LibsignalWebSocketResource | undefined;
private stats: AggregatedStats;
2024-07-18 17:17:56 +00:00
private statsTimer: NodeJS.Timeout;
private shadowingWithReporting: boolean;
private logId: string;
constructor(
private readonly main: WebSocketResource,
private readonly shadowingConnection: AbortableProcess<LibsignalWebSocketResource>,
options: WebSocketResourceOptions
) {
this.stats = AggregatedStats.createEmpty();
this.logId = `WebSocketResourceWithShadowing(${options.name})`;
this.statsTimer = setInterval(
() => this.updateStats(options.name),
STATS_UPDATE_INTERVAL
);
this.shadowingWithReporting =
options.transportOption === TransportOption.ShadowingHigh;
// the idea is that we want to keep the shadowing connection process
// "in the background", so that the main connection wouldn't need to wait on it.
// then when we're connected, `this.shadowing` socket resource is initialized
// or an error reported in case of connection failure
const initializeAfterConnected = async () => {
try {
this.shadowing = await shadowingConnection.resultPromise;
// checking IP one time per connection
if (this.main.ipVersion() !== this.shadowing.ipVersion()) {
this.stats.ipVersionMismatches += 1;
const mainIpType = this.main.ipVersion();
const shadowIpType = this.shadowing.ipVersion();
log.warn(
`${this.logId}: libsignal websocket IP [${shadowIpType}], Desktop websocket IP [${mainIpType}]`
);
}
} catch (error) {
this.stats.connectionFailures += 1;
}
};
drop(initializeAfterConnected());
this.addEventListener('close', (_ev): void => {
clearInterval(this.statsTimer);
this.updateStats(options.name);
});
}
private updateStats(name: string) {
const storedStats = AggregatedStats.loadOrCreateEmpty(name);
let updatedStats = AggregatedStats.add(storedStats, this.stats);
if (
this.shadowingWithReporting &&
AggregatedStats.shouldReportError(updatedStats) &&
!isProduction(window.getVersion())
) {
window.reduxActions.toast.showToast({
toastType: ToastType.TransportError,
});
log.warn(
`${this.logId}: experimental transport toast displayed, flushing transport statistics before resetting`,
updatedStats
);
updatedStats = AggregatedStats.createEmpty();
updatedStats.lastToastTimestamp = Date.now();
}
AggregatedStats.store(updatedStats, name);
this.stats = AggregatedStats.createEmpty();
}
public localPort(): number | undefined {
return this.main.localPort();
}
public addEventListener(
name: 'close',
handler: (ev: CloseEvent) => void
): void {
this.main.addEventListener(name, handler);
}
public close(code = NORMAL_DISCONNECT_CODE, reason?: string): void {
this.main.close(code, reason);
if (this.shadowing) {
this.shadowing.close(code, reason);
this.shadowing = undefined;
} else {
this.shadowingConnection.abort();
}
}
public shutdown(): void {
this.main.shutdown();
if (this.shadowing) {
this.shadowing.shutdown();
this.shadowing = undefined;
} else {
this.shadowingConnection.abort();
}
}
public forceKeepAlive(timeout?: number): void {
this.main.forceKeepAlive(timeout);
}
public async sendRequest(options: SendRequestOptions): Promise<Response> {
const responsePromise = this.main.sendRequest(options);
const response = await responsePromise;
// if we're received a response from the main channel and the status was successful,
// attempting to run a healthcheck on a libsignal transport.
if (
isSuccessfulStatusCode(response.status) &&
this.shouldSendShadowRequest()
) {
drop(this.sendShadowRequest());
}
return response;
}
private async sendShadowRequest(): Promise<void> {
// In the shadowing mode, it could be that we're either
// still connecting libsignal websocket or have already closed it.
// In those cases we're not running shadowing check.
if (!this.shadowing) {
log.info(
`${this.logId}: skipping healthcheck - websocket not connected or already closed`
);
return;
}
try {
const healthCheckResult = await this.shadowing.sendRequest({
verb: 'GET',
path: '/v1/keepalive',
timeout: KEEPALIVE_TIMEOUT_MS,
});
this.stats.requestsCompared += 1;
if (!isSuccessfulStatusCode(healthCheckResult.status)) {
this.stats.healthcheckBadStatus += 1;
log.warn(
`${this.logId}: keepalive via libsignal responded with status [${healthCheckResult.status}]`
);
}
} catch (error) {
this.stats.healthcheckFailures += 1;
log.warn(
`${this.logId}: failed to send keepalive via libsignal`,
Errors.toLogFormat(error)
);
}
}
private shouldSendShadowRequest(): boolean {
return this.shadowingWithReporting || random(0, 100) < 10;
}
}
function isSuccessfulStatusCode(status: number): boolean {
return status >= 200 && status < 300;
}
export default class WebSocketResource
extends EventTarget
implements IWebSocketResource
{
2022-03-23 20:49:27 +00:00
private outgoingId = Long.fromNumber(1, true);
private closed = false;
private readonly outgoingMap = new Map<
2022-03-23 20:49:27 +00:00
string,
(result: SendRequestResult) => void
>();
2021-07-02 19:21:24 +00:00
private readonly boundOnMessage: (message: IMessage) => void;
2022-03-23 20:49:27 +00:00
private activeRequests = new Set<IncomingWebSocketRequest | string>();
private shuttingDown = false;
private shutdownTimer?: Timers.Timeout;
2023-06-07 00:36:38 +00:00
private readonly logId: string;
private readonly localSocketPort: number | undefined;
private readonly socketIpVersion: IpVersion | undefined;
2021-07-02 19:21:24 +00:00
// Public for tests
public readonly keepalive?: KeepAlive;
constructor(
private readonly socket: WebSocket,
2023-06-07 00:36:38 +00:00
private readonly options: WebSocketResourceOptions
2021-07-02 19:21:24 +00:00
) {
super();
2023-06-07 00:36:38 +00:00
this.logId = `WebSocketResource(${options.name})`;
this.localSocketPort = socket.socket.localPort;
if (!socket.socket.localAddress) {
this.socketIpVersion = undefined;
}
if (socket.socket.localAddress == null) {
this.socketIpVersion = undefined;
} else if (net.isIPv4(socket.socket.localAddress)) {
this.socketIpVersion = IpVersion.IPv4;
} else if (net.isIPv6(socket.socket.localAddress)) {
this.socketIpVersion = IpVersion.IPv6;
} else {
this.socketIpVersion = undefined;
}
2023-06-07 00:36:38 +00:00
2021-07-02 19:21:24 +00:00
this.boundOnMessage = this.onMessage.bind(this);
2021-07-02 19:21:24 +00:00
socket.on('message', this.boundOnMessage);
2021-06-09 22:28:54 +00:00
2021-07-02 19:21:24 +00:00
if (options.keepalive) {
const keepalive = new KeepAlive(
this,
2023-06-07 00:36:38 +00:00
options.name,
options.keepalive ?? {}
2021-06-09 22:28:54 +00:00
);
2021-07-02 19:21:24 +00:00
this.keepalive = keepalive;
2021-07-02 19:21:24 +00:00
keepalive.reset();
2023-06-07 00:36:38 +00:00
socket.on('close', () => this.keepalive?.stop());
socket.on('error', (error: Error) => {
2023-06-07 00:36:38 +00:00
log.warn(`${this.logId}: WebSocket error`, Errors.toLogFormat(error));
});
}
socket.on('close', (code, reason) => {
this.closed = true;
2023-06-07 00:36:38 +00:00
log.warn(`${this.logId}: Socket closed`);
this.dispatchEvent(new CloseEvent(code, reason || 'normal'));
});
this.addEventListener('close', () => this.onClose());
2021-07-02 19:21:24 +00:00
}
public ipVersion(): IpVersion | undefined {
return this.socketIpVersion;
}
public localPort(): number | undefined {
return this.localSocketPort;
}
public override addEventListener(
2021-07-09 19:36:10 +00:00
name: 'close',
handler: (ev: CloseEvent) => void
): void;
public override addEventListener(name: string, handler: EventHandler): void {
2021-07-09 19:36:10 +00:00
return super.addEventListener(name, handler);
}
public async sendRequest(options: SendRequestOptions): Promise<Response> {
2021-07-02 19:21:24 +00:00
const id = this.outgoingId;
2022-03-23 20:49:27 +00:00
const idString = id.toString();
strictAssert(!this.outgoingMap.has(idString), 'Duplicate outgoing request');
2022-03-23 20:49:27 +00:00
// Note that this automatically wraps
this.outgoingId = this.outgoingId.add(1);
const bytes = Proto.WebSocketMessage.encode({
type: Proto.WebSocketMessage.Type.REQUEST,
request: {
verb: options.verb,
path: options.path,
body: options.body,
headers: options.headers
? options.headers
.map(([key, value]) => {
return `${key}:${value}`;
})
.slice()
: undefined,
id,
},
}).finish();
2021-09-10 20:16:53 +00:00
strictAssert(
bytes.length <= MAX_MESSAGE_SIZE,
'WebSocket request byte size exceeded'
);
strictAssert(!this.shuttingDown, 'Cannot send request, shutting down');
2022-03-23 20:49:27 +00:00
this.addActive(idString);
const promise = new Promise<SendRequestResult>((resolve, reject) => {
let timer = options.timeout
? Timers.setTimeout(() => {
2022-03-23 20:49:27 +00:00
this.removeActive(idString);
2024-09-19 18:32:17 +00:00
this.close(UNEXPECTED_DISCONNECT_CODE, 'Request timed out');
2023-05-17 16:43:33 +00:00
reject(new Error(`Request timed out; id: [${idString}]`));
}, options.timeout)
: undefined;
2022-03-23 20:49:27 +00:00
this.outgoingMap.set(idString, result => {
if (timer !== undefined) {
Timers.clearTimeout(timer);
timer = undefined;
}
2024-04-18 17:16:05 +00:00
this.keepalive?.reset();
2022-03-23 20:49:27 +00:00
this.removeActive(idString);
resolve(result);
});
});
2021-07-02 19:21:24 +00:00
this.socket.sendBytes(Buffer.from(bytes));
const requestResult = await promise;
return WebSocketResource.intoResponse(requestResult);
}
2021-06-09 22:28:54 +00:00
public forceKeepAlive(timeout?: number): void {
2021-06-09 22:28:54 +00:00
if (!this.keepalive) {
return;
}
drop(this.keepalive.send(timeout));
2021-06-09 22:28:54 +00:00
}
2021-07-02 19:21:24 +00:00
2024-09-19 18:32:17 +00:00
public close(code = NORMAL_DISCONNECT_CODE, reason?: string): void {
2021-07-02 19:21:24 +00:00
if (this.closed) {
2023-06-07 00:36:38 +00:00
log.info(`${this.logId}.close: Already closed! ${code}/${reason}`);
2021-07-02 19:21:24 +00:00
return;
}
2023-06-07 00:36:38 +00:00
log.info(`${this.logId}.close(${code})`);
2021-07-02 19:21:24 +00:00
if (this.keepalive) {
this.keepalive.stop();
}
this.socket.close(code, reason);
this.socket.removeListener('message', this.boundOnMessage);
// On linux the socket can wait a long time to emit its close event if we've
// lost the internet connection. On the order of minutes. This speeds that
// process up.
Timers.setTimeout(() => {
2021-07-02 19:21:24 +00:00
if (this.closed) {
return;
}
2023-06-07 00:36:38 +00:00
log.warn(`${this.logId}.close: Dispatching our own socket close event`);
2021-07-09 19:36:10 +00:00
this.dispatchEvent(new CloseEvent(code, reason || 'normal'));
2023-06-07 00:36:38 +00:00
}, 5 * durations.SECOND);
2021-07-02 19:21:24 +00:00
}
public shutdown(): void {
if (this.closed) {
return;
}
if (this.activeRequests.size === 0) {
2023-06-07 00:36:38 +00:00
log.info(`${this.logId}.shutdown: no active requests, closing`);
2024-09-19 18:32:17 +00:00
this.close(NORMAL_DISCONNECT_CODE, 'Shutdown');
return;
}
this.shuttingDown = true;
2023-06-07 00:36:38 +00:00
log.info(`${this.logId}.shutdown: shutting down`);
this.shutdownTimer = Timers.setTimeout(() => {
if (this.closed) {
return;
}
2023-06-07 00:36:38 +00:00
log.warn(`${this.logId}.shutdown: Failed to shutdown gracefully`);
2024-09-19 18:32:17 +00:00
this.close(NORMAL_DISCONNECT_CODE, 'Shutdown');
}, THIRTY_SECONDS);
}
2021-07-02 19:21:24 +00:00
private onMessage({ type, binaryData }: IMessage): void {
if (type !== 'binary' || !binaryData) {
throw new Error(`Unsupported websocket message type: ${type}`);
}
const message = Proto.WebSocketMessage.decode(binaryData);
if (
message.type === Proto.WebSocketMessage.Type.REQUEST &&
message.request
) {
const handleRequest =
this.options.handleRequest ||
(request => request.respond(404, 'Not found'));
2024-08-06 21:21:15 +00:00
const incomingRequest = new IncomingWebSocketRequestLegacy(
message.request,
(bytes: Buffer): void => {
this.removeActive(incomingRequest);
strictAssert(
bytes.length <= MAX_MESSAGE_SIZE,
'WebSocket response byte size exceeded'
);
this.socket.sendBytes(bytes);
}
);
if (this.shuttingDown) {
incomingRequest.respond(-1, 'Shutting down');
return;
}
this.addActive(incomingRequest);
handleRequest(incomingRequest);
2021-07-02 19:21:24 +00:00
} else if (
message.type === Proto.WebSocketMessage.Type.RESPONSE &&
message.response
) {
const { response } = message;
strictAssert(response.id, 'response without id');
2022-03-23 20:49:27 +00:00
const responseIdString = response.id.toString();
const resolve = this.outgoingMap.get(responseIdString);
this.outgoingMap.delete(responseIdString);
2021-07-02 19:21:24 +00:00
if (!resolve) {
2022-03-23 20:49:27 +00:00
throw new Error(`Received response for unknown request ${response.id}`);
2021-07-02 19:21:24 +00:00
}
resolve({
status: response.status ?? -1,
message: response.message ?? '',
response: dropNull(response.body),
headers: response.headers ?? [],
});
}
}
2021-07-02 19:21:24 +00:00
private onClose(): void {
const outgoing = new Map(this.outgoingMap);
this.outgoingMap.clear();
for (const resolve of outgoing.values()) {
resolve({
status: -1,
message: 'Connection closed',
response: undefined,
headers: [],
});
}
}
2021-07-02 19:21:24 +00:00
2022-03-23 20:49:27 +00:00
private addActive(request: IncomingWebSocketRequest | string): void {
this.activeRequests.add(request);
}
2021-07-02 19:21:24 +00:00
2022-03-23 20:49:27 +00:00
private removeActive(request: IncomingWebSocketRequest | string): void {
if (!this.activeRequests.has(request)) {
2023-06-07 00:36:38 +00:00
log.warn(`${this.logId}.removeActive: removing unknown request`);
return;
}
this.activeRequests.delete(request);
if (this.activeRequests.size !== 0) {
return;
}
if (!this.shuttingDown) {
return;
}
if (this.shutdownTimer) {
Timers.clearTimeout(this.shutdownTimer);
this.shutdownTimer = undefined;
2021-07-02 19:21:24 +00:00
}
2023-06-07 00:36:38 +00:00
log.info(`${this.logId}.removeActive: shutdown complete`);
2024-09-19 18:32:17 +00:00
this.close(NORMAL_DISCONNECT_CODE, 'Shutdown');
2021-07-02 19:21:24 +00:00
}
private static intoResponse(sendRequestResult: SendRequestResult): Response {
const {
status,
message: statusText,
response,
headers: flatResponseHeaders,
} = sendRequestResult;
const headers: Array<[string, string]> = flatResponseHeaders.map(header => {
const [key, value] = header.split(':', 2);
strictAssert(value !== undefined, 'Invalid header!');
return [key, value];
});
return new Response(response, {
status,
statusText,
headers,
});
}
}
2021-07-02 19:21:24 +00:00
export type KeepAliveOptionsType = {
path?: string;
};
// 30 seconds + 5 seconds for closing the socket above.
const KEEPALIVE_INTERVAL_MS = 30 * durations.SECOND;
2023-06-07 00:36:38 +00:00
// If the machine was in suspended mode for more than 5 minutes - trigger
// immediate disconnect.
const STALE_THRESHOLD_MS = 5 * durations.MINUTE;
// If we don't receive a response to keepalive request within 30 seconds -
2023-06-07 00:36:38 +00:00
// close the socket.
const KEEPALIVE_TIMEOUT_MS = 30 * durations.SECOND;
2023-06-07 00:36:38 +00:00
2023-05-17 16:43:33 +00:00
const LOG_KEEPALIVE_AFTER_MS = 500;
/**
* References an {@link IWebSocketResource} and a request path that should
* return promptly to determine whether the connection is still alive.
*
* The response to the request must have a 2xx status code but is otherwise
* ignored. A failing response or a timeout results in the socket being closed
* with {@link UNEXPECTED_DISCONNECT_CODE}.
*
* Use the subclass {@link KeepAlive} if you want to send the request at regular
* intervals.
*/
class KeepAliveSender {
2021-06-09 22:28:54 +00:00
private path: string;
protected wsr: IWebSocketResource;
protected logId: string;
2023-06-07 00:36:38 +00:00
constructor(
websocketResource: IWebSocketResource,
2023-06-07 00:36:38 +00:00
name: string,
opts: KeepAliveOptionsType = {}
) {
2023-06-07 00:36:38 +00:00
this.logId = `WebSocketResources.KeepAlive(${name})`;
this.path = opts.path ?? '/';
this.wsr = websocketResource;
}
public async send(timeout = KEEPALIVE_TIMEOUT_MS): Promise<boolean> {
2023-06-07 00:36:38 +00:00
log.info(`${this.logId}.send: Sending a keepalive message`);
2023-05-17 16:43:33 +00:00
const sentAt = Date.now();
2023-06-07 00:36:38 +00:00
try {
const { status } = await pTimeout(
this.wsr.sendRequest({
verb: 'GET',
path: this.path,
}),
timeout
2023-06-07 00:36:38 +00:00
);
if (status < 200 || status >= 300) {
log.warn(`${this.logId}.send: keepalive response status ${status}`);
2024-09-19 18:32:17 +00:00
this.wsr.close(
UNEXPECTED_DISCONNECT_CODE,
`keepalive response with ${status} code`
);
return false;
2023-06-07 00:36:38 +00:00
}
} catch (error) {
2024-09-19 18:32:17 +00:00
this.wsr.close(
UNEXPECTED_DISCONNECT_CODE,
'No response to keepalive request'
);
return false;
2023-06-07 00:36:38 +00:00
}
2023-05-17 16:43:33 +00:00
const responseTime = Date.now() - sentAt;
if (responseTime > LOG_KEEPALIVE_AFTER_MS) {
log.warn(
2023-06-07 00:36:38 +00:00
`${this.logId}.send: delayed response to keepalive request, ` +
`response time: ${responseTime}ms`
2023-05-17 16:43:33 +00:00
);
}
return true;
}
}
/**
* Manages a timer that checks if a particular {@link WebSocketResource} is
* still alive.
*
* The resource must specifically be a {@link WebSocketResource}. Other kinds of
* resource are expected to manage their own liveness checks. If you want to
* manually send keepalive requests to such resources, use the base class
* {@link KeepAliveSender}.
*/
class KeepAlive extends KeepAliveSender {
private keepAliveTimer: Timers.Timeout | undefined;
private lastAliveAt: number = Date.now();
constructor(
websocketResource: WebSocketResource,
name: string,
opts: KeepAliveOptionsType = {}
) {
super(websocketResource, name, opts);
}
public stop(): void {
this.clearTimers();
}
public override async send(timeout = KEEPALIVE_TIMEOUT_MS): Promise<boolean> {
this.clearTimers();
const isStale = isOlderThan(this.lastAliveAt, STALE_THRESHOLD_MS);
if (isStale) {
log.info(`${this.logId}.send: disconnecting due to stale state`);
this.wsr.close(
UNEXPECTED_DISCONNECT_CODE,
`Last keepalive request was too far in the past: ${this.lastAliveAt}`
);
return false;
}
const isAlive = await super.send(timeout);
if (!isAlive) {
return false;
}
2023-06-07 00:36:38 +00:00
// Successful response on time
this.reset();
return true;
2021-06-09 22:28:54 +00:00
}
public reset(): void {
this.lastAliveAt = Date.now();
2021-06-09 22:28:54 +00:00
this.clearTimers();
this.keepAliveTimer = Timers.setTimeout(
() => this.send(),
KEEPALIVE_INTERVAL_MS
);
2021-06-09 22:28:54 +00:00
}
private clearTimers(): void {
if (this.keepAliveTimer) {
Timers.clearTimeout(this.keepAliveTimer);
2021-06-09 22:28:54 +00:00
this.keepAliveTimer = undefined;
}
}
}