diff --git a/src/http/error-handler.connection.test.ts b/src/http/error-handler.connection.test.ts index 3d23ef11b..2bc059fe6 100644 --- a/src/http/error-handler.connection.test.ts +++ b/src/http/error-handler.connection.test.ts @@ -70,7 +70,7 @@ describe.each(['HTTP', 'S3', 'TUS'] as const)('%s error connections', (protocol) }, onResponseError, }) - await app.register(authenticatedRoutes, { tusServer: tus }) + await app.register(authenticatedRoutes, { tusServer: tus, signed: false }) } else { await app.register(closeConnectionOnError) if (protocol === 'S3') { diff --git a/src/http/plugins/sync-hooks.test.ts b/src/http/plugins/sync-hooks.test.ts index 233556902..93df7993c 100644 --- a/src/http/plugins/sync-hooks.test.ts +++ b/src/http/plugins/sync-hooks.test.ts @@ -162,6 +162,7 @@ describe('sync request lifecycle hooks', () => { register: (app: FastifyInstance) => app.register(publicRoutes, { tusServer: { handle: vi.fn() } as unknown as Server, + signed: false, }), hooks: ['preHandler'], }, diff --git a/src/http/routes/tus/index.test.ts b/src/http/routes/tus/index.test.ts index 2eac18ecd..112036eb7 100644 --- a/src/http/routes/tus/index.test.ts +++ b/src/http/routes/tus/index.test.ts @@ -117,6 +117,7 @@ describe('public tus route request context', () => { rawRes.end() }), } as unknown as Server, + signed: false, }) }) @@ -142,4 +143,9 @@ describe('public tus route request context', () => { sbReqId: 'sb-req-123', }) }) + + it('disposes the db when the response closes', async () => { + await app.inject({ method: 'OPTIONS', url: '/public/object' }) + expect(observedUpload?.db.dispose).toHaveBeenCalled() + }) }) diff --git a/src/http/routes/tus/index.ts b/src/http/routes/tus/index.ts index c13768797..7ef09353f 100644 --- a/src/http/routes/tus/index.ts +++ b/src/http/routes/tus/index.ts @@ -39,6 +39,7 @@ import { onResponseError, onUploadFinish, SIGNED_URL_SUFFIX, + verifySignedUploadRequest, } from './lifecycle' const { @@ -162,6 +163,7 @@ function createTusServer( } const resourceId = UploadId.fromString(uploadId) + await verifySignedUploadRequest(req, resourceId) const bucket = await req.upload.storage .asSuperUser() @@ -210,6 +212,7 @@ export default async function routes(fastify: FastifyInstance) { fastify.register(authenticatedRoutes, { tusServer, + signed: false, }) }) @@ -221,7 +224,7 @@ export default async function routes(fastify: FastifyInstance) { fastify.register(authenticatedRoutes, { tusServer, - operation: '_signed', + signed: true, }) }, { prefix: SIGNED_URL_SUFFIX } @@ -231,6 +234,7 @@ export default async function routes(fastify: FastifyInstance) { fastify.register(async (fastify) => { fastify.register(publicRoutes, { tusServer, + signed: false, }) }) @@ -242,7 +246,7 @@ export default async function routes(fastify: FastifyInstance) { fastify.register(publicRoutes, { tusServer, - operation: '_signed', + signed: true, }) }, { prefix: SIGNED_URL_SUFFIX } @@ -252,7 +256,8 @@ export default async function routes(fastify: FastifyInstance) { function setTusRequestContext( req: FastifyRequest, reply: FastifyReply, - done: HookHandlerDoneFunction + done: HookHandlerDoneFunction, + isSigned: boolean ) { // TUS protocol rejections write directly and skip Fastify's onSend hook. const writeHead = reply.raw.writeHead @@ -262,6 +267,7 @@ function setTusRequestContext( } return Reflect.apply(writeHead, reply.raw, args) } + reply.raw.once('close', () => req.db?.dispose()) ;(req.raw as MultiPartRequest).log = req.log ;(req.raw as MultiPartRequest).upload = { @@ -270,6 +276,7 @@ function setTusRequestContext( owner: req.owner, db: req.db, isUpsert: req.headers['x-upsert'] === 'true', + isSigned, reqId: req.id, sbReqId: req.sbReqId, } @@ -277,7 +284,8 @@ function setTusRequestContext( } export const authenticatedRoutes = fastifyPlugin( - async (fastify: FastifyInstance, options: { tusServer: Server; operation?: string }) => { + async (fastify: FastifyInstance, options: { tusServer: Server; signed: boolean }) => { + const operationSuffix = options.signed ? '_signed' : '' fastify.register(async function authorizationContext(fastify) { fastify.addContentTypeParser('application/offset+octet-stream', (request, payload, done) => done(null) @@ -289,14 +297,16 @@ export const authenticatedRoutes = fastifyPlugin( }) }) - fastify.addHook('preHandler', setTusRequestContext) + fastify.addHook('preHandler', (req, res, done) => + setTusRequestContext(req, res, done, options.signed) + ) fastify.post( '/', { schema: { summary: 'Handle POST request for TUS Resumable uploads', tags: ['resumable'] }, config: { - operation: `${ROUTE_OPERATIONS.TUS_CREATE_UPLOAD}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_CREATE_UPLOAD}${operationSuffix}`, }, }, async (req, res) => { @@ -309,7 +319,7 @@ export const authenticatedRoutes = fastifyPlugin( { schema: { summary: 'Handle POST request for TUS Resumable uploads', tags: ['resumable'] }, config: { - operation: `${ROUTE_OPERATIONS.TUS_CREATE_UPLOAD}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_CREATE_UPLOAD}${operationSuffix}`, }, }, async (req, res) => { @@ -322,7 +332,7 @@ export const authenticatedRoutes = fastifyPlugin( { schema: { summary: 'Handle PUT request for TUS Resumable uploads', tags: ['resumable'] }, config: { - operation: `${ROUTE_OPERATIONS.TUS_UPLOAD_PART}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_UPLOAD_PART}${operationSuffix}`, }, }, async (req, res) => { @@ -337,7 +347,7 @@ export const authenticatedRoutes = fastifyPlugin( tags: ['resumable'], }, config: { - operation: `${ROUTE_OPERATIONS.TUS_UPLOAD_PART}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_UPLOAD_PART}${operationSuffix}`, }, }, async (req, res) => { @@ -349,7 +359,7 @@ export const authenticatedRoutes = fastifyPlugin( { schema: { summary: 'Handle HEAD request for TUS Resumable uploads', tags: ['resumable'] }, config: { - operation: `${ROUTE_OPERATIONS.TUS_GET_UPLOAD}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_GET_UPLOAD}${operationSuffix}`, }, }, async (req, res) => { @@ -364,7 +374,7 @@ export const authenticatedRoutes = fastifyPlugin( tags: ['resumable'], }, config: { - operation: `${ROUTE_OPERATIONS.TUS_DELETE_UPLOAD}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_DELETE_UPLOAD}${operationSuffix}`, }, }, async (req, res) => { @@ -376,13 +386,16 @@ export const authenticatedRoutes = fastifyPlugin( ) export const publicRoutes = fastifyPlugin( - async (fastify: FastifyInstance, options: { tusServer: Server; operation?: string }) => { + async (fastify: FastifyInstance, options: { tusServer: Server; signed: boolean }) => { + const operationSuffix = options.signed ? '_signed' : '' fastify.register(async (fastify) => { fastify.addContentTypeParser('application/offset+octet-stream', (request, payload, done) => done(null) ) - fastify.addHook('preHandler', setTusRequestContext) + fastify.addHook('preHandler', (req, res, done) => + setTusRequestContext(req, res, done, options.signed) + ) fastify.options( '/', @@ -393,7 +406,7 @@ export const publicRoutes = fastifyPlugin( description: 'Handle OPTIONS request for TUS Resumable uploads', }, config: { - operation: `${ROUTE_OPERATIONS.TUS_OPTIONS}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_OPTIONS}${operationSuffix}`, }, }, async (req, res) => { @@ -410,7 +423,7 @@ export const publicRoutes = fastifyPlugin( description: 'Handle OPTIONS request for TUS Resumable uploads', }, config: { - operation: `${ROUTE_OPERATIONS.TUS_OPTIONS}${options.operation || ''}`, + operation: `${ROUTE_OPERATIONS.TUS_OPTIONS}${operationSuffix}`, }, }, async (req, res) => { diff --git a/src/http/routes/tus/lifecycle.test.ts b/src/http/routes/tus/lifecycle.test.ts index 8460e7f42..d6b70f898 100644 --- a/src/http/routes/tus/lifecycle.test.ts +++ b/src/http/routes/tus/lifecycle.test.ts @@ -1,5 +1,3 @@ -import { EventEmitter } from 'node:events' -import type { ServerResponse } from 'node:http' import { logSchema } from '@internal/monitoring' import { Uploader } from '@storage/uploader' import type { DataStore } from '@tus/server' @@ -16,21 +14,16 @@ function createRawTusRequest({ method?: string sbReqId?: string } = {}) { - const response = new EventEmitter() const reqLog = { error: vi.fn(), warn: vi.fn(), } - const dispose = vi.fn() const request = { headers, log: reqLog, method, upload: { - db: { - dispose, - }, isUpsert: false, owner: 'owner-123', storage: { @@ -46,60 +39,20 @@ function createRawTusRequest({ } as unknown as MultiPartRequest return { - dispose, rawReq: { method, runtime: { name: 'node', node: { req: request, - res: response as unknown as ServerResponse, }, }, } as unknown as Parameters[0], reqLog, - response, } } describe('tus lifecycle logging', () => { - it('disposes the db synchronously when the response finishes', async () => { - const { dispose, rawReq, response } = createRawTusRequest({ - method: 'HEAD', - }) - - await onIncomingRequest(rawReq, uploadId, {} as DataStore) - - response.emit('finish') - - expect(dispose).toHaveBeenCalledOnce() - }) - - it('disposes the db when the response closes without finishing', async () => { - const { dispose, rawReq, response } = createRawTusRequest({ - method: 'HEAD', - }) - - await onIncomingRequest(rawReq, uploadId, {} as DataStore) - - response.emit('close') - - expect(dispose).toHaveBeenCalledOnce() - }) - - it('disposes the db only once when a finished response subsequently closes', async () => { - const { dispose, rawReq, response } = createRawTusRequest({ - method: 'HEAD', - }) - - await onIncomingRequest(rawReq, uploadId, {} as DataStore) - - response.emit('finish') - response.emit('close') - - expect(dispose).toHaveBeenCalledOnce() - }) - it('logs upload metadata parse failures with sbReqId through logSchema', async () => { const warningSpy = vi.spyOn(logSchema, 'warning').mockImplementation(() => undefined) const { rawReq, reqLog } = createRawTusRequest({ diff --git a/src/http/routes/tus/lifecycle.ts b/src/http/routes/tus/lifecycle.ts index a131da9e2..1178a3e06 100644 --- a/src/http/routes/tus/lifecycle.ts +++ b/src/http/routes/tus/lifecycle.ts @@ -35,14 +35,6 @@ function getNodeRequest(rawReq: Request): MultiPartRequest { return req } -function getNodeResponse(rawReq: Request) { - const res = rawReq.runtime?.node?.res - if (!res) { - throw ERRORS.InternalError(undefined, 'Response object is missing') - } - return res -} - export type MultiPartRequest = http.IncomingMessage & { executionError?: Error log: FastifyBaseLogger @@ -52,6 +44,8 @@ export type MultiPartRequest = http.IncomingMessage & { db: TenantConnection owner?: string isUpsert: boolean + isSigned: boolean + signatureVerified?: boolean resources?: string[] } } @@ -66,41 +60,40 @@ function getTusError(error: { render(): { statusCode: string; message: string } } } +export async function verifySignedUploadRequest(req: MultiPartRequest, uploadID: UploadId) { + if (!req.upload.isSigned) { + return false + } + + if (req.upload.signatureVerified) { + return true + } + + const signature = req.headers['x-signature'] + if (!signature || typeof signature !== 'string') { + throw ERRORS.InvalidSignature('Missing x-signature header') + } + + const payload = await req.upload.storage + .from(uploadID.bucket) + .verifyObjectSignature(signature, uploadID.objectName, SIGNED_URL_SCOPE_UPLOAD) + + req.upload.owner = payload.owner + req.upload.isUpsert = payload.upsert + req.upload.signatureVerified = true + return true +} + /** * Runs on every TUS incoming request */ export async function onIncomingRequest(rawReq: Request, id: string, datastore: DataStore) { const req = getNodeRequest(rawReq) - const res = getNodeResponse(rawReq) - - const disposeConnection = () => { - // A response can close without finishing when the client disconnects after - // the request body has completed. Remove both listeners so finish + close - // cannot retain the callback or dispose the request lease twice. - res.off('finish', disposeConnection) - res.off('close', disposeConnection) - req.upload.db.dispose() - } - res.once('finish', disposeConnection) - res.once('close', disposeConnection) - const uploadID = UploadId.fromString(id) req.upload.resources = [`${uploadID.bucket}/${uploadID.objectName}`] - // Handle signed url requests - if (req.url?.startsWith(`/upload/resumable/sign`)) { - const signature = req.headers['x-signature'] - if (!signature || (signature && typeof signature !== 'string')) { - throw ERRORS.InvalidSignature('Missing x-signature header') - } - - const payload = await req.upload.storage - .from(uploadID.bucket) - .verifyObjectSignature(signature, uploadID.objectName, SIGNED_URL_SCOPE_UPLOAD) - - req.upload.owner = payload.owner - req.upload.isUpsert = payload.upsert + if (await verifySignedUploadRequest(req, uploadID)) { return } @@ -216,8 +209,7 @@ export function generateUrl( basePath = forwardedPath.replace(/\/+$/, '') + path } - const isSigned = req.url?.endsWith(SIGNED_URL_SUFFIX) - const fullPath = isSigned ? `${basePath}${SIGNED_URL_SUFFIX}` : basePath + const fullPath = req.upload.isSigned ? `${basePath}${SIGNED_URL_SUFFIX}` : basePath if (!parsedPublicUrl && req.headers['x-forwarded-host']) { const port = req.headers['x-forwarded-port'] diff --git a/src/test/tus.test.ts b/src/test/tus.test.ts index 2ea1e7391..f87b8b838 100644 --- a/src/test/tus.test.ts +++ b/src/test/tus.test.ts @@ -1,6 +1,9 @@ +import { once } from 'node:events' import { mkdtemp } from 'node:fs/promises' +import { request as httpRequest, type IncomingMessage } from 'node:http' import { tmpdir } from 'node:os' import path from 'node:path' +import { text } from 'node:stream/consumers' import { CreateBucketCommand, HeadObjectCommand, @@ -58,6 +61,34 @@ function encodeTusMetadata(metadata: Record): string { .join(',') } +async function postRawTusUpload( + baseUrl: string, + target: string, + metadata: Record, + signature?: string +) { + const { hostname, port } = new URL(baseUrl) + const request = httpRequest({ + hostname, + port, + method: 'POST', + path: target, + headers: { + 'tus-resumable': '1.0.0', + 'upload-length': '1', + 'content-type': 'application/offset+octet-stream', + 'upload-metadata': encodeTusMetadata(metadata), + ...(signature ? { 'x-signature': signature } : {}), + }, + }).end('x') + const [response]: IncomingMessage[] = await once(request, 'response') + return { + status: response.statusCode, + body: await text(response), + location: response.headers.location, + } +} + function decodeTusUploadId(location: string): string { const encodedUploadId = location.split('/').pop() @@ -1082,6 +1113,43 @@ describe.each([ expect(response.getStatus()).toEqual(400) } }) + + it('requires a signature for every request target routed to the signed upload scope', async () => { + await storage.createBucket({ id: bucketName, name: bucketName, public: false }) + + const { tusPath } = context.config + const objectName = 'private.txt' + const metadata = { bucketName, objectName, contentType: 'text/plain' } + const targets = [ + `${tusPath}/sign/`, + `${tusPath}/%73ign/`, + `${tusPath}/sig%6E/`, + '/upload/%72esumable/%73ign/', + `${context.baseUrl}${tusPath}/sign/`, + ] + const unsigned = await Promise.all([ + ...targets.map((target) => postRawTusUpload(context.baseUrl, target, metadata)), + postRawTusUpload(context.baseUrl, targets[0], { ...metadata, bucketName: randomUUID() }), + ]) + expect(unsigned.map(({ status, body }) => ({ status, body }))).toEqual( + unsigned.map(() => ({ status: 400, body: 'Missing x-signature header' })) + ) + expect( + await db.findObject(bucketName, objectName, 'id', { dontErrorOnEmpty: true }) + ).toBeUndefined() + + const { token } = await storage + .from(bucketName) + .signUploadObjectUrl(objectName, `${bucketName}/${objectName}`, 3600, undefined, { + upsert: true, + }) + for (const target of [targets[1], targets[4]]) { + const signed = await postRawTusUpload(context.baseUrl, target, metadata, token) + expect(signed.status).toBe(201) + expect(signed.location).toContain(`${tusPath}/sign/`) + } + expect(await db.findObject(bucketName, objectName, 'id')).toBeDefined() + }) }) })