feat(files): content-verify direct uploads and pin pending files to octet-stream (#22533)

## Context Follow-up to #22449 (direct-to-storage upload endpoints). In that flow `createFileUpload` inserts a `PENDING` file record before any bytes exist, and until now it guessed the mime type from the **filename extension** — an untrusted, client-controlled value. This PR makes a pending file opaque and only trusts a mime type that was verified against the actual stored bytes. ## What this does **1. A pending file is always `application/octet-stream`.** `createFileUpload` records the pending file — and signs the presigned PUT — as `application/octet-stream`. The extension is still kept on the stored object name so the content can be checked against it later. **2. Content verification at completion.** `completeFileUpload`, after the existing size check, reads a **bounded prefix** of the stored object (`readReadablePrefix`, capped at 64 KiB — a large object is never buffered in full) and runs the existing `extractFileInfoOrThrow` util to detect the real mime type from the content. It: - writes the detected type alongside `status = UPLOADED`, and - rejects a file whose bytes don't match its declared extension (the record stays `PENDING`, so it can never be served or attached, and is reaped by the pending-file cleanup cron). Serving already overrides `Content-Type` from the DB record, so storing the object as octet-stream is fine. **3. A database constraint as backstop.** `CHK_FILE_PENDING_MIME_OCTET_STREAM` — `"status" != 'PENDING' OR "mimeType" = 'application/octet-stream'` — added to `FileEntity` and applied by a fast instance command (`2-19`). It is added `NOT VALID` on purpose: an instance freshly upgraded past #22449 may still hold `PENDING` rows whose mime came from the old extension-guess path, and `NOT VALID` enforces the invariant on every new/updated row without failing on that legacy backlog (those rows get overwritten to octet-stream when completed — `status` flips to `UPLOADED`, so the check passes — or are reaped while pending). ## Tests - `read-readable-prefix.spec.ts` — prefix reader: short source, early stop on a large source (asserts it tears the stream down without draining it), error propagation, empty stream. - `file-upload.service.spec.ts` — create records octet-stream; complete sniffs and sets the detected type, overrides a spoofed extension with the real content type, and rejects content that can't be matched to the declared extension. - `direct-file-upload.integration-spec.ts` — end-to-end case rejecting a `.png` upload whose bytes are plain text. ## Verification `typecheck` green, `lint:diff-with-main` clean, unit suites pass (17 tests). No GraphQL schema change, so no codegen drift. ## Scope Server-only, part of the incremental direct-upload rollout being split into small PRs. Independent of the reaper-cron PR (#22531). https://claude.ai/code/session_015UH8KWmsB9zdYaog8MFG1d --- _Generated by [Claude Code](https://claude.ai/code/session_015UH8KWmsB9zdYaog8MFG1d)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22533?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-06 10:12:30 +02:00
parent e23f82f700
commit d6b6962604
9 changed files with 297 additions and 18 deletions
@@ -0,0 +1,24 @@
+import { QueryRunner } from 'typeorm';
+
+import { RegisteredInstanceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
+import { FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface';
+
+@RegisteredInstanceCommand('2.19.0', 1783094691548)
+export class AddPendingMimeCheckToFileFastInstanceCommand
+  implements FastInstanceCommand
+{
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "core"."file" DROP CONSTRAINT IF EXISTS "CHK_FILE_PENDING_MIME_OCTET_STREAM"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."file" ADD CONSTRAINT "CHK_FILE_PENDING_MIME_OCTET_STREAM" CHECK ("status" != 'PENDING' OR "mimeType" = 'application/octet-stream') NOT VALID`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "core"."file" DROP CONSTRAINT IF EXISTS "CHK_FILE_PENDING_MIME_OCTET_STREAM"`,
+    );
+  }
+}
@@ -92,6 +92,7 @@ import { AddViewKanbanColumnWidthFastInstanceCommand } from './2-15/2-15-instanc
 import { AddPendingQuestionMessageIdToAgentChatThreadFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-19/2-19-instance-command-fast-1782999138000-add-pending-question-to-agent-chat-thread';
 import { AddWorkspaceDiscoverabilityToWorkspaceFastInstanceCommand } from './2-19/2-19-instance-command-fast-1783004140000-add-workspace-discoverability-to-workspace';
 import { AddStatusToFileFastInstanceCommand } from './2-19/2-19-instance-command-fast-1783082964705-add-status-to-file';
+import { AddPendingMimeCheckToFileFastInstanceCommand } from './2-19/2-19-instance-command-fast-1783094691548-add-pending-mime-check-to-file';
 import { DropMetadataStandardOverridesColumnFastInstanceCommand } from './2-20/2-20-instance-command-fast-1825000000000-drop-metadata-standard-overrides-column';
 import { AddLogoToApplicationRegistrationFastInstanceCommand } from './2-19/2-19-instance-command-fast-1783069672191-add-logo-to-application-registration';
 import { BackfillLogoOnApplicationRegistrationSlowInstanceCommand } from './2-19/2-19-instance-command-slow-1783069673191-backfill-logo-on-application-registration';
@@ -196,4 +197,5 @@ export const INSTANCE_COMMANDS = [
  AddDisplayFieldsToApplicationRegistrationFastInstanceCommand,
  BackfillDisplayFieldsOnApplicationRegistrationSlowInstanceCommand,
  AddStatusToFileFastInstanceCommand,
+  AddPendingMimeCheckToFileFastInstanceCommand,
 ];
@@ -1,4 +1,5 @@
 import {
+  Check,
  Column,
  CreateDateColumn,
  DeleteDateColumn,
@@ -23,6 +24,10 @@ import { WasIntroducedInUpgrade } from 'src/engine/core-modules/upgrade/decorato
 import { WorkspaceRelatedEntity } from 'src/engine/workspace-manager/types/workspace-related-entity';

@Entity('file')
+@Check(
+  'CHK_FILE_PENDING_MIME_OCTET_STREAM',
+  `"status" != 'PENDING' OR "mimeType" = 'application/octet-stream'`,
+)
@Index('IDX_FILE_WORKSPACE_ID', ['workspaceId'])
@Index('IDX_FILE_STATUS', ['status'])
@Unique('IDX_APPLICATION_PATH_WORKSPACE_ID_APPLICATION_ID_UNIQUE', [
@@ -0,0 +1 @@
+export const FILE_CONTENT_SNIFF_BYTE_COUNT = 64 * 1024;
@@ -1,3 +1,5 @@
+import { Readable } from 'stream';
+
 import { Test, type TestingModule } from '@nestjs/testing';
 import { getRepositoryToken } from '@nestjs/typeorm';

@@ -31,6 +33,7 @@ describe('FileUploadService', () => {
    getPresignedUploadUrl: jest.fn(),
    getFileMetadata: jest.fn(),
    writeFileStream: jest.fn(),
+    readFile: jest.fn(),
  };

  const fileUrlService = {
@@ -153,13 +156,13 @@ describe('FileUploadService', () => {
        expect.objectContaining({
          fileId: 'mocked-file-id',
          size: 1024,
-          mimeType: 'application/pdf',
+          mimeType: 'application/octet-stream',
          resourcePath: 'field-metadata-uid/mocked-file-id.pdf',
          settings: { isTemporaryFile: true, toDelete: false },
        }),
      );
      expect(result.uploadUrl).toBe('https://bucket/presigned-put');
-      expect(result.contentType).toBe('application/pdf');
+      expect(result.contentType).toBe('application/octet-stream');
      expect(result.fileId).toBe('mocked-file-id');
    });

@@ -193,12 +196,18 @@ describe('FileUploadService', () => {
      path: 'files-field/field-metadata-uid/file-id.pdf',
      size: 1024,
      applicationId: 'application-id',
-      mimeType: 'application/pdf',
+      mimeType: 'application/octet-stream',
      status: FILE_STATUS.PENDING,
      settings: { isTemporaryFile: true, toDelete: false },
      createdAt: new Date(),
    };

+    const PDF_BYTES = Buffer.from('%PDF-1.4\n%\xE2\xE3\xCF\xD3\n', 'latin1');
+    const PNG_BYTES = Buffer.from(
+      'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==',
+      'base64',
+    );
+
    it('should throw when the file record does not exist', async () => {
      fileRepository.findOne.mockResolvedValueOnce(null);

@@ -240,21 +249,72 @@ describe('FileUploadService', () => {
      expect(fileRepository.update).not.toHaveBeenCalled();
    });

-    it('should flip the file to UPLOADED when the stored size matches', async () => {
+    it('should sniff the content, set the detected mime and flip to UPLOADED', async () => {
      fileRepository.findOne.mockResolvedValueOnce(pendingFile);
      fileStorageService.getFileMetadata.mockResolvedValueOnce({ size: 1024 });
+      fileStorageService.readFile.mockResolvedValueOnce(
+        Readable.from(PDF_BYTES),
+      );

      const result = await service.completeFileUpload({
        workspaceId: 'workspace-id',
        fileId: 'file-id',
      });

+      expect(fileStorageService.readFile).toHaveBeenCalledWith(
+        expect.objectContaining({
+          fileFolder: FileFolder.FilesField,
+          resourcePath: 'field-metadata-uid/file-id.pdf',
+          workspaceId: 'workspace-id',
+        }),
+      );
+      expect(fileRepository.update).toHaveBeenCalledWith(
+        'workspace-id',
+        { id: 'file-id' },
+        { status: FILE_STATUS.UPLOADED, mimeType: 'application/pdf' },
+      );
+      expect(result.url).toBe('https://signed-url');
+    });
+
+    it('should override a spoofed extension with the detected content type', async () => {
+      fileRepository.findOne.mockResolvedValueOnce({
+        ...pendingFile,
+        path: 'files-field/field-metadata-uid/file-id.pdf',
+      });
+      fileStorageService.getFileMetadata.mockResolvedValueOnce({ size: 1024 });
+      fileStorageService.readFile.mockResolvedValueOnce(
+        Readable.from(PNG_BYTES),
+      );
+
+      await service.completeFileUpload({
+        workspaceId: 'workspace-id',
+        fileId: 'file-id',
+      });
+
      expect(fileRepository.update).toHaveBeenCalledWith(
        'workspace-id',
        { id: 'file-id' },
-        { status: FILE_STATUS.UPLOADED },
+        { status: FILE_STATUS.UPLOADED, mimeType: 'image/png' },
      );
-      expect(result.url).toBe('https://signed-url');
+    });
+
+    it('should reject when the content cannot be matched to the declared extension', async () => {
+      fileRepository.findOne.mockResolvedValueOnce({
+        ...pendingFile,
+        path: 'files-field/field-metadata-uid/file-id.png',
+      });
+      fileStorageService.getFileMetadata.mockResolvedValueOnce({ size: 1024 });
+      fileStorageService.readFile.mockResolvedValueOnce(
+        Readable.from(Buffer.from('not-an-image-just-text')),
+      );
+
+      await expect(
+        service.completeFileUpload({
+          workspaceId: 'workspace-id',
+          fileId: 'file-id',
+        }),
+      ).rejects.toThrow();
+      expect(fileRepository.update).not.toHaveBeenCalled();
    });

    it('should be idempotent when the file is already UPLOADED', async () => {
@@ -7,7 +7,6 @@ import { pipeline } from 'stream/promises';
 import { msg } from '@lingui/core/macro';
 import { isNonEmptyString } from '@sniptt/guards';
 import bytes from 'bytes';
-import { lookup } from 'mrmime';
 import { FileFolder } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
 import { Repository } from 'typeorm';
@@ -19,9 +18,9 @@ import { ApplicationService } from 'src/engine/core-modules/application/applicat
 import { FileUploadTokenJwtPayload } from 'src/engine/core-modules/auth/types/file-upload-token-jwt-payload.type';
 import { JwtTokenTypeEnum } from 'src/engine/core-modules/auth/types/jwt-token-type.enum';
 import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service';
-import { TWENTY_MIME_POLICY } from 'src/engine/core-modules/file/constants/twenty-mime-policy.constant';
 import { FileWithSignedUrlDTO } from 'src/engine/core-modules/file/dtos/file-with-sign-url.dto';
 import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity';
+import { FILE_CONTENT_SNIFF_BYTE_COUNT } from 'src/engine/core-modules/file/file-upload/constants/file-content-sniff.constant';
 import { FileUploadTargetDTO } from 'src/engine/core-modules/file/file-upload/dtos/file-upload-target.dto';
 import {
  FileUploadException,
@@ -30,12 +29,14 @@ import {
 import { FileUrlService } from 'src/engine/core-modules/file/file-url/file-url.service';
 import { FILE_STATUS } from 'src/engine/core-modules/file/types/file-status.types';
 import { buildFileInfo } from 'src/engine/core-modules/file/utils/build-file-info.utils';
+import { extractFileInfoOrThrow } from 'src/engine/core-modules/file/utils/extract-file-info-or-throw.utils';
 import { removeFileFolderFromFileEntityPath } from 'src/engine/core-modules/file/utils/remove-file-folder-from-file-entity-path.utils';
 import { JwtWrapperService } from 'src/engine/core-modules/jwt/services/jwt-wrapper.service';
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
 import { InjectWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/inject-workspace-scoped-repository.decorator';
 import { WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository';
+import { readReadablePrefix } from 'src/utils/read-readable-prefix';

 export const DIRECT_UPLOAD_FILE_FOLDERS = [
  FileFolder.FilesField,
@@ -100,14 +101,7 @@ export class FileUploadService {
    }

    const { ext } = buildFileInfo(filename);
-    // The file content cannot be sniffed before it reaches storage, so the
-    // mime type is derived from the extension only. Anything unknown is
-    // stored as octet-stream, and the serving path already forces
-    // Content-Disposition: attachment for non-inline-safe mime types.
-    const mimeType =
-      TWENTY_MIME_POLICY[ext] ??
-      (isNonEmptyString(ext) ? lookup(ext) : undefined) ??
-      'application/octet-stream';
+    const mimeType = 'application/octet-stream';

    const fileId = v4();
    const name = `${fileId}${isNonEmptyString(ext) ? `.${ext}` : ''}`;
@@ -361,19 +355,60 @@ export class FileUploadService {
      );
    }

+    const mimeType = await this.detectUploadedMimeTypeOrThrow({
+      fileFolder: fileFolder as FileFolder,
+      applicationUniversalIdentifier: application.universalIdentifier,
+      workspaceId,
+      resourcePath,
+      filename: file.path,
+    });
+
    await this.fileRepository.update(
      workspaceId,
      { id: fileId },
-      { status: FILE_STATUS.UPLOADED },
+      { status: FILE_STATUS.UPLOADED, mimeType },
    );

    return this.toFileWithSignedUrl({
-      file: { ...file, status: FILE_STATUS.UPLOADED },
+      file: { ...file, status: FILE_STATUS.UPLOADED, mimeType },
      fileFolder: fileFolder as FileFolder,
      workspaceId,
    });
  }

+  private async detectUploadedMimeTypeOrThrow({
+    fileFolder,
+    applicationUniversalIdentifier,
+    workspaceId,
+    resourcePath,
+    filename,
+  }: {
+    fileFolder: FileFolder;
+    applicationUniversalIdentifier: string;
+    workspaceId: string;
+    resourcePath: string;
+    filename: string;
+  }): Promise<string> {
+    const stream = await this.fileStorageService.readFile({
+      fileFolder,
+      applicationUniversalIdentifier,
+      workspaceId,
+      resourcePath,
+    });
+
+    const prefix = await readReadablePrefix(
+      stream,
+      FILE_CONTENT_SNIFF_BYTE_COUNT,
+    );
+
+    const { mimeType } = await extractFileInfoOrThrow({
+      file: prefix,
+      filename,
+    });
+
+    return mimeType;
+  }
+
  private async resolveUploadLocation({
    workspaceId,
    fileFolder,
@@ -0,0 +1,62 @@
+import { Readable } from 'stream';
+
+import { readReadablePrefix } from 'src/utils/read-readable-prefix';
+
+describe('readReadablePrefix', () => {
+  it('should return the whole content when it is shorter than the limit', async () => {
+    const prefix = await readReadablePrefix(
+      Readable.from(Buffer.from('hello')),
+      1024,
+    );
+
+    expect(prefix.toString()).toBe('hello');
+  });
+
+  it('should stop at the limit and not buffer the rest of a large source', async () => {
+    let producedBytes = 0;
+
+    const stream = new Readable({
+      read() {
+        producedBytes += 1024;
+        this.push(Buffer.alloc(1024, 0x61));
+
+        if (producedBytes >= 1024 * 1024) {
+          this.push(null);
+        }
+      },
+    });
+
+    const prefix = await readReadablePrefix(stream, 4096);
+
+    expect(prefix.length).toBe(4096);
+    expect(stream.destroyed).toBe(true);
+    expect(producedBytes).toBeLessThan(1024 * 1024);
+  });
+
+  it('should bound the buffer to maxBytes even when a single chunk overshoots', async () => {
+    const prefix = await readReadablePrefix(
+      Readable.from(Buffer.alloc(64 * 1024, 0x61)),
+      4096,
+    );
+
+    expect(prefix.length).toBe(4096);
+  });
+
+  it('should reject when the stream errors before the limit', async () => {
+    const stream = new Readable({
+      read() {
+        this.destroy(new Error('storage exploded'));
+      },
+    });
+
+    await expect(readReadablePrefix(stream, 4096)).rejects.toThrow(
+      'storage exploded',
+    );
+  });
+
+  it('should return an empty buffer for an empty stream', async () => {
+    const prefix = await readReadablePrefix(Readable.from([]), 4096);
+
+    expect(prefix.length).toBe(0);
+  });
+});
@@ -0,0 +1,55 @@
+import { type Readable } from 'stream';
+
+export const readReadablePrefix = async (
+  stream: Readable,
+  maxBytes: number,
+): Promise<Buffer> => {
+  const chunks: Buffer[] = [];
+  let collected = 0;
+  let settled = false;
+
+  return new Promise((resolve, reject) => {
+    const onData = (chunk: Buffer) => {
+      if (settled) {
+        return;
+      }
+
+      const remaining = maxBytes - collected;
+      const boundedChunk =
+        chunk.length > remaining ? chunk.subarray(0, remaining) : chunk;
+
+      chunks.push(boundedChunk);
+      collected += boundedChunk.length;
+
+      if (collected >= maxBytes) {
+        settled = true;
+        stream.off('data', onData);
+        stream.off('end', onEnd);
+        stream.destroy();
+        resolve(Buffer.concat(chunks));
+      }
+    };
+
+    const onEnd = () => {
+      if (settled) {
+        return;
+      }
+
+      settled = true;
+      resolve(Buffer.concat(chunks));
+    };
+
+    const onError = (error: Error) => {
+      if (settled) {
+        return;
+      }
+
+      settled = true;
+      reject(error);
+    };
+
+    stream.on('data', onData);
+    stream.on('end', onEnd);
+    stream.on('error', onError);
+  });
+};
@@ -228,6 +228,41 @@ describe('direct file upload (createFileUpload / completeFileUpload)', () => {
    );
  });

+  it('should reject completing an upload whose content contradicts its extension', async () => {
+    const spoofedContent = Buffer.from('this is not really a png image');
+
+    const createResponse = await createFileUpload({
+      filename: 'actually-text.png',
+      size: spoofedContent.length,
+      fileFolder: 'FilesField',
+      fieldMetadataId: createdFieldMetadataId,
+    });
+
+    const uploadTarget = createResponse.body.data.createFileUpload;
+
+    uploadedFileIds.push(uploadTarget.fileId);
+
+    const putResponse = await putFileToUploadUrl(
+      uploadTarget.uploadUrl,
+      uploadTarget.contentType,
+      spoofedContent,
+    );
+
+    expect(putResponse.status).toBe(204);
+
+    const completeResponse = await completeFileUpload(uploadTarget.fileId);
+
+    expect(completeResponse.body.errors).toBeDefined();
+    expect(completeResponse.body.errors[0].message).toContain(
+      'does not match its extension',
+    );
+
+    const retryResponse = await completeFileUpload(uploadTarget.fileId);
+
+    expect(retryResponse.body.errors).toBeDefined();
+    expect(retryResponse.body.data?.completeFileUpload ?? null).toBeNull();
+  });
+
  it('should refuse a PUT larger than the declared size', async () => {
    const declaredSize = 10;
    const oversizedContent = Buffer.from(
				`@@ -0,0 +1 @@`
				`export const FILE_CONTENT_SNIFF_BYTE_COUNT = 64 * 1024;`