mirrors/signal-desktop
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

Use streams to download attachments directly to disk

Browse source 
Co-authored-by: trevor-signal <131492920+trevor-signal@users.noreply.github.com>
This commit is contained in:
Scott Nonnenberg 2023-10-30 09:24:28 -07:00 • committed by GitHub
parent 2da49456c6
commit 99b2bc304e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
48 changed files with 2297 additions and 356 deletions
Download patch file Download diff file Expand all files Collapse all files

290
ts/textsecure/ContactsParser.ts
View file

@ -1,159 +1,233 @@
// Copyright 2020 Signal Messenger, LLC
// SPDX-License-Identifier: AGPL-3.0-only
/* eslint-disable max-classes-per-file */
import protobuf from '../protobuf/wrap';
import { createReadStream } from 'fs';
import { Transform } from 'stream';
import { pipeline } from 'stream/promises';
import { SignalService as Proto } from '../protobuf';
import protobuf from '../protobuf/wrap';
import { normalizeAci } from '../util/normalizeAci';
import { isAciString } from '../util/isAciString';
import { DurationInSeconds } from '../util/durations';
import * as Errors from '../types/errors';
import * as log from '../logging/log';
import type { ContactAvatarType } from '../types/Avatar';
import { computeHash } from '../Crypto';
import { dropNull } from '../util/dropNull';
import Avatar = Proto.ContactDetails.IAvatar;
const { Reader } = protobuf;
type OptionalFields = { avatar?: Avatar | null; expireTimer?: number | null };
type DecoderBase<Message extends OptionalFields> = {
decodeDelimited(reader: protobuf.Reader): Message | undefined;
type OptionalFields = {
avatar?: Avatar | null;
expireTimer?: number | null;
number?: string | null;
};
type HydratedAvatar = Avatar & { data: Uint8Array };
type MessageWithAvatar<Message extends OptionalFields> = Omit<
Message,
'avatar'
'avatar' | 'toJSON'
> & {
avatar?: HydratedAvatar;
avatar?: ContactAvatarType;
expireTimer?: DurationInSeconds;
number?: string | undefined;
};
export type ModifiedContactDetails = MessageWithAvatar<Proto.ContactDetails>;
export type ContactDetailsWithAvatar = MessageWithAvatar<Proto.IContactDetails>;
/* eslint-disable @typescript-eslint/brace-style -- Prettier conflicts with ESLint */
abstract class ParserBase<
Message extends OptionalFields,
Decoder extends DecoderBase<Message>,
Result
> implements Iterable<Result>
{
/* eslint-enable @typescript-eslint/brace-style */
export async function parseContactsV2({
absolutePath,
}: {
absolutePath: string;
}): Promise<ReadonlyArray<ContactDetailsWithAvatar>> {
const logId = 'parseContactsV2';
protected readonly reader: protobuf.Reader;
const readStream = createReadStream(absolutePath);
const parseContactsTransform = new ParseContactsTransform();
constructor(bytes: Uint8Array, private readonly decoder: Decoder) {
this.reader = new Reader(bytes);
try {
await pipeline(readStream, parseContactsTransform);
} catch (error) {
try {
readStream.close();
} catch (cleanupError) {
log.error(
`${logId}: Failed to clean up after error`,
Errors.toLogFormat(cleanupError)
);
}
throw error;
}
protected decodeDelimited(): MessageWithAvatar<Message> | undefined {
if (this.reader.pos === this.reader.len) {
return undefined; // eof
readStream.close();
return parseContactsTransform.contacts;
}
// This transform pulls contacts and their avatars from a stream of bytes. This is tricky,
// because the chunk boundaries might fall in the middle of a contact or their avatar.
// So we are ready for decodeDelimited() to throw, and to keep activeContact around
// while we wait for more chunks to get to the expected avatar size.
// Note: exported only for testing
export class ParseContactsTransform extends Transform {
public contacts: Array<ContactDetailsWithAvatar> = [];
public activeContact: Proto.ContactDetails | undefined;
private unused: Uint8Array | undefined;
override async _transform(
chunk: Buffer | undefined,
_encoding: string,
done: (error?: Error) => void
): Promise<void> {
if (!chunk || chunk.byteLength === 0) {
done();
return;
}
try {
const proto = this.decoder.decodeDelimited(this.reader);
if (!proto) {
return undefined;
let data = chunk;
if (this.unused) {
data = Buffer.concat([this.unused, data]);
this.unused = undefined;
}
let avatar: HydratedAvatar | undefined;
if (proto.avatar) {
const attachmentLen = proto.avatar.length ?? 0;
const avatarData = this.reader.buf.slice(
this.reader.pos,
this.reader.pos + attachmentLen
);
this.reader.skip(attachmentLen);
const reader = Reader.create(data);
while (reader.pos < reader.len) {
const startPos = reader.pos;
avatar = {
...proto.avatar,
if (!this.activeContact) {
try {
this.activeContact = Proto.ContactDetails.decodeDelimited(reader);
} catch (err) {
// We get a RangeError if there wasn't enough data to read the next record.
if (err instanceof RangeError) {
// Note: A failed decodeDelimited() does in fact update reader.pos, so we
// must reset to startPos
this.unused = data.subarray(startPos);
done();
return;
}
data: avatarData,
};
// Something deeper has gone wrong; the proto is malformed or something
done(err);
return;
}
}
// Something has really gone wrong if the above parsing didn't throw but gave
// us nothing back. Let's end the parse.
if (!this.activeContact) {
done(new Error('ParseContactsTransform: No active contact!'));
return;
}
const attachmentSize = this.activeContact?.avatar?.length ?? 0;
if (attachmentSize === 0) {
// No avatar attachment for this contact
const prepared = prepareContact(this.activeContact);
if (prepared) {
this.contacts.push(prepared);
}
this.activeContact = undefined;
continue;
}
const spaceLeftAfterRead = reader.len - (reader.pos + attachmentSize);
if (spaceLeftAfterRead >= 0) {
// We've read enough data to read the entire attachment
const avatarData = reader.buf.slice(
reader.pos,
reader.pos + attachmentSize
);
const hash = computeHash(data);
// eslint-disable-next-line no-await-in-loop
const path = await window.Signal.Migrations.writeNewAttachmentData(
avatarData
);
const prepared = prepareContact(this.activeContact, {
...this.activeContact.avatar,
hash,
path,
});
if (prepared) {
this.contacts.push(prepared);
} else {
// eslint-disable-next-line no-await-in-loop
await window.Signal.Migrations.deleteAttachmentData(path);
}
this.activeContact = undefined;
reader.skip(attachmentSize);
} else {
// We have an attachment, but we haven't read enough data yet. We need to
// wait for another chunk.
this.unused = data.subarray(reader.pos);
done();
return;
}
}
let expireTimer: DurationInSeconds | undefined;
if (proto.expireTimer != null) {
expireTimer = DurationInSeconds.fromSeconds(proto.expireTimer);
}
return {
...proto,
avatar,
expireTimer,
};
// No need to push; no downstream consumers!
} catch (error) {
log.error('ProtoParser.next error:', Errors.toLogFormat(error));
return undefined;
done(error);
return;
}
}
public abstract next(): Result | undefined;
*[Symbol.iterator](): Iterator<Result> {
let result = this.next();
while (result !== undefined) {
yield result;
result = this.next();
}
done();
}
}
export class ContactBuffer extends ParserBase<
Proto.ContactDetails,
typeof Proto.ContactDetails,
ModifiedContactDetails
> {
constructor(arrayBuffer: Uint8Array) {
super(arrayBuffer, Proto.ContactDetails);
}
function prepareContact(
proto: Proto.ContactDetails,
avatar?: ContactAvatarType
): ContactDetailsWithAvatar | undefined {
const aci = proto.aci
? normalizeAci(proto.aci, 'ContactBuffer.aci')
: proto.aci;
public override next(): ModifiedContactDetails | undefined {
while (this.reader.pos < this.reader.len) {
const proto = this.decodeDelimited();
if (!proto) {
return undefined;
}
const expireTimer =
proto.expireTimer != null
? DurationInSeconds.fromSeconds(proto.expireTimer)
: undefined;
if (!proto.aci) {
return proto;
}
const verified =
proto.verified && proto.verified.destinationAci
? {
...proto.verified,
const { verified } = proto;
destinationAci: normalizeAci(
proto.verified.destinationAci,
'ContactBuffer.verified.destinationAci'
),
}
: proto.verified;
if (
!isAciString(proto.aci) ||
(verified?.destinationAci && !isAciString(verified.destinationAci))
) {
continue;
}
return {
...proto,
verified:
verified && verified.destinationAci
? {
...verified,
destinationAci: normalizeAci(
verified.destinationAci,
'ContactBuffer.verified.destinationAci'
),
}
: verified,
aci: normalizeAci(proto.aci, 'ContactBuffer.aci'),
};
}
// We reject incoming contacts with invalid aci information
if (
(proto.aci && !isAciString(proto.aci)) ||
(proto.verified?.destinationAci &&
!isAciString(proto.verified.destinationAci))
) {
log.warn('ParseContactsTransform: Dropping contact with invalid aci');
return undefined;
}
const result = {
...proto,
expireTimer,
aci,
verified,
avatar,
number: dropNull(proto.number),
};
return result;
}

74
ts/textsecure/MessageReceiver.ts
View file

@ -6,6 +6,8 @@
import { isBoolean, isNumber, isString, omit } from 'lodash';
import PQueue from 'p-queue';
import { v4 as getGuid } from 'uuid';
import { existsSync } from 'fs';
import { removeSync } from 'fs-extra';
import type {
SealedSenderDecryptionResult,
@ -49,7 +51,7 @@ import { parseIntOrThrow } from '../util/parseIntOrThrow';
import { clearTimeoutIfNecessary } from '../util/clearTimeoutIfNecessary';
import { Zone } from '../util/Zone';
import { DurationInSeconds, SECOND } from '../util/durations';
import type { DownloadedAttachmentType } from '../types/Attachment';
import type { AttachmentType } from '../types/Attachment';
import { Address } from '../types/Address';
import { QualifiedAddress } from '../types/QualifiedAddress';
import { normalizeStoryDistributionId } from '../types/StoryDistributionId';
@ -81,9 +83,10 @@ import {
import { processSyncMessage } from './processSyncMessage';
import type { EventHandler } from './EventTarget';
import EventTarget from './EventTarget';
import { downloadAttachment } from './downloadAttachment';
import { downloadAttachmentV2 } from './downloadAttachment';
import type { IncomingWebSocketRequest } from './WebsocketResources';
import { ContactBuffer } from './ContactsParser';
import type { ContactDetailsWithAvatar } from './ContactsParser';
import { parseContactsV2 } from './ContactsParser';
import type { WebAPIType } from './WebAPI';
import type { Storage } from './Storage';
import { WarnOnlyError } from './Errors';
@ -3504,11 +3507,11 @@ export default class MessageReceiver
private async handleContacts(
envelope: ProcessedEnvelope,
contacts: Proto.SyncMessage.IContacts
contactSyncProto: Proto.SyncMessage.IContacts
): Promise<void> {
const logId = getEnvelopeId(envelope);
log.info(`MessageReceiver: handleContacts ${logId}`);
const { blob } = contacts;
const { blob } = contactSyncProto;
if (!blob) {
throw new Error('MessageReceiver.handleContacts: blob field was missing');
}
@ -3517,21 +3520,50 @@ export default class MessageReceiver
this.removeFromCache(envelope);
const attachmentPointer = await this.handleAttachment(blob, {
disableRetries: true,
timeout: 90 * SECOND,
});
const contactBuffer = new ContactBuffer(attachmentPointer.data);
let attachment: AttachmentType | undefined;
try {
attachment = await this.handleAttachmentV2(blob, {
disableRetries: true,
timeout: 90 * SECOND,
});
const contactSync = new ContactSyncEvent(
Array.from(contactBuffer),
Boolean(contacts.complete),
envelope.receivedAtCounter,
envelope.timestamp
);
await this.dispatchAndWait(logId, contactSync);
const { path } = attachment;
if (!path) {
throw new Error('Failed no path field in returned attachment');
}
const absolutePath =
window.Signal.Migrations.getAbsoluteAttachmentPath(path);
if (!existsSync(absolutePath)) {
throw new Error(
'Contact sync attachment had path, but it was not found on disk'
);
}
log.info('handleContacts: finished');
let contacts: ReadonlyArray<ContactDetailsWithAvatar>;
try {
contacts = await parseContactsV2({
absolutePath,
});
} finally {
if (absolutePath) {
removeSync(absolutePath);
}
}
const contactSync = new ContactSyncEvent(
contacts,
Boolean(contactSyncProto.complete),
envelope.receivedAtCounter,
envelope.timestamp
);
await this.dispatchAndWait(logId, contactSync);
log.info('handleContacts: finished');
} finally {
if (attachment?.path) {
await window.Signal.Migrations.deleteAttachmentData(attachment.path);
}
}
}
private async handleBlocked(
@ -3618,12 +3650,12 @@ export default class MessageReceiver
return this.storage.blocked.isGroupBlocked(groupId);
}
private async handleAttachment(
private async handleAttachmentV2(
attachment: Proto.IAttachmentPointer,
options?: { timeout?: number; disableRetries?: boolean }
): Promise<DownloadedAttachmentType> {
): Promise<AttachmentType> {
const cleaned = processAttachment(attachment);
return downloadAttachment(this.server, cleaned, options);
return downloadAttachmentV2(this.server, cleaned, options);
}
private async handleEndSession(

69
ts/textsecure/WebAPI.ts
View file

@ -22,7 +22,10 @@ import * as durations from '../util/durations';
import type { ExplodePromiseResultType } from '../util/explodePromise';
import { explodePromise } from '../util/explodePromise';
import { getUserAgent } from '../util/getUserAgent';
import { getStreamWithTimeout } from '../util/getStreamWithTimeout';
import {
getTimeoutStream,
getStreamWithTimeout,
} from '../util/getStreamWithTimeout';
import { formatAcceptLanguageHeader } from '../util/userLanguages';
import { toWebSafeBase64, fromWebSafeBase64 } from '../util/webSafeBase64';
import { getBasicAuth } from '../util/getBasicAuth';
@ -970,6 +973,14 @@ export type WebAPIType = {
timeout?: number;
}
) => Promise<Uint8Array>;
getAttachmentV2: (
cdnKey: string,
cdnNumber?: number,
options?: {
disableRetries?: boolean;
timeout?: number;
}
) => Promise<Readable>;
getAvatar: (path: string) => Promise<Uint8Array>;
getHasSubscription: (subscriberId: Uint8Array) => Promise<boolean>;
getGroup: (options: GroupCredentialsType) => Promise<Proto.Group>;
@ -1386,6 +1397,7 @@ export function initialize({
getArtAuth,
getArtProvisioningSocket,
getAttachment,
getAttachmentV2,
getAvatar,
getBadgeImageFile,
getConfig,
@ -2876,6 +2888,61 @@ export function initialize({
}
}
async function getAttachmentV2(
cdnKey: string,
cdnNumber?: number,
options?: {
disableRetries?: boolean;
timeout?: number;
}
): Promise<Readable> {
const abortController = new AbortController();
const cdnUrl = isNumber(cdnNumber)
? cdnUrlObject[cdnNumber] ?? cdnUrlObject['0']
: cdnUrlObject['0'];
// This is going to the CDN, not the service, so we use _outerAjax
const downloadStream = await _outerAjax(
`${cdnUrl}/attachments/${cdnKey}`,
{
certificateAuthority,
disableRetries: options?.disableRetries,
proxyUrl,
responseType: 'stream',
timeout: options?.timeout || 0,
type: 'GET',
redactUrl: _createRedactor(cdnKey),
version,
abortSignal: abortController.signal,
}
);
const timeoutStream = getTimeoutStream({
name: `getAttachment(${cdnKey})`,
timeout: GET_ATTACHMENT_CHUNK_TIMEOUT,
abortController,
});
const combinedStream = downloadStream
// We do this manually; pipe() doesn't flow errors through the streams for us
.on('error', (error: Error) => {
timeoutStream.emit('error', error);
})
.pipe(timeoutStream);
const cancelRequest = (error: Error) => {
combinedStream.emit('error', error);
abortController.abort();
};
registerInflightRequest(cancelRequest);
combinedStream.on('done', () => {
unregisterInFlightRequest(cancelRequest);
});
return combinedStream;
}
async function putEncryptedAttachment(encryptedBin: Uint8Array) {
const response = attachmentV3Response.parse(
await _ajax({

8
ts/textsecure/authorizeArtCreator.ts
View file

@ -37,12 +37,12 @@ export async function authorizeArtCreator({
);
const keys = Bytes.concatenate([aesKey, macKey]);
const { ciphertext } = encryptAttachment(
Proto.ArtProvisioningMessage.encode({
const { ciphertext } = encryptAttachment({
plaintext: Proto.ArtProvisioningMessage.encode({
...auth,
}).finish(),
keys
);
keys,
});
const envelope = Proto.ArtProvisioningEnvelope.encode({
publicKey: ourKeys.pubKey,

162
ts/textsecure/downloadAttachment.ts
View file

@ -1,19 +1,40 @@
// Copyright 2020 Signal Messenger, LLC
// SPDX-License-Identifier: AGPL-3.0-only
import { isNumber } from 'lodash';
import { createWriteStream, existsSync, unlinkSync } from 'fs';
import { isNumber, omit } from 'lodash';
import type { Readable } from 'stream';
import { Transform } from 'stream';
import { pipeline } from 'stream/promises';
import { ensureFile } from 'fs-extra';
import * as log from '../logging/log';
import * as Errors from '../types/errors';
import { strictAssert } from '../util/assert';
import { dropNull } from '../util/dropNull';
import type { DownloadedAttachmentType } from '../types/Attachment';
import {
AttachmentSizeError,
type AttachmentType,
type DownloadedAttachmentType,
} from '../types/Attachment';
import * as MIME from '../types/MIME';
import * as Bytes from '../Bytes';
import { getFirstBytes, decryptAttachment } from '../Crypto';
import {
getFirstBytes,
decryptAttachmentV1,
getAttachmentSizeBucket,
} from '../Crypto';
import {
decryptAttachmentV2,
IV_LENGTH,
ATTACHMENT_MAC_LENGTH,
} from '../AttachmentCrypto';
import type { ProcessedAttachment } from './Types.d';
import type { WebAPIType } from './WebAPI';
import { createName, getRelativePath } from '../windows/attachments';
export async function downloadAttachment(
export async function downloadAttachmentV1(
server: WebAPIType,
attachment: ProcessedAttachment,
options?: {
@ -28,7 +49,6 @@ export async function downloadAttachment(
throw new Error('downloadAttachment: Attachment was missing cdnId!');
}
strictAssert(cdnId, 'attachment without cdnId');
const encrypted = await server.getAttachment(
cdnId,
dropNull(cdnNumber),
@ -41,9 +61,8 @@ export async function downloadAttachment(
}
strictAssert(key, 'attachment has no key');
strictAssert(digest, 'attachment has no digest');
const paddedData = decryptAttachment(
const paddedData = decryptAttachmentV1(
encrypted,
Bytes.fromBase64(key),
Bytes.fromBase64(digest)
@ -67,3 +86,132 @@ export async function downloadAttachment(
data,
};
}
export async function downloadAttachmentV2(
server: WebAPIType,
attachment: ProcessedAttachment,
options?: {
disableRetries?: boolean;
timeout?: number;
}
): Promise<AttachmentType> {
const { cdnId, cdnKey, cdnNumber, contentType, digest, key, size } =
attachment;
const cdn = cdnId || cdnKey;
const logId = `downloadAttachmentV2(${cdn}):`;
strictAssert(cdn, `${logId}: missing cdnId or cdnKey`);
strictAssert(digest, `${logId}: missing digest`);
strictAssert(key, `${logId}: missing key`);
strictAssert(isNumber(size), `${logId}: missing size`);
const downloadStream = await server.getAttachmentV2(
cdn,
dropNull(cdnNumber),
options
);
const cipherTextRelativePath = await downloadToDisk({ downloadStream, size });
const cipherTextAbsolutePath =
window.Signal.Migrations.getAbsoluteAttachmentPath(cipherTextRelativePath);
const relativePath = await decryptAttachmentV2({
ciphertextPath: cipherTextAbsolutePath,
id: cdn,
keys: Bytes.fromBase64(key),
size,
theirDigest: Bytes.fromBase64(digest),
});
if (existsSync(cipherTextAbsolutePath)) {
unlinkSync(cipherTextAbsolutePath);
}
return {
...omit(attachment, 'key'),
path: relativePath,
size,
contentType: contentType
? MIME.stringToMIMEType(contentType)
: MIME.APPLICATION_OCTET_STREAM,
};
}
async function downloadToDisk({
downloadStream,
size,
}: {
downloadStream: Readable;
size: number;
}): Promise<string> {
const relativeTargetPath = getRelativePath(createName());
const absoluteTargetPath =
window.Signal.Migrations.getAbsoluteAttachmentPath(relativeTargetPath);
await ensureFile(absoluteTargetPath);
const writeStream = createWriteStream(absoluteTargetPath);
const targetSize =
getAttachmentSizeBucket(size) * 1.05 + IV_LENGTH + ATTACHMENT_MAC_LENGTH;
const checkSizeTransform = new CheckSizeTransform(targetSize);
try {
await pipeline(downloadStream, checkSizeTransform, writeStream);
} catch (error) {
try {
writeStream.close();
if (absoluteTargetPath && existsSync(absoluteTargetPath)) {
unlinkSync(absoluteTargetPath);
}
} catch (cleanupError) {
log.error(
'downloadToDisk: Error while cleaning up',
Errors.toLogFormat(cleanupError)
);
}
throw error;
}
return relativeTargetPath;
}
// A simple transform that throws if it sees more than maxBytes on the stream.
class CheckSizeTransform extends Transform {
private bytesSeen = 0;
constructor(private maxBytes: number) {
super();
}
override _transform(
chunk: Buffer | undefined,
_encoding: string,
done: (error?: Error) => void
) {
if (!chunk || chunk.byteLength === 0) {
done();
return;
}
try {
this.bytesSeen += chunk.byteLength;
if (this.bytesSeen > this.maxBytes) {
done(
new AttachmentSizeError(
`CheckSizeTransform: Saw ${this.bytesSeen} bytes, max is ${this.maxBytes} bytes`
)
);
return;
}
this.push(chunk);
} catch (error) {
done(error);
return;
}
done();
}
}

3
ts/textsecure/index.ts
View file

@ -5,7 +5,6 @@ import EventTarget from './EventTarget';
import AccountManager from './AccountManager';
import MessageReceiver from './MessageReceiver';
import utils from './Helpers';
import { ContactBuffer } from './ContactsParser';
import SyncRequest from './SyncRequest';
import MessageSender from './SendMessage';
import { Storage } from './Storage';
@ -17,7 +16,6 @@ export type TextSecureType = {
storage: Storage;
AccountManager: typeof AccountManager;
ContactBuffer: typeof ContactBuffer;
EventTarget: typeof EventTarget;
MessageReceiver: typeof MessageReceiver;
MessageSender: typeof MessageSender;
@ -34,7 +32,6 @@ export const textsecure: TextSecureType = {
storage: new Storage(),
AccountManager,
ContactBuffer,
EventTarget,
MessageReceiver,
MessageSender,

4
ts/textsecure/messageReceiverEvents.ts
View file

@ -12,7 +12,7 @@ import type {
ProcessedDataMessage,
ProcessedSent,
} from './Types.d';
import type { ModifiedContactDetails } from './ContactsParser';
import type { ContactDetailsWithAvatar } from './ContactsParser';
import type { CallEventDetails, CallLogEvent } from '../types/CallDisposition';
export class EmptyEvent extends Event {
@ -74,7 +74,7 @@ export class ErrorEvent extends Event {
export class ContactSyncEvent extends Event {
constructor(
public readonly contacts: ReadonlyArray<ModifiedContactDetails>,
public readonly contacts: ReadonlyArray<ContactDetailsWithAvatar>,
public readonly complete: boolean,
public readonly receivedAtCounter: number,
public readonly sentAt: number