feat(server): improve blob sync (#15367)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Standardized `usePresignedURL` configuration for AWS S3 and Cloudflare
R2 (including `enabled`, `urlPrefix`, and `signKey`).
- Storage upload URL generation now supports both direct provider
presigning and server-mediated proxying based on configuration.
- **Bug Fixes**
- Tightened upload and multipart validation (content type/length checks,
header vs query consistency, and stricter expiration handling).
- Improved fallback behavior when direct upload URL initialization
fails.
- **Tests**
- Updated R2 storage proxy end-to-end coverage to match the new
URL/token behavior.
- **Documentation**
  - Refreshed self-hosted JSON schema guidance for upload URL settings.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/server/src/core/storage/r2-proxy.ts

@@ -1,4 +1,4 @@
-import { createHmac, timingSafeEqual } from 'node:crypto';
+import { timingSafeEqual } from 'node:crypto';
 
 import { Controller, Logger, Put, Req, Res } from '@nestjs/common';
 import type { Request, Response } from 'express';
@@ -7,9 +7,10 @@ import {
   BlobInvalid,
   CallMetric,
   Config,
+  createStorageUploadToken,
   PROXY_MULTIPART_PATH,
   PROXY_UPLOAD_PATH,
-  type R2StorageConfig,
+  type S3StorageConfig,
   STORAGE_PROXY_ROOT,
   type StorageProviderConfig,
   toBuffer,
@@ -19,15 +20,9 @@ import { Public } from '../auth/guard';
 import { StorageRuntimeProvider } from '../storage-runtime';
 import { MULTIPART_PART_SIZE } from './constants';
 
-type R2BlobStorageConfig = StorageProviderConfig & {
-  provider: 'cloudflare-r2';
-  config: R2StorageConfig;
-};
-
 type QueryValue = Request['query'][string];
 
-type R2Config = {
-  storage: R2BlobStorageConfig;
+type UploadProxyConfig = {
   signKey: string;
 };
 
@@ -41,25 +36,17 @@ export class R2UploadController {
     private readonly rt: StorageRuntimeProvider
   ) {}
 
-  private getR2Config(): R2Config {
+  private getUploadProxyConfig(): UploadProxyConfig {
     const storage = this.config.storages.blob.storage as StorageProviderConfig;
-    if (storage.provider !== 'cloudflare-r2') {
+    if (storage.provider !== 'cloudflare-r2' && storage.provider !== 'aws-s3') {
       throw new BlobInvalid('Invalid endpoint');
     }
-    const r2Config = storage.config as R2StorageConfig;
-    const signKey = r2Config.usePresignedURL?.signKey;
-    if (
-      !r2Config.usePresignedURL?.enabled ||
-      !r2Config.usePresignedURL.urlPrefix ||
-      !signKey
-    ) {
+    const uploadConfig = (storage.config as S3StorageConfig).usePresignedURL;
+    const signKey = uploadConfig?.signKey;
+    if (!uploadConfig?.enabled || !signKey) {
       throw new BlobInvalid('Invalid endpoint');
     }
-    return { storage: storage as R2BlobStorageConfig, signKey };
-  }
-
-  private sign(canonical: string, signKey: string) {
-    return createHmac('sha256', signKey).update(canonical).digest('base64');
+    return { signKey };
   }
 
   private safeEqual(expected: string, actual: string) {
@@ -75,55 +62,32 @@ export class R2UploadController {
 
   private verifyToken(
     path: string,
-    canonicalFields: (string | number | undefined)[],
-    exp: number,
+    canonicalFields: (string | number)[],
+    expiresAt: number,
     token: string,
     signKey: string
   ) {
-    const canonical = [
+    const expected = createStorageUploadToken(
       path,
-      ...canonicalFields.map(field =>
-        field === undefined ? '' : field.toString()
-      ),
-      exp.toString(),
-    ].join('\n');
-    const expected = `${exp}-${this.sign(canonical, signKey)}`;
+      canonicalFields,
+      expiresAt,
+      signKey
+    );
 
     return this.safeEqual(expected, token);
   }
 
   private expectString(value: QueryValue, field: string): string {
-    if (Array.isArray(value)) {
-      return String(value[0]);
-    }
     if (typeof value === 'string' && value.length > 0) {
       return value;
     }
     throw new BlobInvalid(`Missing ${field}.`);
   }
 
-  private optionalString(value: QueryValue) {
-    if (Array.isArray(value)) {
-      return String(value[0]);
-    }
-    return typeof value === 'string' && value.length > 0 ? value : undefined;
-  }
-
   private number(value: QueryValue, field: string): number {
     const str = this.expectString(value, field);
     const num = Number(str);
-    if (!Number.isFinite(num)) {
-      throw new BlobInvalid(`Invalid ${field}.`);
-    }
-    return num;
-  }
-
-  private optionalNumber(value: QueryValue, field: string): number | undefined {
-    if (value === undefined) {
-      return undefined;
-    }
-    const num = Number(Array.isArray(value) ? value[0] : value);
-    if (!Number.isFinite(num)) {
+    if (!Number.isSafeInteger(num)) {
       throw new BlobInvalid(`Invalid ${field}.`);
     }
     return num;
@@ -135,15 +99,15 @@ export class R2UploadController {
       return undefined;
     }
     const num = Number(raw);
-    if (!Number.isFinite(num) || num < 0) {
+    if (!Number.isSafeInteger(num) || num < 0) {
       throw new BlobInvalid('Invalid Content-Length header');
     }
     return num;
   }
 
-  private ensureNotExpired(exp: number) {
+  private ensureNotExpired(expiresAt: number) {
     const now = Math.floor(Date.now() / 1000);
-    if (exp < now) {
+    if (expiresAt < now) {
       throw new BlobInvalid('Upload URL expired');
     }
   }
@@ -152,25 +116,31 @@ export class R2UploadController {
   @Put('upload')
   @CallMetric('controllers', 'r2_proxy_upload')
   async upload(@Req() req: Request, @Res() res: Response) {
-    const { signKey } = this.getR2Config();
+    const { signKey } = this.getUploadProxyConfig();
 
     const workspaceId = this.expectString(req.query.workspaceId, 'workspaceId');
     const key = this.expectString(req.query.key, 'key');
     const token = this.expectString(req.query.token, 'token');
-    const exp = this.number(req.query.exp, 'exp');
-    const contentType = this.optionalString(req.query.contentType);
-    const contentLengthFromQuery = this.optionalNumber(
+    const expiresAt = this.number(req.query.expiresAt, 'expiresAt');
+    const contentType = this.expectString(req.query.contentType, 'contentType');
+    const contentLengthFromQuery = this.number(
       req.query.contentLength,
       'contentLength'
     );
+    if (
+      !Number.isInteger(contentLengthFromQuery) ||
+      contentLengthFromQuery < 0
+    ) {
+      throw new BlobInvalid('Invalid content length');
+    }
 
-    this.ensureNotExpired(exp);
+    this.ensureNotExpired(expiresAt);
 
     if (
       !this.verifyToken(
         PROXY_UPLOAD_PATH,
         [workspaceId, key, contentType, contentLengthFromQuery],
-        exp,
+        expiresAt,
         token,
         signKey
       )
@@ -188,7 +158,6 @@ export class R2UploadController {
 
     const contentLengthHeader = this.parseContentLength(req);
     if (
-      contentLengthFromQuery !== undefined &&
       contentLengthHeader !== undefined &&
       contentLengthFromQuery !== contentLengthHeader
     ) {
@@ -196,15 +165,11 @@ export class R2UploadController {
     }
 
     const contentLength = contentLengthHeader ?? contentLengthFromQuery;
-    if (contentLength === undefined) {
-      throw new BlobInvalid('Missing Content-Length header');
-    }
     if (record.size && contentLength !== record.size) {
       throw new BlobInvalid('Content length does not match upload metadata');
     }
 
-    const mime = contentType ?? record.mime;
-    if (record.mime && mime && record.mime !== mime) {
+    if (record.mime && contentType && record.mime !== contentType) {
       throw new BlobInvalid('Mime type mismatch');
     }
 
@@ -213,10 +178,7 @@ export class R2UploadController {
         'blob',
         `${workspaceId}/${key}`,
         await toBuffer(req),
-        {
-          contentType: mime,
-          contentLength,
-        }
+        { contentType, contentLength }
       );
     } catch (error) {
       this.logger.error('Failed to proxy upload', error as Error);
@@ -230,26 +192,36 @@ export class R2UploadController {
   @Put('multipart')
   @CallMetric('controllers', 'r2_proxy_multipart')
   async uploadPart(@Req() req: Request, @Res() res: Response) {
-    const { signKey } = this.getR2Config();
+    const { signKey } = this.getUploadProxyConfig();
 
     const workspaceId = this.expectString(req.query.workspaceId, 'workspaceId');
     const key = this.expectString(req.query.key, 'key');
     const uploadId = this.expectString(req.query.uploadId, 'uploadId');
     const token = this.expectString(req.query.token, 'token');
-    const exp = this.number(req.query.exp, 'exp');
+    const expiresAt = this.number(req.query.expiresAt, 'expiresAt');
     const partNumber = this.number(req.query.partNumber, 'partNumber');
+    const contentLengthFromQuery = this.number(
+      req.query.contentLength,
+      'contentLength'
+    );
 
     if (partNumber < 1) {
       throw new BlobInvalid('Invalid part number');
     }
+    if (
+      !Number.isInteger(contentLengthFromQuery) ||
+      contentLengthFromQuery < 1
+    ) {
+      throw new BlobInvalid('Invalid content length');
+    }
 
-    this.ensureNotExpired(exp);
+    this.ensureNotExpired(expiresAt);
 
     if (
       !this.verifyToken(
         PROXY_MULTIPART_PATH,
-        [workspaceId, key, uploadId, partNumber],
-        exp,
+        [workspaceId, key, uploadId, partNumber, contentLengthFromQuery],
+        expiresAt,
         token,
         signKey
       )
@@ -272,6 +244,9 @@ export class R2UploadController {
     if (contentLength === undefined || contentLength === 0) {
       throw new BlobInvalid('Missing Content-Length header');
     }
+    if (contentLength !== contentLengthFromQuery) {
+      throw new BlobInvalid('Content length mismatch');
+    }
 
     const maxPartNumber = Math.ceil(record.size / MULTIPART_PART_SIZE);
     if (partNumber > maxPartNumber) {

packages/backend/server/src/core/storage/wrappers/blob.ts

@@ -1,17 +1,17 @@
-import { createHmac } from 'node:crypto';
-
 import { Injectable, Logger } from '@nestjs/common';
 
 import {
+  BlobInvalid,
   type BlobOutputType,
   Config,
+  createStorageUploadToken,
   EventBus,
   type GetObjectMetadata,
   OnEvent,
   PROXY_MULTIPART_PATH,
   PROXY_UPLOAD_PATH,
   type PutObjectMetadata,
-  type R2StorageConfig,
+  type S3StorageConfig,
   SIGNED_URL_EXPIRED,
   type StorageProviderConfig,
   URLHelper,
@@ -19,6 +19,7 @@ import {
 import { Models } from '../../../models';
 import type { StorageProviderCapabilities } from '../../../native';
 import { StorageRuntimeProvider } from '../../storage-runtime';
+import { MULTIPART_PART_SIZE } from '../constants';
 
 declare global {
   interface Events {
@@ -50,7 +51,12 @@ type BlobGetResult = {
   metadata?: GetObjectMetadata;
 };
 
-type R2ProxyConfig = {
+type UploadURLConfig = {
+  signKey?: string;
+  urlPrefix?: string;
+};
+
+type UploadProxyConfig = {
   signKey: string;
   urlPrefix: string;
 };
@@ -82,7 +88,17 @@ export class WorkspaceBlobStorage {
 
   async capabilities(): Promise<StorageProviderCapabilities> {
     const capabilities = await this.rt.providerCapabilities('blob');
-    if (!this.r2ProxyConfig()) {
+    const config = this.uploadURLConfig();
+    if (!config) {
+      return {
+        ...capabilities,
+        presignPut: false,
+        multipartDirect: false,
+        proxyUpload: false,
+        serverMediatedOnly: true,
+      };
+    }
+    if (!config.signKey) {
       return capabilities;
     }
     return {
@@ -116,11 +132,22 @@ export class WorkspaceBlobStorage {
     key: string,
     metadata?: PutObjectMetadata
   ) {
-    const proxy = this.r2ProxyConfig();
-    if (proxy) {
-      return this.createProxyUploadUrl(workspaceId, key, metadata, proxy);
+    const config = this.uploadURLConfig();
+    if (!config) return;
+    if (config.signKey) {
+      return this.createProxyUploadUrl(workspaceId, key, metadata, {
+        signKey: config.signKey,
+        urlPrefix: config.urlPrefix ?? this.url.baseUrl,
+      });
     }
-    return this.rt.presignPut('blob', `${workspaceId}/${key}`, metadata);
+    const presigned = await this.rt.presignPut(
+      'blob',
+      `${workspaceId}/${key}`,
+      metadata
+    );
+    return config.urlPrefix && presigned
+      ? this.withURLPrefix(presigned, config.urlPrefix)
+      : presigned;
   }
 
   async createMultipartUpload(
@@ -141,22 +168,36 @@ export class WorkspaceBlobStorage {
     uploadId: string,
     partNumber: number
   ) {
-    const proxy = this.r2ProxyConfig();
-    if (proxy) {
+    const config = this.uploadURLConfig();
+    if (!config) return;
+    const contentLength = await this.multipartPartContentLength(
+      workspaceId,
+      key,
+      uploadId,
+      partNumber
+    );
+    if (config.signKey) {
       return this.createProxyMultipartUrl(
         workspaceId,
         key,
         uploadId,
         partNumber,
-        proxy
+        contentLength,
+        {
+          signKey: config.signKey,
+          urlPrefix: config.urlPrefix ?? this.url.baseUrl,
+        }
       );
     }
-    return this.rt.presignUploadPart(
+    const presigned = await this.rt.presignUploadPart(
       'blob',
       `${workspaceId}/${key}`,
       uploadId,
       partNumber
     );
+    return config.urlPrefix && presigned
+      ? this.withURLPrefix(presigned, config.urlPrefix)
+      : presigned;
   }
 
   async listMultipartUploadParts(
@@ -308,56 +349,38 @@ export class WorkspaceBlobStorage {
     await this.delete(workspaceId, key, true);
   }
 
-  private r2ProxyConfig() {
+  private uploadURLConfig(): UploadURLConfig | undefined {
     const storage = this.config.storages.blob.storage as StorageProviderConfig;
-    if (storage.provider !== 'cloudflare-r2') {
+    if (storage.provider !== 'cloudflare-r2' && storage.provider !== 'aws-s3') {
       return;
     }
-    const r2 = storage.config as R2StorageConfig;
-    const usePresignedURL = r2.usePresignedURL;
-    if (
-      !usePresignedURL?.enabled ||
-      !usePresignedURL.urlPrefix ||
-      !usePresignedURL.signKey
-    ) {
+    const usePresignedURL = (storage.config as S3StorageConfig).usePresignedURL;
+    if (!usePresignedURL?.enabled) {
       return;
     }
     return {
-      signKey: usePresignedURL.signKey,
-      urlPrefix: usePresignedURL.urlPrefix,
+      signKey: usePresignedURL.signKey || undefined,
+      urlPrefix: usePresignedURL.urlPrefix || undefined,
     };
   }
 
-  private signProxy(
-    path: string,
-    canonicalFields: (string | number | undefined)[],
-    exp: number,
-    signKey: string
-  ) {
-    const canonical = [
-      path,
-      ...canonicalFields.map(field =>
-        field === undefined ? '' : field.toString()
-      ),
-      exp.toString(),
-    ].join('\n');
-    return `${exp}-${createHmac('sha256', signKey).update(canonical).digest('base64')}`;
-  }
-
   private createProxyUploadUrl(
     workspaceId: string,
     key: string,
     metadata: PutObjectMetadata | undefined,
-    proxy: R2ProxyConfig
+    proxy: UploadProxyConfig
   ) {
     const contentType = metadata?.contentType ?? 'application/octet-stream';
     const contentLength = metadata?.contentLength;
+    if (contentLength === undefined) {
+      throw new BlobInvalid('Missing upload content length');
+    }
     const expiresAt = new Date(Date.now() + SIGNED_URL_EXPIRED * 1000);
-    const exp = Math.floor(expiresAt.getTime() / 1000);
-    const token = this.signProxy(
+    const expiresAtSeconds = Math.floor(expiresAt.getTime() / 1000);
+    const token = createStorageUploadToken(
       PROXY_UPLOAD_PATH,
       [workspaceId, key, contentType, contentLength],
-      exp,
+      expiresAtSeconds,
       proxy.signKey
     );
     return {
@@ -366,7 +389,7 @@ export class WorkspaceBlobStorage {
         key,
         contentType,
         contentLength,
-        exp,
+        expiresAt: expiresAtSeconds,
         token,
       }),
       headers: {},
@@ -379,14 +402,15 @@ export class WorkspaceBlobStorage {
     key: string,
     uploadId: string,
     partNumber: number,
-    proxy: R2ProxyConfig
+    contentLength: number,
+    proxy: UploadProxyConfig
   ) {
     const expiresAt = new Date(Date.now() + SIGNED_URL_EXPIRED * 1000);
-    const exp = Math.floor(expiresAt.getTime() / 1000);
-    const token = this.signProxy(
+    const expiresAtSeconds = Math.floor(expiresAt.getTime() / 1000);
+    const token = createStorageUploadToken(
       PROXY_MULTIPART_PATH,
-      [workspaceId, key, uploadId, partNumber],
-      exp,
+      [workspaceId, key, uploadId, partNumber, contentLength],
+      expiresAtSeconds,
       proxy.signKey
     );
     return {
@@ -395,7 +419,8 @@ export class WorkspaceBlobStorage {
         key,
         uploadId,
         partNumber,
-        exp,
+        contentLength,
+        expiresAt: expiresAtSeconds,
         token,
       }),
       headers: {},
@@ -418,4 +443,42 @@ export class WorkspaceBlobStorage {
     }
     return url.toString();
   }
+
+  private withURLPrefix<T extends { url: string }>(
+    presigned: T,
+    urlPrefix: string
+  ): T {
+    const url = new URL(presigned.url);
+    const prefix = new URL(urlPrefix);
+    if (prefix.pathname !== '/' || prefix.search || prefix.hash) {
+      throw new BlobInvalid('Upload URL prefix must contain only an origin');
+    }
+    url.protocol = prefix.protocol;
+    url.host = prefix.host;
+    return { ...presigned, url: url.toString() };
+  }
+
+  private async multipartPartContentLength(
+    workspaceId: string,
+    key: string,
+    uploadId: string,
+    partNumber: number
+  ) {
+    const record = await this.models.blob.get(workspaceId, key);
+    if (!record || record.status === 'completed') {
+      throw new BlobInvalid('Multipart upload is not pending');
+    }
+    if (record.uploadId !== uploadId) {
+      throw new BlobInvalid('Upload id mismatch');
+    }
+    const offset = (partNumber - 1) * MULTIPART_PART_SIZE;
+    if (
+      !Number.isInteger(partNumber) ||
+      partNumber < 1 ||
+      offset >= record.size
+    ) {
+      throw new BlobInvalid('Invalid part number');
+    }
+    return Math.min(MULTIPART_PART_SIZE, record.size - offset);
+  }
 }

packages/backend/server/src/core/workspaces/resolvers/blob.ts

@@ -248,63 +248,70 @@ export class WorkspaceBlobResolver {
     }
 
     const metadata = { contentType: mime, contentLength: size };
-    const capabilities = await this.storage.capabilities();
     let init: BlobUploadInit | null = null;
     let uploadIdForRecord: string | null = null;
 
-    // try to resume multipart uploads
-    if (capabilities.multipartDirect && record && record.uploadId) {
-      const uploadedParts = await this.storage.listMultipartUploadParts(
-        workspaceId,
-        key,
-        record.uploadId
-      );
+    try {
+      const capabilities = await this.storage.capabilities();
 
-      if (uploadedParts) {
-        return {
-          method: BlobUploadMethod.MULTIPART,
-          blobKey: key,
-          uploadId: record.uploadId,
-          partSize: MULTIPART_PART_SIZE,
-          uploadedParts,
-        };
-      }
-    }
+      // try to resume multipart uploads
+      if (capabilities.multipartDirect && record && record.uploadId) {
+        const uploadedParts = await this.storage.listMultipartUploadParts(
+          workspaceId,
+          key,
+          record.uploadId
+        );
 
-    if (capabilities.multipartDirect && size >= MULTIPART_THRESHOLD) {
-      const multipart = await this.storage.createMultipartUpload(
-        workspaceId,
-        key,
-        metadata
-      );
-      if (multipart) {
-        uploadIdForRecord = multipart.uploadId;
-        init = {
-          method: BlobUploadMethod.MULTIPART,
-          blobKey: key,
-          uploadId: multipart.uploadId,
-          partSize: MULTIPART_PART_SIZE,
-          expiresAt: multipart.expiresAt,
-          uploadedParts: [],
-        };
+        if (uploadedParts) {
+          return {
+            method: BlobUploadMethod.MULTIPART,
+            blobKey: key,
+            uploadId: record.uploadId,
+            partSize: MULTIPART_PART_SIZE,
+            uploadedParts,
+          };
+        }
       }
-    }
 
-    if (!init && capabilities.presignPut) {
-      const presigned = await this.storage.presignPut(
-        workspaceId,
-        key,
-        metadata
-      );
-      if (presigned) {
-        init = {
-          method: BlobUploadMethod.PRESIGNED,
-          blobKey: key,
-          uploadUrl: presigned.url,
-          headers: presigned.headers,
-          expiresAt: presigned.expiresAt,
-        };
+      if (capabilities.multipartDirect && size >= MULTIPART_THRESHOLD) {
+        const multipart = await this.storage.createMultipartUpload(
+          workspaceId,
+          key,
+          metadata
+        );
+        if (multipart) {
+          uploadIdForRecord = multipart.uploadId;
+          init = {
+            method: BlobUploadMethod.MULTIPART,
+            blobKey: key,
+            uploadId: multipart.uploadId,
+            partSize: MULTIPART_PART_SIZE,
+            expiresAt: multipart.expiresAt,
+            uploadedParts: [],
+          };
+        }
       }
+
+      if (!init && capabilities.presignPut) {
+        const presigned = await this.storage.presignPut(
+          workspaceId,
+          key,
+          metadata
+        );
+        if (presigned) {
+          init = {
+            method: BlobUploadMethod.PRESIGNED,
+            blobKey: key,
+            uploadUrl: presigned.url,
+            headers: presigned.headers,
+            expiresAt: presigned.expiresAt,
+          };
+        }
+      }
+    } catch (error) {
+      this.logger.warn('Failed to initialize direct blob upload', error);
+      init = null;
+      uploadIdForRecord = null;
     }
 
     if (!init) {