From 5348e34d5520703aee0407c4c022e6a12c0ec12d Mon Sep 17 00:00:00 2001 From: Ricardo Garim Date: Thu, 9 Jul 2026 16:11:04 -0300 Subject: [PATCH 1/4] fix: import Slack files as attachments instead of raw URLs --- .changeset/deep-ways-poke.md | 7 ++ .../server/PendingFileImporter.ts | 89 ++++++++++++------- .../importer-slack/server/SlackImporter.ts | 6 +- .../classes/converters/MessageConverter.ts | 1 + .../core-typings/src/import/IImportMessage.ts | 1 + packages/models/src/models/Messages.ts | 1 + 6 files changed, 70 insertions(+), 35 deletions(-) create mode 100644 .changeset/deep-ways-poke.md diff --git a/.changeset/deep-ways-poke.md b/.changeset/deep-ways-poke.md new file mode 100644 index 0000000000000..918b70379880e --- /dev/null +++ b/.changeset/deep-ways-poke.md @@ -0,0 +1,7 @@ +--- +'@rocket.chat/core-typings': patch +'@rocket.chat/models': patch +'@rocket.chat/meteor': patch +--- + +Fixes the Slack importer storing shared files as raw URLs in the message body. Imported file messages now stay hidden until "Download Pending Files" button fetches them, then display as native attachments with image previews. Failed downloads (e.g. invalidated export links) are no longer silently saved as the file's content — they are counted as errors and can be retried. diff --git a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts index 35b88c4267138..f2340d07ff462 100644 --- a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts +++ b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts @@ -3,10 +3,11 @@ import https from 'node:https'; import { api } from '@rocket.chat/core-services'; import type { IImport, MessageAttachment, IUpload, IImporterShortSelection } from '@rocket.chat/core-typings'; -import { Messages } from '@rocket.chat/models'; +import { Messages, Users } from '@rocket.chat/models'; import { Random } from '@rocket.chat/random'; import { FileUpload } from '../../file-upload/server'; +import { parseFileIntoMessageAttachments } from '../../file-upload/server/methods/sendFileMessage'; import { Importer, ProgressStep } from '../../importer/server'; import type { ConverterOptions } from '../../importer/server/classes/ImportDataConverter'; import type { ImporterProgress } from '../../importer/server/classes/ImporterProgress'; @@ -79,6 +80,12 @@ export class PendingFileImporter extends Importer { currentSize -= details.size; }; + const failFile = async (details: { size: number }) => { + await this.addCountError(1); + count--; + currentSize -= details.size; + }; + const logError = this.logger.error.bind(this.logger); try { @@ -107,7 +114,6 @@ export class PendingFileImporter extends Importer { rid: message.rid, }; - const requestModule = /https/i.test(url) ? https : http; const fileStore = FileUpload.getStore('Uploads'); nextSize = details.size; @@ -116,41 +122,25 @@ export class PendingFileImporter extends Importer { currentSize += nextSize; downloadedFileIds.push(_importFile.id); - requestModule.get(url, (res) => { - const contentType = res.headers['content-type']; - if (!details.type && contentType) { - details.type = contentType; - } + void this.downloadFile(url, details) + .then(async (rawData) => { + // Bypass the fileStore filters + const file = await fileStore._doInsert(details, rawData); - const rawData: Uint8Array[] = []; - res.on('data', (chunk) => { - rawData.push(chunk); + const rocketChatUrl = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`); + const user = await Users.findOneById(message.u._id); + const attachment = user + ? (await parseFileIntoMessageAttachments(file, message.rid, user)).attachments[0] + : this.getMessageAttachment(file, rocketChatUrl); - // Update progress more often on large files - this.reportProgress(); - }); - res.on('error', async (err) => { + await Messages.setImportFileRocketChatAttachment(_importFile.id, rocketChatUrl, attachment); await completeFile(details); - logError({ err }); + importedRoomIds.add(message.rid); + }) + .catch(async (err) => { + logError({ msg: 'Failed to download pending file', url, err }); + await failFile(details); }); - - res.on('end', async () => { - try { - // Bypass the fileStore filters - const file = await fileStore._doInsert(details, Buffer.concat(rawData)); - - const url = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`); - const attachment = this.getMessageAttachment(file, url); - - await Messages.setImportFileRocketChatAttachment(_importFile.id, url, attachment); - await completeFile(details); - importedRoomIds.add(message.rid); - } catch (err) { - await completeFile(details); - logError({ err }); - } - }); - }); } catch (err) { this.logger.error({ err }); } @@ -172,6 +162,39 @@ export class PendingFileImporter extends Importer { return this.getProgress(); } + private downloadFile(url: string, details: { type?: string }): Promise { + return new Promise((resolve, reject) => { + const requestModule = /https/i.test(url) ? https : http; + + requestModule + .get(url, (res) => { + res.on('error', reject); + + if (!res.statusCode || res.statusCode < 200 || res.statusCode >= 300) { + // Error responses (e.g. a redirect to a login page) would otherwise be saved as the file's content. + res.resume(); // discard the body and free the socket + reject(new Error(`Unexpected response status ${res.statusCode}`)); + return; + } + + const contentType = res.headers['content-type']; + if (!details.type && contentType) { + details.type = contentType; + } + + const rawData: Uint8Array[] = []; + res.on('data', (chunk) => { + rawData.push(chunk); + + // Update progress more often on large files + this.reportProgress(); + }); + res.on('end', () => resolve(Buffer.concat(rawData))); + }) + .on('error', reject); + }); + } + getMessageAttachment(file: IUpload, url: string): MessageAttachment { if (file.type) { if (/^image\/.+/.test(file.type)) { diff --git a/apps/meteor/app/importer-slack/server/SlackImporter.ts b/apps/meteor/app/importer-slack/server/SlackImporter.ts index d7ccfbdec6c49..e7b0926b1cfa1 100644 --- a/apps/meteor/app/importer-slack/server/SlackImporter.ts +++ b/apps/meteor/app/importer-slack/server/SlackImporter.ts @@ -496,7 +496,8 @@ export class SlackImporter extends Importer { _id: fileId, rid: newMessage.rid, ts: newMessage.ts, - msg: message.file.url_private_download || '', + msg: '', + _hidden: true, _importFile: this.convertSlackFileToPendingFile(message.file), u: { _id: newMessage.u._id, @@ -568,7 +569,8 @@ export class SlackImporter extends Importer { _id: fileId, rid: slackChannelId, ts: newMessage.ts, - msg: file.url_private_download || '', + msg: '', + _hidden: true, _importFile: this.convertSlackFileToPendingFile(file), u: { _id: this._replaceSlackUserId(message.user), diff --git a/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts b/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts index a8f785bdf17c7..7bf6a3e4e8746 100644 --- a/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts +++ b/apps/meteor/app/importer/server/classes/converters/MessageConverter.ts @@ -104,6 +104,7 @@ export class MessageConverter extends RecordConverter { mentions, channels, _importFile: data._importFile, + _hidden: data._hidden, url: data.url, attachments: data.attachments, bot: data.bot, diff --git a/packages/core-typings/src/import/IImportMessage.ts b/packages/core-typings/src/import/IImportMessage.ts index 59dd231557096..0f4aa0d924685 100644 --- a/packages/core-typings/src/import/IImportMessage.ts +++ b/packages/core-typings/src/import/IImportMessage.ts @@ -48,4 +48,5 @@ export interface IImportMessage { url?: string; _importFile?: IImportPendingFile; + _hidden?: boolean; } diff --git a/packages/models/src/models/Messages.ts b/packages/models/src/models/Messages.ts index 0335b38c9d08e..906c1b5eb7e6f 100644 --- a/packages/models/src/models/Messages.ts +++ b/packages/models/src/models/Messages.ts @@ -663,6 +663,7 @@ export class MessagesRaw extends BaseRaw implements IMessagesModel { return this.updateMany(query, { $set: { + '_hidden': false, '_importFile.rocketChatUrl': rocketChatUrl, '_importFile.downloaded': true, }, From 3830a3f9eaa6bcbbdecf2e386e4e7e02662c7f1b Mon Sep 17 00:00:00 2001 From: Ricardo Garim Date: Mon, 13 Jul 2026 08:16:21 -0300 Subject: [PATCH 2/4] chore: auto-start pending file downloads with stall timeout --- .../app/file-upload/server/lib/FileUpload.ts | 9 +- .../server/PendingFileImporter.ts | 224 +++++++----------- .../importer-slack/server/SlackImporter.ts | 19 +- .../app/importer/server/classes/Importer.ts | 10 + .../core-typings/src/IMessage/IMessage.ts | 1 + .../src/models/IMessagesModel.ts | 3 +- packages/models/src/models/Messages.ts | 84 +++---- 7 files changed, 165 insertions(+), 185 deletions(-) diff --git a/apps/meteor/app/file-upload/server/lib/FileUpload.ts b/apps/meteor/app/file-upload/server/lib/FileUpload.ts index 3acc63bbb79cf..4b9a474c5afd7 100644 --- a/apps/meteor/app/file-upload/server/lib/FileUpload.ts +++ b/apps/meteor/app/file-upload/server/lib/FileUpload.ts @@ -5,7 +5,7 @@ import { unlink, rename, writeFile } from 'node:fs/promises'; import type * as http from 'node:http'; import type * as https from 'node:https'; import stream from 'node:stream'; -import { finished } from 'node:stream/promises'; +import { pipeline } from 'node:stream/promises'; import URL from 'node:url'; import { isArrayBufferView } from 'node:util/types'; @@ -903,13 +903,16 @@ export class FileUploadClass { } else if (isArrayBufferView(content)) { await fs.promises.writeFile(tmpFile, content); } else if (content instanceof stream.Readable) { - await finished(content.pipe(fs.createWriteStream(tmpFile)), { cleanup: true }); + await pipeline(content, fs.createWriteStream(tmpFile)); } else { throw new Error('Invalid file type'); } - return ufsComplete(fileId, this.name, { session: options?.session }); + return await ufsComplete(fileId, this.name, { session: options?.session }); } catch (e) { + // The record is created before the content is written; drop it so a failed + // stream doesn't leave an orphaned upload record and tmp file behind. + await this.store.removeById(fileId, { session: options?.session }).catch(() => undefined); throw e; } } diff --git a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts index f2340d07ff462..e2d5008ac85eb 100644 --- a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts +++ b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts @@ -1,23 +1,20 @@ import http from 'node:http'; import https from 'node:https'; +import type { Readable } from 'node:stream'; import { api } from '@rocket.chat/core-services'; -import type { IImport, MessageAttachment, IUpload, IImporterShortSelection } from '@rocket.chat/core-typings'; +import type { MessageAttachment, IUpload, IImporterShortSelection, IMessageWithPendingFileImport } from '@rocket.chat/core-typings'; import { Messages, Users } from '@rocket.chat/models'; import { Random } from '@rocket.chat/random'; import { FileUpload } from '../../file-upload/server'; import { parseFileIntoMessageAttachments } from '../../file-upload/server/methods/sendFileMessage'; import { Importer, ProgressStep } from '../../importer/server'; -import type { ConverterOptions } from '../../importer/server/classes/ImportDataConverter'; import type { ImporterProgress } from '../../importer/server/classes/ImporterProgress'; -import type { ImporterInfo } from '../../importer/server/definitions/ImporterInfo'; -export class PendingFileImporter extends Importer { - constructor(info: ImporterInfo, importRecord: IImport, converterOptions: ConverterOptions = {}) { - super(info, importRecord, converterOptions); - } +const LEASE_MS = 2 * 60 * 1000; +export class PendingFileImporter extends Importer { async prepareFileCount() { this.logger.debug('start preparing import operation'); await super.updateProgress(ProgressStep.PREPARING_STARTED); @@ -38,160 +35,117 @@ export class PendingFileImporter extends Importer { this.reportProgress(); setImmediate(() => { - void this.startImport({}); + this.startImport({}).catch(() => undefined); }); return fileCount; } - override async startImport(importSelection: IImporterShortSelection): Promise { - const downloadedFileIds: string[] = []; - const maxFileCount = 10; - const maxFileSize = 1024 * 1024 * 500; + override async startImport(_importSelection: IImporterShortSelection): Promise { + const importedRoomIds = new Set(); + const inFlightFileIds = new Set(); + const skipMessageIds = new Set(); - let count = 0; - let currentSize = 0; - let nextSize = 0; + const processMessage = async (message: IMessageWithPendingFileImport) => { + const { _importFile } = message; + const url = _importFile.downloadUrl; - const waitForFiles = async () => { - if (count + 1 < maxFileCount && currentSize + nextSize < maxFileSize) { + if (!url?.startsWith('http')) { + skipMessageIds.add(message._id); + await this.addCountError(1); return; } - return new Promise((resolve) => { - const handler = setInterval(() => { - if (count + 1 >= maxFileCount) { - return; - } - - if (currentSize + nextSize >= maxFileSize && count > 0) { - return; - } - - clearInterval(handler); - resolve(); - }, 1000); - }); - }; - - const completeFile = async (details: { size: number }) => { - await this.addCountCompleted(1); - count--; - currentSize -= details.size; - }; - - const failFile = async (details: { size: number }) => { - await this.addCountError(1); - count--; - currentSize -= details.size; + if (inFlightFileIds.has(_importFile.id)) { + skipMessageIds.add(message._id); + return; + } + inFlightFileIds.add(_importFile.id); + + const details: { message_id: string; name: string; size: number; userId: string; rid: string; type?: string } = { + message_id: `${message._id}-file-${_importFile.id}`, + name: _importFile.name || Random.id(), + size: _importFile.size || 0, + userId: message.u._id, + rid: message.rid, + }; + + const renewal = setInterval(() => void Messages.renewPendingFileImportLease(message._id, LEASE_MS), LEASE_MS / 4); + + try { + const fileStream = await this.downloadFile(url, details); + + // Bypass the fileStore filters + const file = await FileUpload.getStore('Uploads')._doInsert(details, fileStream); + + const rocketChatUrl = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`); + const user = await Users.findOneById(message.u._id); + const attachment = user + ? (await parseFileIntoMessageAttachments(file, message.rid, user)).attachments[0] + : this.getMessageAttachment(file, rocketChatUrl); + + const result = await Messages.setImportFileRocketChatAttachment(_importFile.id, rocketChatUrl, attachment); + await this.addCountCompleted('modifiedCount' in result ? result.modifiedCount : 1); + importedRoomIds.add(message.rid); + } catch (err) { + this.logger.error({ msg: 'Failed to download pending file', url: url.split('?')[0], err }); + skipMessageIds.add(message._id); + inFlightFileIds.delete(_importFile.id); + await this.addCountError(1); + } finally { + clearInterval(renewal); + } }; - const logError = this.logger.error.bind(this.logger); - - try { - const pendingFileMessageList = Messages.findAllImportedMessagesWithFilesToDownload(); - const importedRoomIds = new Set(); - for await (const message of pendingFileMessageList) { - try { - const { _importFile } = message; - - if (!_importFile || _importFile.downloaded || downloadedFileIds.includes(_importFile.id)) { - await this.addCountCompleted(1); - continue; - } - - const url = _importFile.downloadUrl; - if (!url?.startsWith('http')) { - await this.addCountCompleted(1); - continue; - } - - const details: { message_id: string; name: string; size: number; userId: string; rid: string; type?: string } = { - message_id: `${message._id}-file-${_importFile.id}`, - name: _importFile.name || Random.id(), - size: _importFile.size || 0, - userId: message.u._id, - rid: message.rid, - }; - - const fileStore = FileUpload.getStore('Uploads'); - - nextSize = details.size; - await waitForFiles(); - count++; - currentSize += nextSize; - downloadedFileIds.push(_importFile.id); - - void this.downloadFile(url, details) - .then(async (rawData) => { - // Bypass the fileStore filters - const file = await fileStore._doInsert(details, rawData); - - const rocketChatUrl = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`); - const user = await Users.findOneById(message.u._id); - const attachment = user - ? (await parseFileIntoMessageAttachments(file, message.rid, user)).attachments[0] - : this.getMessageAttachment(file, rocketChatUrl); - - await Messages.setImportFileRocketChatAttachment(_importFile.id, rocketChatUrl, attachment); - await completeFile(details); - importedRoomIds.add(message.rid); - }) - .catch(async (err) => { - logError({ msg: 'Failed to download pending file', url, err }); - await failFile(details); - }); - } catch (err) { - this.logger.error({ err }); + const worker = async () => { + for (;;) { + const message = await Messages.findOneAndClaimPendingFileImport(LEASE_MS, Array.from(skipMessageIds)); + if (!message) { + return; } - } - void api.broadcast('notify.importedMessages', { roomIds: Array.from(importedRoomIds) }); - } catch (error) { - // If the cursor expired, restart the method - if (this.isCursorNotFoundError(error)) { - this.logger.info('CursorNotFound'); - return this.startImport(importSelection); + await processMessage(message); } + }; + const results = await Promise.allSettled(Array.from({ length: 10 }, () => worker())); + + const crashes = results.filter((result): result is PromiseRejectedResult => result.status === 'rejected'); + if (crashes.length > 0) { + crashes.forEach((crash) => this.logger.error({ msg: 'Pending file worker crashed', err: crash.reason })); await super.updateProgress(ProgressStep.ERROR); - throw error; + throw crashes[0].reason; } + void api.broadcast('notify.importedMessages', { roomIds: Array.from(importedRoomIds) }); + await super.updateProgress(ProgressStep.DONE); return this.getProgress(); } - private downloadFile(url: string, details: { type?: string }): Promise { + private downloadFile(url: string, details: { type?: string }): Promise { return new Promise((resolve, reject) => { const requestModule = /https/i.test(url) ? https : http; - requestModule - .get(url, (res) => { - res.on('error', reject); - - if (!res.statusCode || res.statusCode < 200 || res.statusCode >= 300) { - // Error responses (e.g. a redirect to a login page) would otherwise be saved as the file's content. - res.resume(); // discard the body and free the socket - reject(new Error(`Unexpected response status ${res.statusCode}`)); - return; - } - - const contentType = res.headers['content-type']; - if (!details.type && contentType) { - details.type = contentType; - } - - const rawData: Uint8Array[] = []; - res.on('data', (chunk) => { - rawData.push(chunk); - - // Update progress more often on large files - this.reportProgress(); - }); - res.on('end', () => resolve(Buffer.concat(rawData))); - }) - .on('error', reject); + const request = requestModule.get(url, (res) => { + if (!res.statusCode || res.statusCode < 200 || res.statusCode >= 300) { + // Error responses (e.g. a redirect to a login page) would otherwise be saved as the file's content. + res.on('error', () => undefined); // already rejected; just avoid crashing while draining + res.resume(); // discard the body and free the socket + reject(new Error(`Unexpected response status ${res.statusCode}`)); + return; + } + + if (!details.type) { + details.type = res.headers['content-type'] || 'application/octet-stream'; + } + + resolve(res); + }); + + request.on('error', reject); + + request.setTimeout(60 * 1000, () => request.destroy(new Error('Download stalled'))); }); } diff --git a/apps/meteor/app/importer-slack/server/SlackImporter.ts b/apps/meteor/app/importer-slack/server/SlackImporter.ts index e7b0926b1cfa1..c2b1511b7345d 100644 --- a/apps/meteor/app/importer-slack/server/SlackImporter.ts +++ b/apps/meteor/app/importer-slack/server/SlackImporter.ts @@ -1,9 +1,11 @@ +import { Import } from '@rocket.chat/core-services'; import type { IImportUser, IImportMessage, IImportPendingFile } from '@rocket.chat/core-typings'; import { Messages, Settings, ImportData } from '@rocket.chat/models'; import type { IZipEntry } from 'adm-zip'; -import { Importer, ProgressStep, ImporterWebsocket } from '../../importer/server'; +import { Importer, ProgressStep, ImporterWebsocket, Importers } from '../../importer/server'; import type { ImporterProgress } from '../../importer/server/classes/ImporterProgress'; +import { PendingFileImporter } from '../../importer-pending-files/server/PendingFileImporter'; import { notifyOnSettingChanged } from '../../lib/server/lib/notifyListener'; import { MentionsParser } from '../../mentions/lib/MentionsParser'; import { settings } from '../../settings/server'; @@ -119,6 +121,21 @@ type SlackAttachment = { export class SlackImporter extends Importer { private _useUpsert = false; + protected override async onImportComplete(): Promise { + const pendingFilesImporter = Importers.get('pending-files'); + if (!pendingFilesImporter) { + return; + } + + if ((await Messages.countAllImportedMessagesWithFilesToDownload()) === 0) { + return; + } + + const operation = await Import.newOperation(this.importRecord.user, pendingFilesImporter.name, pendingFilesImporter.key); + const instance = new PendingFileImporter(pendingFilesImporter, operation); + await instance.prepareFileCount(); + } + async prepareChannelsFile(entry: IZipEntry): Promise { await super.updateProgress(ProgressStep.PREPARING_CHANNELS); const data = (JSON.parse(entry.getData().toString()) as SlackChannel[]).filter( diff --git a/apps/meteor/app/importer/server/classes/Importer.ts b/apps/meteor/app/importer/server/classes/Importer.ts index c39da71ed2a31..c7a14778714f3 100644 --- a/apps/meteor/app/importer/server/classes/Importer.ts +++ b/apps/meteor/app/importer/server/classes/Importer.ts @@ -226,6 +226,14 @@ export class Importer { await this.applySettingValues(this.oldSettings); } + if (this.progress.step === ProgressStep.DONE) { + try { + await this.onImportComplete?.(); + } catch (err) { + this.logger.error({ msg: 'Failed to run the post-import step', err }); + } + } + const timeTook = Date.now() - started; this.logger.log({ msg: 'Import completed', durationMs: timeTook }); }); @@ -268,6 +276,8 @@ export class Importer { return this.progress; } + protected onImportComplete?(): Promise; + /** * Updates the progress step of this importer. * It also changes some internal settings at various stages of the import. diff --git a/packages/core-typings/src/IMessage/IMessage.ts b/packages/core-typings/src/IMessage/IMessage.ts index a180c7d7528fa..335815d200ff1 100644 --- a/packages/core-typings/src/IMessage/IMessage.ts +++ b/packages/core-typings/src/IMessage/IMessage.ts @@ -403,6 +403,7 @@ export type IMessageWithPendingFileImport = IMessage & { original: Record; rocketChatUrl?: string; downloaded?: boolean; + lockedUntil?: Date; }; }; diff --git a/packages/model-typings/src/models/IMessagesModel.ts b/packages/model-typings/src/models/IMessagesModel.ts index 0d16750f012c8..fd37184682fa6 100644 --- a/packages/model-typings/src/models/IMessagesModel.ts +++ b/packages/model-typings/src/models/IMessagesModel.ts @@ -281,7 +281,8 @@ export interface IMessagesModel extends IBaseModel { setAsReadById(_id: string): Promise; countThreads(): Promise; addThreadFollowerByThreadId(tmid: string, userId: string): Promise; - findAllImportedMessagesWithFilesToDownload(): FindCursor; + findOneAndClaimPendingFileImport(leaseMs: number, excludedMessageIds: IMessage['_id'][]): Promise; + renewPendingFileImportLease(messageId: IMessage['_id'], leaseMs: number): Promise; countAllImportedMessagesWithFilesToDownload(): Promise; findAgentLastMessageByVisitorLastMessageTs(roomId: string, visitorLastMessageTs: Date): Promise; removeThreadFollowerByThreadId(tmid: string, userId: string): Promise; diff --git a/packages/models/src/models/Messages.ts b/packages/models/src/models/Messages.ts index 906c1b5eb7e6f..073eaa7e1ef24 100644 --- a/packages/models/src/models/Messages.ts +++ b/packages/models/src/models/Messages.ts @@ -657,20 +657,18 @@ export class MessagesRaw extends BaseRaw implements IMessagesModel { rocketChatUrl: string, attachment: MessageAttachment, ): Promise { - const query = { - '_importFile.id': importFileId, - }; - - return this.updateMany(query, { - $set: { - '_hidden': false, - '_importFile.rocketChatUrl': rocketChatUrl, - '_importFile.downloaded': true, - }, - $addToSet: { - attachments: attachment, + return this.updateMany( + { '_importFile.id': importFileId, '_importFile.downloaded': { $ne: true } }, + { + $set: { + '_hidden': false, + '_importFile.rocketChatUrl': rocketChatUrl, + '_importFile.downloaded': true, + }, + $unset: { '_importFile.lockedUntil': 1 }, + $addToSet: { attachments: attachment }, }, - }); + ); } countVisibleByRoomIdBetweenTimestampsInclusive(roomId: string, afterTimestamp: Date, beforeTimestamp: Date): Promise { @@ -1720,42 +1718,38 @@ export class MessagesRaw extends BaseRaw implements IMessagesModel { return this.findOne(query, { sort: { ts: 1 } }); } - findAllImportedMessagesWithFilesToDownload(): FindCursor { - const query = { - '_importFile.downloadUrl': { - $exists: true, - }, - '_importFile.rocketChatUrl': { - $exists: false, - }, - '_importFile.downloaded': { - $ne: true, - }, - '_importFile.external': { - $ne: true, - }, - }; + findOneAndClaimPendingFileImport(leaseMs: number, excludedMessageIds: IMessage['_id'][]): Promise { + const now = new Date(); - return this.find(query); + return this.col.findOneAndUpdate( + { + '_id': { $nin: excludedMessageIds }, + '_importFile.downloadUrl': { $exists: true }, + '_importFile.rocketChatUrl': { $exists: false }, + '_importFile.downloaded': { $ne: true }, + '_importFile.external': { $ne: true }, + '$or': [{ '_importFile.lockedUntil': { $exists: false } }, { '_importFile.lockedUntil': { $lt: now } }], + }, + { $set: { '_importFile.lockedUntil': new Date(now.getTime() + leaseMs) } }, + { returnDocument: 'after' }, + ) as Promise; } - countAllImportedMessagesWithFilesToDownload(): Promise { - const query = { - '_importFile.downloadUrl': { - $exists: true, - }, - '_importFile.rocketChatUrl': { - $exists: false, - }, - '_importFile.downloaded': { - $ne: true, - }, - '_importFile.external': { - $ne: true, - }, - }; + renewPendingFileImportLease(messageId: IMessage['_id'], leaseMs: number): Promise { + // The downloaded guard keeps a renewal racing the finalize from re-adding a stale lock. + return this.updateOne( + { '_id': messageId, '_importFile.downloaded': { $ne: true } }, + { $set: { '_importFile.lockedUntil': new Date(Date.now() + leaseMs) } }, + ); + } - return this.countDocuments(query); + countAllImportedMessagesWithFilesToDownload(): Promise { + return this.countDocuments({ + '_importFile.downloadUrl': { $exists: true }, + '_importFile.rocketChatUrl': { $exists: false }, + '_importFile.downloaded': { $ne: true }, + '_importFile.external': { $ne: true }, + }); } decreaseReplyCountById(_id: string, inc = -1): Promise { From 14d70a7e98a5ce917d84e7fa013b7066ff7b4934 Mon Sep 17 00:00:00 2001 From: Ricardo Garim Date: Mon, 13 Jul 2026 09:53:31 -0300 Subject: [PATCH 3/4] fix: block SSRF in pending file downloads --- .../server/PendingFileImporter.ts | 49 ++++++++++--------- 1 file changed, 26 insertions(+), 23 deletions(-) diff --git a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts index e2d5008ac85eb..9c079d3d427ad 100644 --- a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts +++ b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts @@ -1,16 +1,17 @@ -import http from 'node:http'; -import https from 'node:https'; -import type { Readable } from 'node:stream'; +import { Transform, type Readable } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; import { api } from '@rocket.chat/core-services'; import type { MessageAttachment, IUpload, IImporterShortSelection, IMessageWithPendingFileImport } from '@rocket.chat/core-typings'; import { Messages, Users } from '@rocket.chat/models'; import { Random } from '@rocket.chat/random'; +import { serverFetch } from '@rocket.chat/server-fetch'; import { FileUpload } from '../../file-upload/server'; import { parseFileIntoMessageAttachments } from '../../file-upload/server/methods/sendFileMessage'; import { Importer, ProgressStep } from '../../importer/server'; import type { ImporterProgress } from '../../importer/server/classes/ImporterProgress'; +import { settings } from '../../settings/server'; const LEASE_MS = 2 * 60 * 1000; @@ -123,30 +124,32 @@ export class PendingFileImporter extends Importer { return this.getProgress(); } - private downloadFile(url: string, details: { type?: string }): Promise { - return new Promise((resolve, reject) => { - const requestModule = /https/i.test(url) ? https : http; + private async downloadFile(url: string, details: { type?: string }): Promise { + const response = await serverFetch(url, { ignoreSsrfValidation: false, allowList: settings.get('SSRF_Allowlist') }); - const request = requestModule.get(url, (res) => { - if (!res.statusCode || res.statusCode < 200 || res.statusCode >= 300) { - // Error responses (e.g. a redirect to a login page) would otherwise be saved as the file's content. - res.on('error', () => undefined); // already rejected; just avoid crashing while draining - res.resume(); // discard the body and free the socket - reject(new Error(`Unexpected response status ${res.statusCode}`)); - return; - } - - if (!details.type) { - details.type = res.headers['content-type'] || 'application/octet-stream'; - } - - resolve(res); - }); + if (!response.ok) { + response.body.resume(); + throw new Error(`Unexpected response status ${response.status}`); + } - request.on('error', reject); + if (!details.type) { + details.type = response.headers.get('content-type') || 'application/octet-stream'; + } - request.setTimeout(60 * 1000, () => request.destroy(new Error('Download stalled'))); + // serverFetch only times out connecting; fail the download if the body goes idle for 60s. + // Per-chunk reset (not a total cap) so large files aren't cut off mid-download. + let idle: ReturnType; + const meter = new Transform({ + transform(chunk, _encoding, callback) { + idle.refresh(); + callback(null, chunk); + }, }); + idle = setTimeout(() => meter.destroy(new Error('Download stalled')), 60 * 1000); + meter.on('close', () => clearTimeout(idle)); + + void pipeline(response.body, meter).catch(() => undefined); + return meter; } getMessageAttachment(file: IUpload, url: string): MessageAttachment { From 7635fbac92e24e397f1207c191fc4a0f4e6141e8 Mon Sep 17 00:00:00 2001 From: Ricardo Garim Date: Mon, 13 Jul 2026 10:13:54 -0300 Subject: [PATCH 4/4] fix: delete orphaned upload when pending-file import fails --- .../importer-pending-files/server/PendingFileImporter.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts index 9c079d3d427ad..c3434420d10e2 100644 --- a/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts +++ b/apps/meteor/app/importer-pending-files/server/PendingFileImporter.ts @@ -73,11 +73,12 @@ export class PendingFileImporter extends Importer { const renewal = setInterval(() => void Messages.renewPendingFileImportLease(message._id, LEASE_MS), LEASE_MS / 4); + let file: IUpload | undefined; try { const fileStream = await this.downloadFile(url, details); // Bypass the fileStore filters - const file = await FileUpload.getStore('Uploads')._doInsert(details, fileStream); + file = await FileUpload.getStore('Uploads')._doInsert(details, fileStream); const rocketChatUrl = FileUpload.getPath(`${file._id}/${encodeURI(file.name || '')}`); const user = await Users.findOneById(message.u._id); @@ -90,6 +91,11 @@ export class PendingFileImporter extends Importer { importedRoomIds.add(message.rid); } catch (err) { this.logger.error({ msg: 'Failed to download pending file', url: url.split('?')[0], err }); + if (file) { + await FileUpload.getStore('Uploads') + .deleteById(file._id) + .catch(() => undefined); + } skipMessageIds.add(message._id); inFlightFileIds.delete(_importFile.id); await this.addCountError(1);