fix: support s3 chunked upload (#684)

Author: fenos

src/http/plugins/signature-v4.ts

@@ -7,6 +7,10 @@ import { ERRORS } from '@internal/errors'
 
 import { getConfig } from '../../config'
 import { MultipartFile, MultipartValue } from '@fastify/multipart'
+import {
+  ChunkSignatureV4Parser,
+  V4StreamingAlgorithm,
+} from '@storage/protocols/s3/signature-v4-stream'
 
 const {
   anonKeyAsync,
@@ -27,6 +31,7 @@ type AWSRequest = FastifyRequest<{ Querystring: { 'X-Amz-Credential'?: string }
 declare module 'fastify' {
   interface FastifyRequest {
     multiPartFileStream?: MultipartFile
+    streamingSignatureV4?: ChunkSignatureV4Parser
   }
 }
 
@@ -80,6 +85,15 @@ export const signatureV4 = fastifyPlugin(
         request.jwt = token
         request.jwtPayload = payload
         request.owner = payload.sub
+
+        if (SignatureV4.isChunkedUpload(request.headers)) {
+          request.streamingSignatureV4 = createStreamingSignatureV4Parser({
+            signatureV4,
+            streamAlgorithm: request.headers['x-amz-content-sha256'] as V4StreamingAlgorithm,
+            clientSignature,
+            trailers: request.headers['x-amz-trailer'] as string,
+          })
+        }
         return
       }
 
@@ -92,6 +106,15 @@ export const signatureV4 = fastifyPlugin(
       request.jwt = jwt
       request.jwtPayload = claims
       request.owner = claims.sub
+
+      if (SignatureV4.isChunkedUpload(request.headers)) {
+        request.streamingSignatureV4 = createStreamingSignatureV4Parser({
+          signatureV4,
+          streamAlgorithm: request.headers['x-amz-content-sha256'] as V4StreamingAlgorithm,
+          clientSignature,
+          trailers: request.headers['x-amz-trailer'] as string,
+        })
+      }
     })
   },
   { name: 'auth-signature-v4' }
@@ -202,3 +225,43 @@ async function createServerSignature(tenantId: string, clientSignature: ClientSi
 
   return { signature, claims: undefined, token: await serviceKeyAsync }
 }
+
+interface CreateSignatureV3ParserOpts {
+  signatureV4: SignatureV4
+  streamAlgorithm: string
+  clientSignature: ClientSignature
+  trailers: string
+}
+
+function createStreamingSignatureV4Parser(opts: CreateSignatureV3ParserOpts) {
+  const algorithm = opts.streamAlgorithm as V4StreamingAlgorithm
+  const trailers = opts.trailers
+
+  const chunkedSignatureV4 = new ChunkSignatureV4Parser({
+    maxChunkSize: 8 * 1024 * 1024,
+    maxHeaderLength: 256,
+    streamingAlgorithm: algorithm,
+    trailerHeaderNames: trailers.split(','),
+  })
+
+  chunkedSignatureV4.on(
+    'signatureReadyForVerification',
+    (signature: string, _: number, hash: string, previousSign) => {
+      const isValid = opts.signatureV4.validateChunkSignature(
+        algorithm,
+        opts.clientSignature,
+        hash,
+        signature,
+        previousSign || opts.clientSignature.signature
+      )
+
+      if (!isValid) {
+        throw ERRORS.SignatureDoesNotMatch(
+          'The request signature we calculated does not match the signature you provided. Check your key and signing method.'
+        )
+      }
+    }
+  )
+
+  return chunkedSignatureV4
+}

src/http/routes/s3/commands/put-object.ts

@@ -6,7 +6,7 @@ import { fileUploadFromRequest, getStandardMaxFileSizeLimit } from '@storage/upl
 import { ERRORS } from '@internal/errors'
 import { pipeline } from 'stream/promises'
 import { ByteLimitTransformStream } from '@storage/protocols/s3/byte-limit-stream'
-import stream from 'stream'
+import stream, { PassThrough, Readable } from 'stream'
 
 const PutObjectInput = {
   summary: 'Put Object',
@@ -35,7 +35,6 @@ const PutObjectInput = {
       'content-encoding': { type: 'string' },
       expires: { type: 'string' },
     },
-    required: ['content-length'],
   },
 } as const
 
@@ -80,18 +79,25 @@ export default function PutObject(s3Router: S3Router) {
         fileSizeLimit: bucket.file_size_limit || undefined,
       })
 
-      return s3Protocol.putObject(
-        {
-          Body: uploadRequest.body,
-          Bucket: req.Params.Bucket,
-          Key: key,
-          CacheControl: uploadRequest.cacheControl,
-          ContentType: uploadRequest.mimeType,
-          Expires: req.Headers?.['expires'] ? new Date(req.Headers?.['expires']) : undefined,
-          ContentEncoding: req.Headers?.['content-encoding'],
-          Metadata: metadata,
-        },
-        { signal: ctx.signals.body, isTruncated: uploadRequest.isTruncated }
+      return pipeline(
+        uploadRequest.body,
+        new ByteLimitTransformStream(uploadRequest.maxFileSize),
+        ctx.req.streamingSignatureV4 || new PassThrough(),
+        async (fileStream) => {
+          return s3Protocol.putObject(
+            {
+              Body: fileStream as Readable,
+              Bucket: req.Params.Bucket,
+              Key: key,
+              CacheControl: uploadRequest.cacheControl,
+              ContentType: uploadRequest.mimeType,
+              Expires: req.Headers?.['expires'] ? new Date(req.Headers?.['expires']) : undefined,
+              ContentEncoding: req.Headers?.['content-encoding'],
+              Metadata: metadata,
+            },
+            { signal: ctx.signals.body, isTruncated: uploadRequest.isTruncated }
+          )
+        }
       )
     }
   )

src/http/routes/s3/commands/upload-part.ts

@@ -1,6 +1,8 @@
 import { S3ProtocolHandler } from '@storage/protocols/s3/s3-handler'
 import { S3Router } from '../router'
 import { ROUTE_OPERATIONS } from '../../operations'
+import { pipeline } from 'stream/promises'
+import { PassThrough, Readable } from 'stream'
 
 const UploadPartInput = {
   summary: 'Upload Part',
@@ -25,11 +27,11 @@ const UploadPartInput = {
     properties: {
       host: { type: 'string' },
       'x-amz-content-sha256': { type: 'string' },
+      'x-amz-decoded-content-length': { type: 'integer' },
       'x-amz-date': { type: 'string' },
       'content-type': { type: 'string' },
       'content-length': { type: 'integer' },
     },
-    required: ['content-length'],
   },
 } as const
 
@@ -44,14 +46,39 @@ export default function UploadPart(s3Router: S3Router) {
     (req, ctx) => {
       const s3Protocol = new S3ProtocolHandler(ctx.storage, ctx.tenantId, ctx.owner)
 
-      return s3Protocol.uploadPart({
-        Body: ctx.req.raw,
-        UploadId: req.Querystring?.uploadId,
-        Bucket: req.Params.Bucket,
-        Key: req.Params['*'],
-        PartNumber: req.Querystring?.partNumber,
-        ContentLength: req.Headers?.['content-length'],
-      })
+      if (ctx.req.streamingSignatureV4) {
+        const passThrough = new PassThrough()
+        ctx.req.raw.pipe(passThrough)
+        ctx.req.raw.on('error', (err) => {
+          passThrough.destroy(err)
+        })
+
+        return pipeline(passThrough, ctx.req.streamingSignatureV4, async (body) => {
+          return s3Protocol.uploadPart(
+            {
+              Body: body as Readable,
+              UploadId: req.Querystring?.uploadId,
+              Bucket: req.Params.Bucket,
+              Key: req.Params['*'],
+              ContentLength: req.Headers?.['x-amz-decoded-content-length'],
+              PartNumber: req.Querystring?.partNumber,
+            },
+            { signal: ctx.req.signals.body.signal }
+          )
+        })
+      }
+
+      return s3Protocol.uploadPart(
+        {
+          Body: ctx.req.raw,
+          UploadId: req.Querystring?.uploadId,
+          Bucket: req.Params.Bucket,
+          Key: req.Params['*'],
+          PartNumber: req.Querystring?.partNumber,
+          ContentLength: req.Headers?.['content-length'],
+        },
+        { signal: ctx.req.signals.body.signal }
+      )
     }
   )
 }

src/storage/protocols/s3/s3-handler.ts

@@ -535,7 +535,7 @@ export class S3ProtocolHandler {
    * @param command
    * @param signal
    */
-  async uploadPart(command: UploadPartCommandInput, signal?: AbortSignal) {
+  async uploadPart(command: UploadPartCommandInput, { signal }: { signal?: AbortSignal }) {
     if (signal?.aborted) {
       throw ERRORS.AbortedTerminate('UploadPart aborted')
     }