libsignal authenticated websocket
This commit is contained in:
Sergey Skrobotov
GitHub
parent
31bcb1e4cc
commit
de33410be1
10 changed files with 470 additions and 286 deletions
|
|
@ -26,6 +26,7 @@ export type ConfigKeyType =
|
|||
| 'desktop.retryRespondMaxAge'
|
||||
| 'desktop.senderKey.retry'
|
||||
| 'desktop.senderKeyMaxAge'
|
||||
| 'desktop.experimentalTransport.enableAuth'
|
||||
| 'desktop.experimentalTransportEnabled.alpha'
|
||||
| 'desktop.experimentalTransportEnabled.beta'
|
||||
| 'desktop.experimentalTransportEnabled.prod'
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import { assert } from 'chai';
|
|||
import Long from 'long';
|
||||
|
||||
import MessageReceiver from '../textsecure/MessageReceiver';
|
||||
import { IncomingWebSocketRequest } from '../textsecure/WebsocketResources';
|
||||
import { IncomingWebSocketRequestLegacy } from '../textsecure/WebsocketResources';
|
||||
import type { WebAPIType } from '../textsecure/WebAPI';
|
||||
import type { DecryptionErrorEvent } from '../textsecure/messageReceiverEvents';
|
||||
import { generateAci } from '../types/ServiceId';
|
||||
|
|
@ -53,7 +53,7 @@ describe('MessageReceiver', () => {
|
|||
}).finish();
|
||||
|
||||
messageReceiver.handleRequest(
|
||||
new IncomingWebSocketRequest(
|
||||
new IncomingWebSocketRequestLegacy(
|
||||
{
|
||||
id: Long.fromNumber(1),
|
||||
verb: 'PUT',
|
||||
|
|
|
|||
|
|
@ -16,7 +16,9 @@ import Long from 'long';
|
|||
import { dropNull } from '../util/dropNull';
|
||||
import { SignalService as Proto } from '../protobuf';
|
||||
|
||||
import WebSocketResource from '../textsecure/WebsocketResources';
|
||||
import WebSocketResource, {
|
||||
ServerRequestType,
|
||||
} from '../textsecure/WebsocketResources';
|
||||
|
||||
describe('WebSocket-Resource', () => {
|
||||
class FakeSocket extends EventEmitter {
|
||||
|
|
@ -72,8 +74,7 @@ describe('WebSocket-Resource', () => {
|
|||
new WebSocketResource(socket as WebSocket, {
|
||||
name: 'test',
|
||||
handleRequest(request: any) {
|
||||
assert.strictEqual(request.verb, 'PUT');
|
||||
assert.strictEqual(request.path, '/some/path');
|
||||
assert.strictEqual(request.requestType, ServerRequestType.ApiMessage);
|
||||
assert.deepEqual(request.body, new Uint8Array([1, 2, 3]));
|
||||
request.respond(200, 'OK');
|
||||
},
|
||||
|
|
@ -87,7 +88,7 @@ describe('WebSocket-Resource', () => {
|
|||
request: {
|
||||
id: requestId,
|
||||
verb: 'PUT',
|
||||
path: '/some/path',
|
||||
path: ServerRequestType.ApiMessage.toString(),
|
||||
body: new Uint8Array([1, 2, 3]),
|
||||
},
|
||||
}).finish(),
|
||||
|
|
|
|||
|
|
@ -15,39 +15,40 @@ import type {
|
|||
WebAPIType,
|
||||
} from './WebAPI';
|
||||
import type {
|
||||
CompatSignedPreKeyType,
|
||||
CompatPreKeyType,
|
||||
CompatSignedPreKeyType,
|
||||
KeyPairType,
|
||||
KyberPreKeyType,
|
||||
PniKeyMaterialType,
|
||||
} from './Types.d';
|
||||
import ProvisioningCipher from './ProvisioningCipher';
|
||||
import type { IncomingWebSocketRequest } from './WebsocketResources';
|
||||
import { ServerRequestType } from './WebsocketResources';
|
||||
import createTaskWithTimeout from './TaskWithTimeout';
|
||||
import * as Bytes from '../Bytes';
|
||||
import * as Errors from '../types/errors';
|
||||
import { senderCertificateService } from '../services/senderCertificate';
|
||||
import { backupsService } from '../services/backups';
|
||||
import {
|
||||
decryptDeviceName,
|
||||
deriveAccessKey,
|
||||
deriveStorageServiceKey,
|
||||
encryptDeviceName,
|
||||
generateRegistrationId,
|
||||
getRandomBytes,
|
||||
decryptDeviceName,
|
||||
encryptDeviceName,
|
||||
deriveStorageServiceKey,
|
||||
} from '../Crypto';
|
||||
import {
|
||||
generateKeyPair,
|
||||
generateSignedPreKey,
|
||||
generatePreKey,
|
||||
generateKyberPreKey,
|
||||
generatePreKey,
|
||||
generateSignedPreKey,
|
||||
} from '../Curve';
|
||||
import type { ServiceIdString, AciString, PniString } from '../types/ServiceId';
|
||||
import type { AciString, PniString, ServiceIdString } from '../types/ServiceId';
|
||||
import {
|
||||
ServiceIdKind,
|
||||
normalizePni,
|
||||
toTaggedPni,
|
||||
isUntaggedPniString,
|
||||
normalizePni,
|
||||
ServiceIdKind,
|
||||
toTaggedPni,
|
||||
} from '../types/ServiceId';
|
||||
import { normalizeAci } from '../util/normalizeAci';
|
||||
import { drop } from '../util/drop';
|
||||
|
|
@ -367,8 +368,7 @@ export default class AccountManager extends EventTarget {
|
|||
const wsr = await this.server.getProvisioningResource({
|
||||
handleRequest(request: IncomingWebSocketRequest) {
|
||||
if (
|
||||
request.path === '/v1/address' &&
|
||||
request.verb === 'PUT' &&
|
||||
request.requestType === ServerRequestType.ProvisioningAddress &&
|
||||
request.body
|
||||
) {
|
||||
const proto = Proto.ProvisioningUuid.decode(request.body);
|
||||
|
|
@ -388,8 +388,7 @@ export default class AccountManager extends EventTarget {
|
|||
setProvisioningUrl(url);
|
||||
request.respond(200, 'OK');
|
||||
} else if (
|
||||
request.path === '/v1/message' &&
|
||||
request.verb === 'PUT' &&
|
||||
request.requestType === ServerRequestType.ProvisioningMessage &&
|
||||
request.body
|
||||
) {
|
||||
const envelope = Proto.ProvisionEnvelope.decode(request.body);
|
||||
|
|
@ -397,7 +396,7 @@ export default class AccountManager extends EventTarget {
|
|||
wsr.close();
|
||||
envelopeCallbacks?.resolve(envelope);
|
||||
} else {
|
||||
log.error('Unknown websocket message', request.path);
|
||||
log.error('Unknown websocket message', request.requestType);
|
||||
}
|
||||
},
|
||||
});
|
||||
|
|
|
|||
|
|
@ -13,13 +13,13 @@ import type {
|
|||
UnidentifiedSenderMessageContent,
|
||||
} from '@signalapp/libsignal-client';
|
||||
import {
|
||||
ContentHint,
|
||||
CiphertextMessageType,
|
||||
ContentHint,
|
||||
DecryptionErrorMessage,
|
||||
groupDecrypt,
|
||||
PlaintextContent,
|
||||
PreKeySignalMessage,
|
||||
Pni,
|
||||
PreKeySignalMessage,
|
||||
processSenderKeyDistributionMessage,
|
||||
ProtocolAddress,
|
||||
PublicKey,
|
||||
|
|
@ -40,7 +40,7 @@ import {
|
|||
SignedPreKeys,
|
||||
} from '../LibSignalStores';
|
||||
import { verifySignature } from '../Curve';
|
||||
import { strictAssert, assertDev } from '../util/assert';
|
||||
import { assertDev, strictAssert } from '../util/assert';
|
||||
import type { BatcherType } from '../util/batcher';
|
||||
import { createBatcher } from '../util/batcher';
|
||||
import { drop } from '../util/drop';
|
||||
|
|
@ -48,6 +48,7 @@ import { dropNull } from '../util/dropNull';
|
|||
import { parseIntOrThrow } from '../util/parseIntOrThrow';
|
||||
import { clearTimeoutIfNecessary } from '../util/clearTimeoutIfNecessary';
|
||||
import { Zone } from '../util/Zone';
|
||||
import * as durations from '../util/durations';
|
||||
import { DurationInSeconds, SECOND } from '../util/durations';
|
||||
import type { AttachmentType } from '../types/Attachment';
|
||||
import { Address } from '../types/Address';
|
||||
|
|
@ -55,13 +56,13 @@ import { QualifiedAddress } from '../types/QualifiedAddress';
|
|||
import { normalizeStoryDistributionId } from '../types/StoryDistributionId';
|
||||
import type { ServiceIdString } from '../types/ServiceId';
|
||||
import {
|
||||
ServiceIdKind,
|
||||
normalizeServiceId,
|
||||
normalizePni,
|
||||
isPniString,
|
||||
isUntaggedPniString,
|
||||
isServiceIdString,
|
||||
fromPniObject,
|
||||
isPniString,
|
||||
isServiceIdString,
|
||||
isUntaggedPniString,
|
||||
normalizePni,
|
||||
normalizeServiceId,
|
||||
ServiceIdKind,
|
||||
toTaggedPni,
|
||||
} from '../types/ServiceId';
|
||||
import { normalizeAci } from '../util/normalizeAci';
|
||||
|
|
@ -75,70 +76,70 @@ import createTaskWithTimeout from './TaskWithTimeout';
|
|||
import {
|
||||
processAttachment,
|
||||
processDataMessage,
|
||||
processPreview,
|
||||
processGroupV2Context,
|
||||
processPreview,
|
||||
} from './processDataMessage';
|
||||
import { processSyncMessage } from './processSyncMessage';
|
||||
import type { EventHandler } from './EventTarget';
|
||||
import EventTarget from './EventTarget';
|
||||
import { downloadAttachment } from './downloadAttachment';
|
||||
import type { IncomingWebSocketRequest } from './WebsocketResources';
|
||||
import { ServerRequestType } from './WebsocketResources';
|
||||
import { parseContactsV2 } from './ContactsParser';
|
||||
import type { WebAPIType } from './WebAPI';
|
||||
import type { Storage } from './Storage';
|
||||
import { WarnOnlyError } from './Errors';
|
||||
import * as Bytes from '../Bytes';
|
||||
import type {
|
||||
IRequestHandler,
|
||||
ProcessedAttachment,
|
||||
ProcessedDataMessage,
|
||||
ProcessedPreview,
|
||||
ProcessedSyncMessage,
|
||||
ProcessedSent,
|
||||
ProcessedEnvelope,
|
||||
IRequestHandler,
|
||||
ProcessedPreview,
|
||||
ProcessedSent,
|
||||
ProcessedSyncMessage,
|
||||
UnprocessedType,
|
||||
} from './Types.d';
|
||||
import type {
|
||||
ConversationToDelete,
|
||||
DeleteForMeSyncEventData,
|
||||
DeleteForMeSyncTarget,
|
||||
MessageToDelete,
|
||||
ReadSyncEventData,
|
||||
ViewSyncEventData,
|
||||
} from './messageReceiverEvents';
|
||||
import {
|
||||
CallEventSyncEvent,
|
||||
CallLinkUpdateSyncEvent,
|
||||
CallLogEventSyncEvent,
|
||||
ConfigurationEvent,
|
||||
ContactSyncEvent,
|
||||
DecryptionErrorEvent,
|
||||
DeleteForMeSyncEvent,
|
||||
DeliveryEvent,
|
||||
EmptyEvent,
|
||||
EnvelopeQueuedEvent,
|
||||
EnvelopeUnsealedEvent,
|
||||
ProgressEvent,
|
||||
TypingEvent,
|
||||
ErrorEvent,
|
||||
DeliveryEvent,
|
||||
DecryptionErrorEvent,
|
||||
SentEvent,
|
||||
ProfileKeyUpdateEvent,
|
||||
InvalidPlaintextEvent,
|
||||
MessageEvent,
|
||||
RetryRequestEvent,
|
||||
ReadEvent,
|
||||
ViewEvent,
|
||||
ConfigurationEvent,
|
||||
ViewOnceOpenSyncEvent,
|
||||
MessageRequestResponseEvent,
|
||||
FetchLatestEvent,
|
||||
InvalidPlaintextEvent,
|
||||
KeysEvent,
|
||||
StickerPackEvent,
|
||||
MessageEvent,
|
||||
MessageRequestResponseEvent,
|
||||
ProfileKeyUpdateEvent,
|
||||
ProgressEvent,
|
||||
ReadEvent,
|
||||
ReadSyncEvent,
|
||||
ViewSyncEvent,
|
||||
ContactSyncEvent,
|
||||
RetryRequestEvent,
|
||||
SentEvent,
|
||||
StickerPackEvent,
|
||||
StoryRecipientUpdateEvent,
|
||||
CallLogEventSyncEvent,
|
||||
CallLinkUpdateSyncEvent,
|
||||
DeleteForMeSyncEvent,
|
||||
} from './messageReceiverEvents';
|
||||
import type {
|
||||
MessageToDelete,
|
||||
DeleteForMeSyncEventData,
|
||||
DeleteForMeSyncTarget,
|
||||
ConversationToDelete,
|
||||
ViewSyncEventData,
|
||||
ReadSyncEventData,
|
||||
TypingEvent,
|
||||
ViewEvent,
|
||||
ViewOnceOpenSyncEvent,
|
||||
ViewSyncEvent,
|
||||
} from './messageReceiverEvents';
|
||||
import * as log from '../logging/log';
|
||||
import * as durations from '../util/durations';
|
||||
import { areArraysMatchingSets } from '../util/areArraysMatchingSets';
|
||||
import { generateBlurHash } from '../util/generateBlurHash';
|
||||
import { TEXT_ATTACHMENT } from '../types/MIME';
|
||||
|
|
@ -385,11 +386,11 @@ export default class MessageReceiver
|
|||
public handleRequest(request: IncomingWebSocketRequest): void {
|
||||
// We do the message decryption here, instead of in the ordered pending queue,
|
||||
// to avoid exposing the time it took us to process messages through the time-to-ack.
|
||||
log.info('MessageReceiver: got request', request.verb, request.path);
|
||||
if (request.path !== '/api/v1/message') {
|
||||
log.info('MessageReceiver: got request', request.requestType);
|
||||
if (request.requestType !== ServerRequestType.ApiMessage) {
|
||||
request.respond(200, 'OK');
|
||||
|
||||
if (request.verb === 'PUT' && request.path === '/api/v1/queue/empty') {
|
||||
if (request.requestType === ServerRequestType.ApiEmptyQueue) {
|
||||
drop(
|
||||
this.incomingQueue.add(
|
||||
createTaskWithTimeout(
|
||||
|
|
@ -406,8 +407,6 @@ export default class MessageReceiver
|
|||
}
|
||||
|
||||
const job = async () => {
|
||||
const headers = request.headers || [];
|
||||
|
||||
if (!request.body) {
|
||||
throw new Error(
|
||||
'MessageReceiver.handleRequest: request.body was falsey!'
|
||||
|
|
@ -429,7 +428,10 @@ export default class MessageReceiver
|
|||
receivedAtCounter: incrementMessageCounter(),
|
||||
receivedAtDate: Date.now(),
|
||||
// Calculate the message age (time on server).
|
||||
messageAgeSec: this.calculateMessageAge(headers, serverTimestamp),
|
||||
messageAgeSec: this.calculateMessageAge(
|
||||
request.timestamp,
|
||||
serverTimestamp
|
||||
),
|
||||
|
||||
// Proto.Envelope fields
|
||||
type: decoded.type ?? Proto.Envelope.Type.UNKNOWN,
|
||||
|
|
@ -728,32 +730,15 @@ export default class MessageReceiver
|
|||
}
|
||||
|
||||
private calculateMessageAge(
|
||||
headers: ReadonlyArray<string>,
|
||||
serverTimestamp?: number
|
||||
timestamp: number | undefined,
|
||||
serverTimestamp: number | undefined
|
||||
): number {
|
||||
let messageAgeSec = 0; // Default to 0 in case of unreliable parameters.
|
||||
|
||||
if (serverTimestamp) {
|
||||
// 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) {
|
||||
const timestamp = Number(match[1]);
|
||||
|
||||
// One final sanity check, the timestamp when a message is pulled from
|
||||
// the server should be later than when it was pushed.
|
||||
if (timestamp > serverTimestamp) {
|
||||
messageAgeSec = Math.floor((timestamp - serverTimestamp) / 1000);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return messageAgeSec;
|
||||
// Default to 0 in case of unreliable parameters.
|
||||
// One final sanity check, the timestamp when a message is pulled from
|
||||
// the server should be later than when it was pushed.
|
||||
return serverTimestamp && timestamp && timestamp > serverTimestamp
|
||||
? Math.floor((timestamp - serverTimestamp) / 1000)
|
||||
: 0;
|
||||
}
|
||||
|
||||
private async addToQueue<T>(
|
||||
|
|
|
|||
|
|
@ -13,14 +13,14 @@ import { AbortableProcess } from '../util/AbortableProcess';
|
|||
import { strictAssert } from '../util/assert';
|
||||
import {
|
||||
BackOff,
|
||||
FIBONACCI_TIMEOUTS,
|
||||
EXTENDED_FIBONACCI_TIMEOUTS,
|
||||
FIBONACCI_TIMEOUTS,
|
||||
} from '../util/BackOff';
|
||||
import * as durations from '../util/durations';
|
||||
import { sleep } from '../util/sleep';
|
||||
import { drop } from '../util/drop';
|
||||
import { createProxyAgent } from '../util/createProxyAgent';
|
||||
import type { ProxyAgent } from '../util/createProxyAgent';
|
||||
import { createProxyAgent } from '../util/createProxyAgent';
|
||||
import { SocketStatus } from '../types/SocketStatus';
|
||||
import * as Errors from '../types/errors';
|
||||
import * as Bytes from '../Bytes';
|
||||
|
|
@ -32,7 +32,8 @@ import type {
|
|||
WebSocketResourceOptions,
|
||||
} from './WebsocketResources';
|
||||
import WebSocketResource, {
|
||||
LibsignalWebSocketResource,
|
||||
connectAuthenticatedLibsignal,
|
||||
connectUnauthenticatedLibsignal,
|
||||
TransportOption,
|
||||
WebSocketResourceWithShadowing,
|
||||
} from './WebsocketResources';
|
||||
|
|
@ -166,22 +167,38 @@ export class SocketManager extends EventListener {
|
|||
|
||||
this.setStatus(SocketStatus.CONNECTING);
|
||||
|
||||
const process = this.connectResource({
|
||||
name: AUTHENTICATED_CHANNEL_NAME,
|
||||
path: '/v1/websocket/',
|
||||
query: { login: username, password },
|
||||
proxyAgent: await this.getProxyAgent(),
|
||||
resourceOptions: {
|
||||
name: AUTHENTICATED_CHANNEL_NAME,
|
||||
keepalive: { path: '/v1/keepalive' },
|
||||
handleRequest: (req: IncomingWebSocketRequest): void => {
|
||||
this.queueOrHandleRequest(req);
|
||||
},
|
||||
},
|
||||
extraHeaders: {
|
||||
'X-Signal-Receive-Stories': String(!this.hasStoriesDisabled),
|
||||
},
|
||||
});
|
||||
const proxyAgent = await this.getProxyAgent();
|
||||
const useLibsignalTransport =
|
||||
window.Signal.RemoteConfig.isEnabled(
|
||||
'desktop.experimentalTransport.enableAuth'
|
||||
) && this.transportOption(proxyAgent) === TransportOption.Libsignal;
|
||||
|
||||
const process = useLibsignalTransport
|
||||
? connectAuthenticatedLibsignal({
|
||||
libsignalNet: this.libsignalNet,
|
||||
name: AUTHENTICATED_CHANNEL_NAME,
|
||||
credentials: this.credentials,
|
||||
handler: (req: IncomingWebSocketRequest): void => {
|
||||
this.queueOrHandleRequest(req);
|
||||
},
|
||||
receiveStories: !this.hasStoriesDisabled,
|
||||
})
|
||||
: this.connectResource({
|
||||
name: AUTHENTICATED_CHANNEL_NAME,
|
||||
path: '/v1/websocket/',
|
||||
query: { login: username, password },
|
||||
resourceOptions: {
|
||||
name: AUTHENTICATED_CHANNEL_NAME,
|
||||
keepalive: { path: '/v1/keepalive' },
|
||||
handleRequest: (req: IncomingWebSocketRequest): void => {
|
||||
this.queueOrHandleRequest(req);
|
||||
},
|
||||
},
|
||||
extraHeaders: {
|
||||
'X-Signal-Receive-Stories': String(!this.hasStoriesDisabled),
|
||||
},
|
||||
proxyAgent,
|
||||
});
|
||||
|
||||
// Cancel previous connect attempt or close socket
|
||||
this.authenticated?.abort();
|
||||
|
|
@ -575,10 +592,10 @@ export class SocketManager extends EventListener {
|
|||
}
|
||||
|
||||
private connectLibsignalUnauthenticated(): AbortableProcess<IWebSocketResource> {
|
||||
return LibsignalWebSocketResource.connect(
|
||||
this.libsignalNet,
|
||||
UNAUTHENTICATED_CHANNEL_NAME
|
||||
);
|
||||
return connectUnauthenticatedLibsignal({
|
||||
libsignalNet: this.libsignalNet,
|
||||
name: UNAUTHENTICATED_CHANNEL_NAME,
|
||||
});
|
||||
}
|
||||
|
||||
private async getUnauthenticatedResource(): Promise<IWebSocketResource> {
|
||||
|
|
@ -724,10 +741,10 @@ export class SocketManager extends EventListener {
|
|||
options: WebSocketResourceOptions
|
||||
): AbortableProcess<IWebSocketResource> {
|
||||
// creating an `AbortableProcess` of libsignal websocket connection
|
||||
const shadowingConnection = LibsignalWebSocketResource.connect(
|
||||
this.libsignalNet,
|
||||
options.name
|
||||
);
|
||||
const shadowingConnection = connectUnauthenticatedLibsignal({
|
||||
libsignalNet: this.libsignalNet,
|
||||
name: options.name,
|
||||
});
|
||||
const shadowWrapper = async () => {
|
||||
// if main connection results in an error,
|
||||
// it's propagated as the error of the resulting process
|
||||
|
|
|
|||
|
|
@ -37,6 +37,8 @@ import { random } from 'lodash';
|
|||
import type { ChatServiceDebugInfo } from '@signalapp/libsignal-client/Native';
|
||||
|
||||
import type { Net } from '@signalapp/libsignal-client';
|
||||
import { Buffer } from 'node:buffer';
|
||||
import type { ChatServerMessageAck } from '@signalapp/libsignal-client/dist/net';
|
||||
import type { EventHandler } from './EventTarget';
|
||||
import EventTarget from './EventTarget';
|
||||
|
||||
|
|
@ -54,6 +56,7 @@ import { isProduction } from '../util/version';
|
|||
|
||||
import { ToastType } from '../types/Toast';
|
||||
import { AbortableProcess } from '../util/AbortableProcess';
|
||||
import type { WebAPICredentials } from './Types';
|
||||
|
||||
const THIRTY_SECONDS = 30 * durations.SECOND;
|
||||
|
||||
|
|
@ -166,16 +169,49 @@ export namespace AggregatedStats {
|
|||
}
|
||||
}
|
||||
|
||||
export class IncomingWebSocketRequest {
|
||||
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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export class IncomingWebSocketRequestLegacy
|
||||
implements IncomingWebSocketRequest
|
||||
{
|
||||
private readonly id: Long;
|
||||
|
||||
public readonly verb: string;
|
||||
|
||||
public readonly path: string;
|
||||
public readonly requestType: ServerRequestType;
|
||||
|
||||
public readonly body: Uint8Array | undefined;
|
||||
|
||||
public readonly headers: ReadonlyArray<string>;
|
||||
public readonly timestamp: number | undefined;
|
||||
|
||||
constructor(
|
||||
request: Proto.IWebSocketRequestMessage,
|
||||
|
|
@ -186,10 +222,9 @@ export class IncomingWebSocketRequest {
|
|||
strictAssert(request.path, 'request without path');
|
||||
|
||||
this.id = request.id;
|
||||
this.verb = request.verb;
|
||||
this.path = request.path;
|
||||
this.requestType = resolveType(request.path, request.verb);
|
||||
this.body = dropNull(request.body);
|
||||
this.headers = request.headers || [];
|
||||
this.timestamp = resolveTimestamp(request.headers || []);
|
||||
}
|
||||
|
||||
public respond(status: number, message: string): void {
|
||||
|
|
@ -202,6 +237,35 @@ export class IncomingWebSocketRequest {
|
|||
}
|
||||
}
|
||||
|
||||
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;
|
||||
|
|
@ -266,6 +330,100 @@ export interface IWebSocketResource extends IResource {
|
|||
localPort(): number | undefined;
|
||||
}
|
||||
|
||||
export function connectUnauthenticatedLibsignal({
|
||||
libsignalNet,
|
||||
name,
|
||||
}: {
|
||||
libsignalNet: Net.Net;
|
||||
name: string;
|
||||
}): AbortableProcess<LibsignalWebSocketResource> {
|
||||
return connectLibsignal(libsignalNet.newUnauthenticatedChatService(), name);
|
||||
}
|
||||
|
||||
export function connectAuthenticatedLibsignal({
|
||||
libsignalNet,
|
||||
name,
|
||||
credentials,
|
||||
handler,
|
||||
receiveStories,
|
||||
}: {
|
||||
libsignalNet: Net.Net;
|
||||
name: string;
|
||||
credentials: WebAPICredentials;
|
||||
handler: (request: IncomingWebSocketRequest) => void;
|
||||
receiveStories: boolean;
|
||||
}): AbortableProcess<LibsignalWebSocketResource> {
|
||||
const listener = {
|
||||
onIncomingMessage(
|
||||
envelope: Buffer,
|
||||
timestamp: number,
|
||||
ack: ChatServerMessageAck
|
||||
): void {
|
||||
const request = new IncomingWebSocketRequestLibsignal(
|
||||
ServerRequestType.ApiMessage,
|
||||
envelope,
|
||||
timestamp,
|
||||
ack
|
||||
);
|
||||
handler(request);
|
||||
},
|
||||
onQueueEmpty(): void {
|
||||
const request = new IncomingWebSocketRequestLibsignal(
|
||||
ServerRequestType.ApiEmptyQueue,
|
||||
undefined,
|
||||
undefined,
|
||||
undefined
|
||||
);
|
||||
handler(request);
|
||||
},
|
||||
onConnectionInterrupted(): void {
|
||||
log.warn(`LibsignalWebSocketResource(${name}): connection interrupted`);
|
||||
},
|
||||
};
|
||||
return connectLibsignal(
|
||||
libsignalNet.newAuthenticatedChatService(
|
||||
credentials.username,
|
||||
credentials.password,
|
||||
receiveStories,
|
||||
listener
|
||||
),
|
||||
name
|
||||
);
|
||||
}
|
||||
|
||||
function connectLibsignal(
|
||||
chatService: Net.ChatService,
|
||||
name: string
|
||||
): AbortableProcess<LibsignalWebSocketResource> {
|
||||
const connectAsync = async () => {
|
||||
try {
|
||||
const debugInfo = await chatService.connect();
|
||||
log.info(`LibsignalWebSocketResource(${name}) connected`, debugInfo);
|
||||
return new LibsignalWebSocketResource(
|
||||
chatService,
|
||||
IpVersion.fromDebugInfoCode(debugInfo.ipType)
|
||||
);
|
||||
} catch (error) {
|
||||
// Handle any errors that occur during connection
|
||||
log.error(
|
||||
`LibsignalWebSocketResource(${name}) connection failed`,
|
||||
Errors.toLogFormat(error)
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
return new AbortableProcess<LibsignalWebSocketResource>(
|
||||
`LibsignalWebSocketResource.connect(${name})`,
|
||||
{
|
||||
abort() {
|
||||
// if interrupted, trying to disconnect
|
||||
drop(chatService.disconnect());
|
||||
},
|
||||
},
|
||||
connectAsync()
|
||||
);
|
||||
}
|
||||
|
||||
export class LibsignalWebSocketResource
|
||||
extends EventTarget
|
||||
implements IWebSocketResource
|
||||
|
|
@ -277,40 +435,6 @@ export class LibsignalWebSocketResource
|
|||
super();
|
||||
}
|
||||
|
||||
public static connect(
|
||||
libsignalNet: Net.Net,
|
||||
name: string
|
||||
): AbortableProcess<LibsignalWebSocketResource> {
|
||||
const chatService = libsignalNet.newChatService();
|
||||
const connectAsync = async () => {
|
||||
try {
|
||||
const debugInfo = await chatService.connectUnauthenticated();
|
||||
log.info(`LibsignalWebSocketResource(${name}) connected`, debugInfo);
|
||||
return new LibsignalWebSocketResource(
|
||||
chatService,
|
||||
IpVersion.fromDebugInfoCode(debugInfo.ipType)
|
||||
);
|
||||
} catch (error) {
|
||||
// Handle any errors that occur during connection
|
||||
log.error(
|
||||
`LibsignalWebSocketResource(${name}) connection failed`,
|
||||
Errors.toLogFormat(error)
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
return new AbortableProcess<LibsignalWebSocketResource>(
|
||||
`LibsignalWebSocketResource.connect(${name})`,
|
||||
{
|
||||
abort() {
|
||||
// if interrupted, trying to disconnect
|
||||
drop(chatService.disconnect());
|
||||
},
|
||||
},
|
||||
connectAsync()
|
||||
);
|
||||
}
|
||||
|
||||
public localPort(): number | undefined {
|
||||
return undefined;
|
||||
}
|
||||
|
|
@ -348,14 +472,13 @@ export class LibsignalWebSocketResource
|
|||
public async sendRequestGetDebugInfo(
|
||||
options: SendRequestOptions
|
||||
): Promise<[Response, ChatServiceDebugInfo]> {
|
||||
const { response, debugInfo } =
|
||||
await this.chatService.unauthenticatedFetchAndDebug({
|
||||
verb: options.verb,
|
||||
path: options.path,
|
||||
headers: options.headers ? options.headers : [],
|
||||
body: options.body,
|
||||
timeoutMillis: options.timeout,
|
||||
});
|
||||
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,
|
||||
|
|
@ -765,7 +888,7 @@ export default class WebSocketResource
|
|||
this.options.handleRequest ||
|
||||
(request => request.respond(404, 'Not found'));
|
||||
|
||||
const incomingRequest = new IncomingWebSocketRequest(
|
||||
const incomingRequest = new IncomingWebSocketRequestLegacy(
|
||||
message.request,
|
||||
(bytes: Buffer): void => {
|
||||
this.removeActive(incomingRequest);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue