fix(api): enforce attachment upload provenance (#1578)

Author: hampus-fluxer

fluxer_api/src/api/Validator.ts

@@ -144,6 +144,23 @@ type ValidatorOptions<
 	post?: Hook<T, E, P, Target, V>;
 };
 
+export function inputValidationErrorFromZodIssues(issues: ZodError['issues']): InputValidationError {
+	const errors: Array<ValidationError> = [];
+	const localizedErrors: Array<LocalizedValidationError> = [];
+	const seen = new Set<string>();
+	for (const issue of issues) {
+		const path = issue.path.length > 0 ? issue.path.map(String).join('.') : 'root';
+		const code = getValidationErrorCode(issue.message);
+		const key = `${path}|${code}`;
+		if (seen.has(key)) continue;
+		seen.add(key);
+		const variables = extractVariablesFromIssue(issue);
+		errors.push({path, message: code, code});
+		localizedErrors.push({path, code, variables});
+	}
+	return new InputValidationError(errors, localizedErrors);
+}
+
 export const Validator = <
 	T extends ZodTypeAny,
 	Target extends keyof ValidationTargets,
@@ -246,20 +263,7 @@ export const Validator = <
 			}
 		}
 		if (!result.success) {
-			const errors: Array<ValidationError> = [];
-			const localizedErrors: Array<LocalizedValidationError> = [];
-			const seen = new Set<string>();
-			for (const issue of result.error.issues) {
-				const path = issue.path.length > 0 ? issue.path.map(String).join('.') : 'root';
-				const code = getValidationErrorCode(issue.message);
-				const key = `${path}|${code}`;
-				if (seen.has(key)) continue;
-				seen.add(key);
-				const variables = extractVariablesFromIssue(issue);
-				errors.push({path, message: code, code});
-				localizedErrors.push({path, code, variables});
-			}
-			throw new InputValidationError(errors, localizedErrors);
+			throw inputValidationErrorFromZodIssues(result.error.issues);
 		}
 		c.req.addValidatedData(target, result.data as ValidationTargets[Target]);
 		await next();

fluxer_api/src/api/channel/repositories/message/AttachmentUploadTraceRepository.ts

@@ -36,6 +36,13 @@ interface MarkAttachmentUploadCompletedInput {
 	completedAt?: Date;
 }
 
+interface GetPendingAttachmentUploadInput {
+	uploadKey: string;
+	userId: UserID;
+	channelId: ChannelID;
+	uploadMode?: AttachmentUploadMode;
+}
+
 export class AttachmentUploadTraceRepository {
 	async getByUploadKey(uploadKey: string): Promise<AttachmentUploadTraceByKeyRow | null> {
 		return await fetchOne<AttachmentUploadTraceByKeyRow>(GET_UPLOAD_TRACE_BY_KEY_QUERY.bind({upload_key: uploadKey}));
@@ -47,6 +54,20 @@ export class AttachmentUploadTraceRepository {
 		);
 	}
 
+	async getPendingUpload(input: GetPendingAttachmentUploadInput): Promise<AttachmentUploadTraceByKeyRow | null> {
+		const existing = await this.getByUploadKey(input.uploadKey);
+		if (
+			!existing ||
+			existing.user_id !== input.userId ||
+			existing.channel_id !== input.channelId ||
+			existing.attachment_id != null ||
+			(input.uploadMode !== undefined && existing.upload_mode !== input.uploadMode)
+		) {
+			return null;
+		}
+		return existing;
+	}
+
 	async recordRequestedUpload(input: RecordAttachmentUploadRequestInput): Promise<AttachmentUploadTraceByKeyRow> {
 		const now = input.requestedAt ?? new Date();
 		const row: AttachmentUploadTraceByKeyRow = {
@@ -89,7 +110,7 @@ export class AttachmentUploadTraceRepository {
 		attachmentId: AttachmentID,
 	): Promise<AttachmentUploadTraceByAttachmentRow | null> {
 		const existing = await this.getByUploadKey(uploadKey);
-		if (!existing) {
+		if (!existing || existing.attachment_id != null) {
 			return null;
 		}
 		const now = new Date();

fluxer_api/src/api/channel/services/AttachmentUploadService.ts

@@ -278,7 +278,20 @@ export class AttachmentUploadService {
 		await this.getUploadPermissionAndLimit({userId, channelId});
 		const bucket = Config.s3.buckets.uploads;
 		return Promise.all(
-			uploads.map(async ({upload_filename, upload_id}) => {
+			uploads.map(async ({upload_filename, upload_id}, index) => {
+				const pendingUpload = await this.attachmentUploadTraceRepository.getPendingUpload({
+					uploadKey: upload_filename,
+					userId,
+					channelId,
+					uploadMode: 'presigned_multipart',
+				});
+				if (!pendingUpload) {
+					throw InputValidationError.fromCode(
+						`uploads.${index}.upload_filename`,
+						ValidationErrorCodes.UPLOADED_ATTACHMENT_NOT_FOUND,
+						{filename: upload_filename},
+					);
+				}
 				const parts = await runAttachmentStorageOperation(() =>
 					this.storageService.listParts({
 						bucket,

fluxer_api/src/api/channel/services/ChannelService.ts

@@ -142,6 +142,7 @@ export class ChannelService {
 			favoriteMemeRepository,
 			guildAuditLogService,
 			messagePersistenceService,
+			attachmentUploadTraceRepository,
 			limitConfigService,
 			directMessageSpamMitigationService,
 		);

fluxer_api/src/api/channel/services/MessageService.ts

@@ -18,6 +18,7 @@ import type {IUserRepository} from '../../user/IUserRepository';
 import type {DirectMessageSpamMitigationService} from '../../user/services/DirectMessageSpamMitigationService';
 import type {WorkerTaskName} from '../../worker/WorkerLaneConfig';
 import type {IChannelRepositoryAggregate} from '../repositories/IChannelRepositoryAggregate';
+import type {AttachmentUploadTraceRepository} from '../repositories/message/AttachmentUploadTraceRepository';
 import {MessageAnonymizationService} from './message/MessageAnonymizationService';
 import {MessageChannelAuthService} from './message/MessageChannelAuthService';
 import {MessageDeleteService} from './message/MessageDeleteService';
@@ -66,6 +67,7 @@ export class MessageService {
 		favoriteMemeRepository: IFavoriteMemeRepository,
 		guildAuditLogService: GuildAuditLogService,
 		persistenceService: MessagePersistenceService,
+		attachmentUploadTraceRepository: AttachmentUploadTraceRepository,
 		limitConfigService: LimitConfigService,
 		directMessageSpamMitigationService: DirectMessageSpamMitigationService,
 	) {
@@ -125,6 +127,7 @@ export class MessageService {
 			processingService: this.processing,
 			dispatchService: this.dispatch,
 			embedAttachmentResolver: this.persistence.getEmbedAttachmentResolver(),
+			attachmentUploadTraceRepository,
 			operationsHelpers,
 			limitConfigService,
 			directMessageSpamMitigationService,

fluxer_api/src/api/channel/services/message/AttachmentProcessingService.ts

@@ -12,7 +12,7 @@ import type {GuildResponse} from '@fluxer/schema/src/domains/guild/GuildResponse
 import {isSupportedMediaContentType} from '@pkgs/mime_utils/src/ContentTypeUtils';
 import type {IVirusScanService} from '@pkgs/virus_scan/src/IVirusScanService';
 import {temporaryFile} from 'tempy';
-import {createAttachmentID} from '../../../BrandedTypes';
+import {createAttachmentID, type UserID} from '../../../BrandedTypes';
 import {Config} from '../../../Config';
 import type {MessageAttachment} from '../../../database/types/MessageTypes';
 import {contentModerationService, type ModerationContext} from '../../../infrastructure/ContentModerationService';
@@ -54,6 +54,7 @@ interface ProcessAttachmentParams {
 	message: Message;
 	attachment: AttachmentToProcess;
 	index: number;
+	uploadUserId: UserID;
 	channel?: Channel;
 	guild?: GuildResponse | null;
 	member?: GuildMemberResponse | null;
@@ -88,6 +89,7 @@ export class AttachmentProcessingService {
 	async computeAttachments(params: {
 		message: Message;
 		attachments: Array<AttachmentToProcess>;
+		uploadUserId: UserID;
 		channel?: Channel;
 		guild?: GuildResponse | null;
 		member?: GuildMemberResponse | null;
@@ -105,6 +107,7 @@ export class AttachmentProcessingService {
 					message: params.message,
 					attachment,
 					index,
+					uploadUserId: params.uploadUserId,
 					channel: params.channel,
 					guild: params.guild,
 					member: params.member,
@@ -132,26 +135,30 @@ export class AttachmentProcessingService {
 				}
 			}),
 		);
-		await Promise.all(
-			results.map(async (result) => {
-				const bound = await this.attachmentUploadTraceRepository.bindAttachment(
+		const bindingResults = await Promise.all(
+			results.map(async (result, index) => ({
+				index,
+				result,
+				bound: await this.attachmentUploadTraceRepository.bindAttachment(
 					result.copyOperation.sourceKey,
 					result.attachment.attachment_id,
-				);
-				if (!bound) {
-					Logger.warn(
-						{
-							attachmentId: result.attachment.attachment_id.toString(),
-							uploadKey: result.copyOperation.sourceKey,
-						},
-						'Missing attachment upload trace while binding processed attachment',
-					);
-				}
-			}),
+				),
+			})),
 		);
 		for (const result of results) {
 			void this.deleteUploadObject(result.copyOperation.sourceBucket, result.copyOperation.sourceKey);
 		}
+		const unboundResult = bindingResults.find(({bound}) => bound === null);
+		if (unboundResult) {
+			for (const result of results) {
+				this.deleteUploadObject(result.copyOperation.destinationBucket, result.copyOperation.destinationKey);
+			}
+			throw InputValidationError.fromCode(
+				`attachments.${unboundResult.index}.upload_filename`,
+				ValidationErrorCodes.UPLOADED_ATTACHMENT_NOT_FOUND,
+				{filename: unboundResult.result.attachment.filename},
+			);
+		}
 		const processedAttachments: Array<MessageAttachment> = results.map((result, index) => {
 			const finalObject = copyResults[index];
 			if (result.applyFinalObjectMetadata && finalObject) {
@@ -171,6 +178,18 @@ export class AttachmentProcessingService {
 
 	private async processAttachment(params: ProcessAttachmentParams): Promise<ProcessedAttachment> {
 		const {message, attachment, index, nsfwMode} = params;
+		const pendingUpload = await this.attachmentUploadTraceRepository.getPendingUpload({
+			uploadKey: attachment.upload_filename,
+			userId: params.uploadUserId,
+			channelId: message.channelId,
+		});
+		if (!pendingUpload) {
+			throw InputValidationError.fromCode(
+				`attachments.${index}.upload_filename`,
+				ValidationErrorCodes.UPLOADED_ATTACHMENT_NOT_FOUND,
+				{filename: attachment.filename},
+			);
+		}
 		const uploadedFile = await this.storageService.getObjectMetadata(
 			Config.s3.buckets.uploads,
 			attachment.upload_filename,

fluxer_api/src/api/channel/services/message/MessageEditService.ts

@@ -66,13 +66,24 @@ export class MessageEditService {
 			userId,
 			channelId,
 		});
-		const [canEmbedLinks, canMentionEveryone] = await Promise.all([
+		const hasNewAttachments =
+			data.attachments?.some(
+				(attachment) =>
+					'upload_filename' in attachment &&
+					typeof attachment.upload_filename === 'string' &&
+					attachment.upload_filename.length > 0,
+			) ?? false;
+		const [canEmbedLinks, canMentionEveryone, canAttachFiles] = await Promise.all([
 			hasPermission(Permissions.EMBED_LINKS),
 			hasPermission(Permissions.MENTION_EVERYONE),
+			hasPermission(Permissions.ATTACH_FILES),
 		]);
 		if (data.embeds && data.embeds.length > 0 && !canEmbedLinks) {
 			throw new MissingPermissionsError();
 		}
+		if (hasNewAttachments && !canAttachFiles) {
+			throw new MissingPermissionsError();
+		}
 		if (isOperationDisabled(guild, GuildOperations.SEND_MESSAGE)) {
 			throw new FeatureTemporarilyDisabledError();
 		}
@@ -155,6 +166,7 @@ export class MessageEditService {
 				channel,
 				guild,
 				member,
+				attachmentUploadUserId: userId,
 				allowEmbeds: canEmbedLinks,
 				isBot: user?.isBot,
 				isBugHunterBot,

fluxer_api/src/api/channel/services/message/MessagePersistenceService.ts

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import assert from 'node:assert/strict';
 import {MessageFlags, Permissions, SENDABLE_MESSAGE_FLAGS} from '@fluxer/constants/src/ChannelConstants';
 import {UserFlags} from '@fluxer/constants/src/UserConstants';
 import {ValidationErrorCodes} from '@fluxer/constants/src/ValidationErrorCodes';
@@ -84,6 +85,7 @@ interface CreateMessageParams {
 	flags: number;
 	embeds?: Array<RichEmbedRequest>;
 	attachments?: Array<AttachmentToProcess>;
+	attachmentUploadUserId?: UserID;
 	processedAttachments?: Array<MessageAttachment>;
 	stickerIds?: Array<StickerID>;
 	messageReference?: MessageReference;
@@ -292,12 +294,18 @@ export class MessagePersistenceService {
 		if (!params.attachments || params.attachments.length === 0) {
 			return null;
 		}
+		const uploadUserId = params.attachmentUploadUserId;
+		assert(
+			uploadUserId !== undefined,
+			'Attachment upload actor must be resolved before processing new attachments',
+		);
 		return this.attachmentService.computeAttachments({
 			message: {
 				id: params.messageId,
 				channelId: params.channelId,
 			} as Message,
 			attachments: params.attachments,
+			uploadUserId,
 			channel: params.channel,
 			guild: params.guild,
 			member: params.member,
@@ -412,6 +420,7 @@ export class MessagePersistenceService {
 		guild: GuildResponse | null;
 		member?: GuildMemberResponse | null;
 		allowEmbeds?: boolean;
+		attachmentUploadUserId?: UserID;
 		isBot?: boolean;
 		isBugHunterBot?: boolean;
 		locale?: string | null;
@@ -493,9 +502,15 @@ export class MessagePersistenceService {
 				}
 				let processedNewAttachments: Array<MessageAttachment> = [];
 				if (newAttachments.length > 0) {
+					const uploadUserId = params.attachmentUploadUserId;
+					assert(
+						uploadUserId !== undefined,
+						'Attachment upload actor must be resolved before processing new attachments',
+					);
 					const attachmentResult = await this.attachmentService.computeAttachments({
 						message,
 						attachments: newAttachments,
+						uploadUserId,
 						channel,
 						guild,
 						member,

fluxer_api/src/api/channel/services/message/MessageRequestParser.ts

@@ -20,6 +20,7 @@ import {createLimitMatchContext} from '../../../limits/LimitMatchContextBuilder'
 import type {User} from '../../../models/User';
 import type {HonoEnv} from '../../../types/HonoEnv';
 import {parseJsonPreservingLargeIntegers} from '../../../utils/LosslessJsonParser';
+import {inputValidationErrorFromZodIssues} from '../../../Validator';
 import {type AttachmentRequestData, mergeUploadWithClientData, type UploadedAttachment} from '../../AttachmentDTOs';
 import type {IChannelRepository} from '../../IChannelRepository';
 import type {MessageRequest, MessageUpdateRequest} from '../../MessageTypes';
@@ -54,7 +55,7 @@ export async function parseMultipartMessageData(
 	options?.onPayloadParsed?.(mergedJsonData);
 	const validationResult = schema.safeParse(mergedJsonData);
 	if (!validationResult.success) {
-		throw InputValidationError.fromCode('message_data', ValidationErrorCodes.INVALID_MESSAGE_DATA);
+		throw inputValidationErrorFromZodIssues(validationResult.error.issues);
 	}
 	const data = validationResult.data as Partial<MessageRequest> &
 		Partial<MessageUpdateRequest> & {