diff --git a/fluxer_api/src/api/Config.ts b/fluxer_api/src/api/Config.ts index 817ffcf4e..e32b81674 100644 --- a/fluxer_api/src/api/Config.ts +++ b/fluxer_api/src/api/Config.ts @@ -432,6 +432,7 @@ export function buildAPIConfigFromMaster(master: MasterConfig): APIConfig { }, presignedAttachmentUploadsEnabled: master.services.api.presigned_attachment_uploads_enabled ?? false, presignedDownloadsEnabled: master.services.api.presigned_downloads_enabled ?? false, + presignedHarvestDownloadsEnabled: master.services.api.presigned_harvest_downloads_enabled ?? true, attachmentDecayEnabled: master.attachment_decay_enabled, deletionGracePeriodHours: master.dev.test_mode_enabled ? 0.01 : master.deletion_grace_period_hours, inactivityDeletionThresholdDays: master.inactivity_deletion_threshold_days, diff --git a/fluxer_api/src/api/config/APIConfig.ts b/fluxer_api/src/api/config/APIConfig.ts index 9f028aecf..3b685ce92 100644 --- a/fluxer_api/src/api/config/APIConfig.ts +++ b/fluxer_api/src/api/config/APIConfig.ts @@ -330,6 +330,7 @@ export interface APIConfig { }; presignedAttachmentUploadsEnabled: boolean; presignedDownloadsEnabled: boolean; + presignedHarvestDownloadsEnabled: boolean; attachmentDecayEnabled: boolean; deletionGracePeriodHours: number; inactivityDeletionThresholdDays?: number; diff --git a/fluxer_api/src/api/openapi/openapi.json b/fluxer_api/src/api/openapi/openapi.json index 7ce50d0d9..66f718a9e 100644 --- a/fluxer_api/src/api/openapi/openapi.json +++ b/fluxer_api/src/api/openapi/openapi.json @@ -10588,6 +10588,70 @@ ] } }, + "/harvest-downloads/{harvestId}": { + "get": { + "operationId": "download_data_harvest_archive", + "summary": "Download data harvest archive", + "tags": ["Users"], + "responses": { + "204": {"description": "No Content"}, + "400": { + "description": "Bad Request - The request was malformed or contained invalid data", + "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}} + }, + "429": { + "description": "Too Many Requests - You are being rate limited", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "code": {"type": "string", "enum": ["RATE_LIMITED"]}, + "message": {"type": "string"}, + "retry_after": {"type": "number", "description": "Seconds to wait before retrying"}, + "global": {"type": "boolean", "description": "Whether this is a global rate limit"} + }, + "required": ["code", "message", "retry_after"] + } + } + }, + "headers": { + "Retry-After": { + "description": "Number of seconds to wait before retrying (only on 429)", + "schema": {"type": "integer"} + }, + "X-RateLimit-Limit": { + "description": "The number of requests that can be made in the current window", + "schema": {"type": "integer"} + }, + "X-RateLimit-Remaining": { + "description": "The number of remaining requests that can be made", + "schema": {"type": "integer"} + }, + "X-RateLimit-Reset": { + "description": "Unix timestamp when the rate limit resets", + "schema": {"type": "integer"} + } + } + }, + "500": { + "description": "Internal Server Error - An unexpected error occurred", + "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}} + } + }, + "x-mint": {"metadata": {"title": "Download data harvest archive"}}, + "description": "Streams a completed data harvest archive. Authorised by a signed, expiring token rather than a session, so the link works from the harvest completion email. Only active when presigned harvest downloads are disabled.", + "parameters": [ + { + "name": "harvestId", + "in": "path", + "required": true, + "schema": {"type": "string"}, + "description": "The harvestId" + } + ] + } + }, "/invites/{invite_code}": { "get": { "operationId": "get_invite", @@ -35895,7 +35959,7 @@ "HarvestDownloadUrlResponse": { "type": "object", "properties": { - "download_url": {"type": "string", "description": "The presigned URL to download the harvest archive"}, + "download_url": {"type": "string", "description": "The temporary URL to download the harvest archive"}, "expires_at": {"type": "string", "description": "ISO 8601 timestamp when the harvest download expires"} }, "required": ["download_url", "expires_at"] diff --git a/fluxer_api/src/api/rate_limit_configs/UserRateLimitConfig.ts b/fluxer_api/src/api/rate_limit_configs/UserRateLimitConfig.ts index a7c872d88..2fc15ed1c 100644 --- a/fluxer_api/src/api/rate_limit_configs/UserRateLimitConfig.ts +++ b/fluxer_api/src/api/rate_limit_configs/UserRateLimitConfig.ts @@ -216,6 +216,10 @@ export const UserRateLimitConfigs = { bucket: 'user:harvest:download', config: {limit: 10, windowMs: ms('1 minute')}, } as RouteRateLimitConfig, + USER_HARVEST_DOWNLOAD_FILE: { + bucket: 'user:harvest:download_file', + config: {limit: 60, windowMs: ms('1 minute')}, + } as RouteRateLimitConfig, USER_ENTRANCE_SOUND_LIST: { bucket: 'user:entrance_sound:list', config: {limit: 30, windowMs: ms('1 minute')}, diff --git a/fluxer_api/src/api/user/controllers/UserContentController.ts b/fluxer_api/src/api/user/controllers/UserContentController.ts index c624fa473..85d3228d9 100644 --- a/fluxer_api/src/api/user/controllers/UserContentController.ts +++ b/fluxer_api/src/api/user/controllers/UserContentController.ts @@ -1,5 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +import {Readable} from 'node:stream'; +import {HARVEST_DOWNLOAD_PATH} from '@app/api/user/services/HarvestDownloadUrl'; import {HarvestIdParam, MessageIdParam} from '@fluxer/schema/src/domains/common/CommonParamSchemas'; import {MessageListResponse} from '@fluxer/schema/src/domains/message/MessageResponseSchemas'; import { @@ -269,6 +271,50 @@ export function UserContentController(app: HonoApp) { return ctx.json(harvest, 200); }, ); + app.get( + `${HARVEST_DOWNLOAD_PATH}/:harvestId`, + RateLimitMiddleware(RateLimitConfigs.USER_HARVEST_DOWNLOAD_FILE), + Validator('param', HarvestIdParam), + OpenAPI({ + operationId: 'download_data_harvest_archive', + summary: 'Download data harvest archive', + responseSchema: null, + statusCode: 200, + security: [], + tags: ['Users'], + description: + 'Streams a completed data harvest archive. Authorised by a signed, expiring token rather than a session, so the link works from the harvest completion email. Only active when presigned harvest downloads are disabled.', + }), + async (ctx) => { + const {harvestId} = ctx.req.valid('param'); + const token = ctx.req.query('token'); + if (!token) { + return ctx.text('Not Found', 404); + } + const result = await ctx.get('userContentRequestService').streamHarvestDownload({ + harvestId, + token, + range: ctx.req.header('range') ?? undefined, + storageService: ctx.get('storageService'), + }); + if (!result) { + return ctx.text('Not Found', 404); + } + const headers = new Headers(); + headers.set('Content-Type', result.contentType ?? 'application/zip'); + headers.set('Content-Disposition', `attachment; filename="${encodeURIComponent(result.filename)}"`); + headers.set('Content-Length', String(result.contentLength)); + headers.set('Cache-Control', 'private, no-store'); + headers.set('Accept-Ranges', 'bytes'); + if (result.contentRange) { + headers.set('Content-Range', result.contentRange); + } + return new Response(Readable.toWeb(result.body) as ReadableStream, { + status: result.contentRange ? 206 : 200, + headers, + }); + }, + ); app.get( '/users/@me/harvest/:harvestId/download', RateLimitMiddleware(RateLimitConfigs.USER_HARVEST_DOWNLOAD), diff --git a/fluxer_api/src/api/user/services/HarvestDownloadToken.ts b/fluxer_api/src/api/user/services/HarvestDownloadToken.ts new file mode 100644 index 000000000..f30a8c1b4 --- /dev/null +++ b/fluxer_api/src/api/user/services/HarvestDownloadToken.ts @@ -0,0 +1,71 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {createHmac, timingSafeEqual} from 'node:crypto'; +import {isJsonRecord, parseJsonWithGuard} from '../../utils/JsonBoundaryUtils'; + +const KEY_DERIVATION_LABEL = 'fluxer.harvest-download.v1'; + +interface HarvestDownloadTokenPayload { + userId: string; + harvestId: string; + storageKey: string; + expiresAt: number; +} + +function deriveKey(secret: string): Buffer { + return createHmac('sha256', secret).update(KEY_DERIVATION_LABEL).digest(); +} + +function computeSignature(payloadBase64: string, secret: string): Buffer { + return createHmac('sha256', deriveKey(secret)).update(payloadBase64).digest(); +} + +function isHarvestDownloadTokenPayload(value: unknown): value is HarvestDownloadTokenPayload { + if (!isJsonRecord(value)) return false; + return ( + typeof value['userId'] === 'string' && + typeof value['harvestId'] === 'string' && + typeof value['storageKey'] === 'string' && + typeof value['expiresAt'] === 'number' && + Number.isFinite(value['expiresAt']) + ); +} + +export function signHarvestDownloadToken(payload: HarvestDownloadTokenPayload, secret: string): string { + const payloadBase64 = Buffer.from(JSON.stringify(payload)).toString('base64url'); + const signature = computeSignature(payloadBase64, secret).toString('base64url'); + return `${payloadBase64}.${signature}`; +} + +export function verifyHarvestDownloadToken(token: string, secret: string): HarvestDownloadTokenPayload | null { + const dotIndex = token.indexOf('.'); + if (dotIndex === -1) { + return null; + } + const payloadBase64 = token.slice(0, dotIndex); + const signatureBase64 = token.slice(dotIndex + 1); + const expectedSignature = computeSignature(payloadBase64, secret); + let providedSignature: Buffer; + try { + providedSignature = Buffer.from(signatureBase64, 'base64url'); + } catch { + return null; + } + if (expectedSignature.length !== providedSignature.length) { + return null; + } + if (!timingSafeEqual(expectedSignature, providedSignature)) { + return null; + } + let payload: HarvestDownloadTokenPayload | null; + try { + const payloadJson = Buffer.from(payloadBase64, 'base64url').toString('utf-8'); + payload = parseJsonWithGuard(payloadJson, isHarvestDownloadTokenPayload); + } catch { + return null; + } + if (!payload || Date.now() > payload.expiresAt) { + return null; + } + return payload; +} diff --git a/fluxer_api/src/api/user/services/HarvestDownloadUrl.ts b/fluxer_api/src/api/user/services/HarvestDownloadUrl.ts new file mode 100644 index 000000000..6949610f1 --- /dev/null +++ b/fluxer_api/src/api/user/services/HarvestDownloadUrl.ts @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {Config} from '../../Config'; +import type {IStorageService} from '../../infrastructure/IStorageService'; +import {signHarvestDownloadToken} from './HarvestDownloadToken'; + +export const HARVEST_DOWNLOAD_PATH = '/harvest-downloads'; + +export async function buildHarvestDownloadUrl(params: { + userId: bigint; + harvestId: bigint; + storageKey: string; + expiresInSeconds: number; + storageService: IStorageService; +}): Promise { + const {userId, harvestId, storageKey, expiresInSeconds, storageService} = params; + if (Config.presignedHarvestDownloadsEnabled) { + return storageService.getPresignedDownloadURL({ + bucket: Config.s3.buckets.harvests, + key: storageKey, + expiresIn: expiresInSeconds, + }); + } + const token = signHarvestDownloadToken( + { + userId: userId.toString(), + harvestId: harvestId.toString(), + storageKey, + expiresAt: Date.now() + expiresInSeconds * 1000, + }, + Config.auth.connectionInitiationSecret, + ); + const base = Config.endpoints.apiPublic.replace(/\/+$/u, ''); + return `${base}${HARVEST_DOWNLOAD_PATH}/${harvestId}?token=${encodeURIComponent(token)}`; +} diff --git a/fluxer_api/src/api/user/services/UserContentRequestService.ts b/fluxer_api/src/api/user/services/UserContentRequestService.ts index ee3b9fb06..2facd1a17 100644 --- a/fluxer_api/src/api/user/services/UserContentRequestService.ts +++ b/fluxer_api/src/api/user/services/UserContentRequestService.ts @@ -145,6 +145,15 @@ export class UserContentRequestService { return this.userContentService.getHarvestDownloadUrl(params.userId, params.harvestId, params.storageService); } + async streamHarvestDownload(params: { + harvestId: bigint; + token: string; + range?: string; + storageService: IStorageService; + }) { + return this.userContentService.streamHarvestDownload(params); + } + private async mapSavedMessageEntry(userId: UserID, entry: SavedMessageEntry): Promise { return { id: entry.messageId.toString(), diff --git a/fluxer_api/src/api/user/services/UserContentService.ts b/fluxer_api/src/api/user/services/UserContentService.ts index 1555e705a..bc90919c6 100644 --- a/fluxer_api/src/api/user/services/UserContentService.ts +++ b/fluxer_api/src/api/user/services/UserContentService.ts @@ -1,6 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later import crypto from 'node:crypto'; +import type {Readable} from 'node:stream'; import {MAX_BOOKMARKS_NON_PREMIUM} from '@fluxer/constants/src/LimitConstants'; import {UnknownChannelError} from '@fluxer/errors/src/domains/channel/UnknownChannelError'; import {UnknownMessageError} from '@fluxer/errors/src/domains/channel/UnknownMessageError'; @@ -24,7 +25,7 @@ import {snowflakeToDate} from '@fluxer/snowflake/src/Snowflake'; import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService'; import {ms} from 'itty-time'; import type {ApiContext} from '../../ApiContext'; -import {type ChannelID, createChannelID, type MessageID, type UserID} from '../../BrandedTypes'; +import {type ChannelID, createChannelID, createUserID, type MessageID, type UserID} from '../../BrandedTypes'; import {Config} from '../../Config'; import type {IChannelRepository} from '../../channel/IChannelRepository'; import type {ChannelService} from '../../channel/services/ChannelService'; @@ -48,6 +49,8 @@ import type {IUserContentRepository} from '../repositories/IUserContentRepositor import {UserHarvest, type UserHarvestResponse} from '../UserHarvestModel'; import {UserHarvestRepository} from '../UserHarvestRepository'; import {BaseUserUpdatePropagator} from './BaseUserUpdatePropagator'; +import {verifyHarvestDownloadToken} from './HarvestDownloadToken'; +import {buildHarvestDownloadUrl} from './HarvestDownloadUrl'; export interface SavedMessageEntry { channelId: ChannelID; @@ -472,10 +475,12 @@ export class UserContentService { throw new HarvestExpiredError(); } const ZIP_EXPIRY_MS = ms('7 days'); - const downloadUrl = await storageService.getPresignedDownloadURL({ - bucket: Config.s3.buckets.harvests, - key: harvest.storageKey, - expiresIn: ZIP_EXPIRY_MS / 1000, + const downloadUrl = await buildHarvestDownloadUrl({ + userId, + harvestId, + storageKey: harvest.storageKey, + expiresInSeconds: ZIP_EXPIRY_MS / 1000, + storageService, }); const expiresAt = new Date(Date.now() + ZIP_EXPIRY_MS); return { @@ -484,6 +489,61 @@ export class UserContentService { }; } + async streamHarvestDownload(params: { + harvestId: bigint; + token: string; + range?: string; + storageService: IStorageService; + }): Promise<{ + body: Readable; + contentLength: number; + contentRange?: string | null; + contentType?: string | null; + filename: string; + } | null> { + if (Config.presignedHarvestDownloadsEnabled) { + return null; + } + const payload = verifyHarvestDownloadToken(params.token, Config.auth.connectionInitiationSecret); + if (!payload || payload.harvestId !== params.harvestId.toString()) { + Logger.debug({harvestId: params.harvestId.toString()}, 'Harvest download rejected: invalid or expired token'); + return null; + } + let userId: UserID; + try { + userId = createUserID(BigInt(payload.userId)); + } catch { + return null; + } + const harvestRepository = new UserHarvestRepository(); + const harvest = await harvestRepository.findByUserAndHarvestId(userId, params.harvestId); + if (!harvest || !harvest.completedAt || !harvest.storageKey || harvest.failedAt) { + return null; + } + if (harvest.downloadUrlExpiresAt && harvest.downloadUrlExpiresAt < new Date()) { + return null; + } + if (harvest.storageKey !== payload.storageKey) { + Logger.debug({harvestId: params.harvestId.toString()}, 'Harvest download rejected: storage key mismatch'); + return null; + } + const object = await params.storageService.streamObject({ + bucket: Config.s3.buckets.harvests, + key: harvest.storageKey, + range: params.range, + }); + if (!object) { + return null; + } + return { + body: object.body, + contentLength: object.contentLength, + contentRange: object.contentRange, + contentType: object.contentType ?? 'application/zip', + filename: `fluxer-data-${params.harvestId}.zip`, + }; + } + async requestBulkMessageDeletion(params: {userId: UserID; delayMs?: number}): Promise { const {userId, delayMs = ms('1 day')} = params; const scheduledAt = new Date(Date.now() + delayMs); diff --git a/fluxer_api/src/api/user/services/__tests__/HarvestDownloadToken.test.ts b/fluxer_api/src/api/user/services/__tests__/HarvestDownloadToken.test.ts new file mode 100644 index 000000000..a4e190221 --- /dev/null +++ b/fluxer_api/src/api/user/services/__tests__/HarvestDownloadToken.test.ts @@ -0,0 +1,68 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {describe, expect, test} from 'vitest'; +import {signHarvestDownloadToken, verifyHarvestDownloadToken} from '../HarvestDownloadToken'; + +const SECRET = 'test-connection-secret'; + +function payload(overrides: Partial[0]> = {}) { + return { + userId: '123', + harvestId: '456', + storageKey: 'exports/123/456/user-data.zip', + expiresAt: Date.now() + 60_000, + ...overrides, + }; +} + +describe('HarvestDownloadToken', () => { + test('round trips a valid token', () => { + const original = payload(); + const verified = verifyHarvestDownloadToken(signHarvestDownloadToken(original, SECRET), SECRET); + expect(verified).toEqual(original); + }); + + test('rejects a token signed with a different secret', () => { + const token = signHarvestDownloadToken(payload(), SECRET); + expect(verifyHarvestDownloadToken(token, 'other-secret')).toBeNull(); + }); + + test('rejects an expired token', () => { + const token = signHarvestDownloadToken(payload({expiresAt: Date.now() - 1}), SECRET); + expect(verifyHarvestDownloadToken(token, SECRET)).toBeNull(); + }); + + test('rejects a tampered payload', () => { + const token = signHarvestDownloadToken(payload(), SECRET); + const [encoded, signature] = token.split('.'); + const decoded = JSON.parse(Buffer.from(encoded as string, 'base64url').toString('utf-8')); + decoded.userId = '999'; + const forged = `${Buffer.from(JSON.stringify(decoded)).toString('base64url')}.${signature}`; + expect(verifyHarvestDownloadToken(forged, SECRET)).toBeNull(); + }); + + test('rejects a tampered signature', () => { + const token = signHarvestDownloadToken(payload(), SECRET); + expect(verifyHarvestDownloadToken(`${token.slice(0, -2)}xy`, SECRET)).toBeNull(); + }); + + test('rejects malformed input', () => { + for (const bad of ['', '.', 'nodot', 'a.b', '..', 'a.']) { + expect(verifyHarvestDownloadToken(bad, SECRET)).toBeNull(); + } + }); + + test('rejects a payload missing required fields', () => { + const encoded = Buffer.from(JSON.stringify({userId: '1'})).toString('base64url'); + const token = signHarvestDownloadToken(payload(), SECRET); + const signature = token.split('.')[1]; + expect(verifyHarvestDownloadToken(`${encoded}.${signature}`, SECRET)).toBeNull(); + }); + + test('is not verifiable as a raw HMAC of the base secret', () => { + const {createHmac} = require('node:crypto') as typeof import('node:crypto'); + const encoded = Buffer.from(JSON.stringify(payload())).toString('base64url'); + const rawSignature = createHmac('sha256', SECRET).update(encoded).digest('base64url'); + expect(verifyHarvestDownloadToken(`${encoded}.${rawSignature}`, SECRET)).toBeNull(); + }); +}); diff --git a/fluxer_api/src/api/user/tests/HarvestDownloadStreaming.test.ts b/fluxer_api/src/api/user/tests/HarvestDownloadStreaming.test.ts new file mode 100644 index 000000000..40e624554 --- /dev/null +++ b/fluxer_api/src/api/user/tests/HarvestDownloadStreaming.test.ts @@ -0,0 +1,168 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {beforeEach, describe, expect, test} from 'vitest'; +import {createTestAccount} from '../../auth/tests/AuthTestUtils'; +import {getConfig} from '../../Config'; +import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness'; +import {signHarvestDownloadToken} from '../services/HarvestDownloadToken'; +import {fetchHarvestDownload, markHarvestCompleted, requestHarvest} from './HarvestTestUtils'; + +const ZIP_BYTES = new TextEncoder().encode('PK pretend zip'); + +async function withStreamingHarvestDownloads(callback: () => Promise): Promise { + const config = getConfig(); + const original = config.presignedHarvestDownloadsEnabled; + config.presignedHarvestDownloadsEnabled = false; + try { + await callback(); + } finally { + config.presignedHarvestDownloadsEnabled = original; + } +} + +async function completedHarvest(harness: ApiTestHarness) { + const account = await createTestAccount(harness); + const {harvest_id} = await requestHarvest(harness, account.token); + await markHarvestCompleted(account.userId, harvest_id, new Date(Date.now() + 6 * 24 * 60 * 60 * 1000)); + const storageKey = `test/${harvest_id}.zip`; + await harness.storageService.uploadObject({ + bucket: getConfig().s3.buckets.harvests, + key: storageKey, + body: ZIP_BYTES, + contentType: 'application/zip', + }); + return {account, harvestId: harvest_id, storageKey}; +} + +function downloadPath(url: string): string { + return url.slice(url.indexOf('/harvest-downloads')); +} + +async function get(harness: ApiTestHarness, path: string): Promise { + return harness.app.request(path, {headers: {'x-forwarded-for': '203.0.113.7'}}); +} + +describe('Harvest download streaming', () => { + let harness: ApiTestHarness; + beforeEach(async () => { + harness = await createApiTestHarness(); + }); + + test('defaults to presigned URLs and leaves the streaming route closed', async () => { + const {account, harvestId, storageKey} = await completedHarvest(harness); + expect(getConfig().presignedHarvestDownloadsEnabled).toBe(true); + + const download = await fetchHarvestDownload(harness, account.token, harvestId); + expect(download.download_url).toBe('https://presigned.url/test'); + + const token = signHarvestDownloadToken( + {userId: account.userId, harvestId, storageKey, expiresAt: Date.now() + 60_000}, + getConfig().auth.connectionInitiationSecret, + ); + const response = await get(harness, `/harvest-downloads/${harvestId}?token=${encodeURIComponent(token)}`); + expect(response.status).toBe(404); + }); + + test('serves the archive through the API when presigned downloads are disabled', async () => { + await withStreamingHarvestDownloads(async () => { + const {account, harvestId} = await completedHarvest(harness); + const download = await fetchHarvestDownload(harness, account.token, harvestId); + + expect(download.download_url).not.toContain('presigned.url'); + expect(download.download_url).toContain(`/harvest-downloads/${harvestId}`); + expect(download.download_url).toContain('token='); + + const response = await get(harness, downloadPath(download.download_url)); + expect(response.status).toBe(200); + expect(response.headers.get('content-type')).toBe('application/zip'); + expect(response.headers.get('content-disposition')).toContain('attachment'); + expect(response.headers.get('cache-control')).toBe('private, no-store'); + expect(new Uint8Array(await response.arrayBuffer())).toEqual(ZIP_BYTES); + }); + }); + + test('rejects a tampered token', async () => { + await withStreamingHarvestDownloads(async () => { + const {account, harvestId} = await completedHarvest(harness); + const download = await fetchHarvestDownload(harness, account.token, harvestId); + const path = downloadPath(download.download_url); + const response = await get(harness, `${path.slice(0, -2)}xy`); + expect(response.status).toBe(404); + }); + }); + + test('rejects a token signed with the wrong secret', async () => { + await withStreamingHarvestDownloads(async () => { + const {account, harvestId, storageKey} = await completedHarvest(harness); + const token = signHarvestDownloadToken( + {userId: account.userId, harvestId, storageKey, expiresAt: Date.now() + 60_000}, + 'not-the-configured-secret', + ); + const response = await get(harness, `/harvest-downloads/${harvestId}?token=${encodeURIComponent(token)}`); + expect(response.status).toBe(404); + }); + }); + + test('rejects an expired token even while the harvest is still valid', async () => { + await withStreamingHarvestDownloads(async () => { + const {account, harvestId, storageKey} = await completedHarvest(harness); + const token = signHarvestDownloadToken( + {userId: account.userId, harvestId, storageKey, expiresAt: Date.now() - 1}, + getConfig().auth.connectionInitiationSecret, + ); + const response = await get(harness, `/harvest-downloads/${harvestId}?token=${encodeURIComponent(token)}`); + expect(response.status).toBe(404); + }); + }); + + test('rejects a token replayed against a different harvest id', async () => { + await withStreamingHarvestDownloads(async () => { + const first = await completedHarvest(harness); + const second = await completedHarvest(harness); + const token = signHarvestDownloadToken( + { + userId: first.account.userId, + harvestId: first.harvestId, + storageKey: first.storageKey, + expiresAt: Date.now() + 60_000, + }, + getConfig().auth.connectionInitiationSecret, + ); + const response = await get(harness, `/harvest-downloads/${second.harvestId}?token=${encodeURIComponent(token)}`); + expect(response.status).toBe(404); + }); + }); + + test('rejects a token whose storage key no longer matches the harvest', async () => { + await withStreamingHarvestDownloads(async () => { + const {account, harvestId} = await completedHarvest(harness); + const token = signHarvestDownloadToken( + {userId: account.userId, harvestId, storageKey: 'test/some-other-key.zip', expiresAt: Date.now() + 60_000}, + getConfig().auth.connectionInitiationSecret, + ); + const response = await get(harness, `/harvest-downloads/${harvestId}?token=${encodeURIComponent(token)}`); + expect(response.status).toBe(404); + }); + }); + + test('rejects a token minted for a different user', async () => { + await withStreamingHarvestDownloads(async () => { + const {harvestId, storageKey} = await completedHarvest(harness); + const other = await createTestAccount(harness); + const token = signHarvestDownloadToken( + {userId: other.userId, harvestId, storageKey, expiresAt: Date.now() + 60_000}, + getConfig().auth.connectionInitiationSecret, + ); + const response = await get(harness, `/harvest-downloads/${harvestId}?token=${encodeURIComponent(token)}`); + expect(response.status).toBe(404); + }); + }); + + test('requires a token', async () => { + await withStreamingHarvestDownloads(async () => { + const {harvestId} = await completedHarvest(harness); + const response = await get(harness, `/harvest-downloads/${harvestId}`); + expect(response.status).toBe(404); + }); + }); +}); diff --git a/fluxer_api/src/api/worker/tasks/HarvestUserData.ts b/fluxer_api/src/api/worker/tasks/HarvestUserData.ts index 0fe25a717..a1ee673ae 100644 --- a/fluxer_api/src/api/worker/tasks/HarvestUserData.ts +++ b/fluxer_api/src/api/worker/tasks/HarvestUserData.ts @@ -50,6 +50,7 @@ import type {User} from '../../models/User'; import type {UserGuildSettings} from '../../models/UserGuildSettings'; import type {UserSettings} from '../../models/UserSettings'; import type {WebAuthnCredential} from '../../models/WebAuthnCredential'; +import {buildHarvestDownloadUrl} from '../../user/services/HarvestDownloadUrl'; import {resolveSessionClientInfo} from '../../utils/SessionClientIdentity'; import {createArchiveJsonBuffer} from '../utils/ArchiveJson'; import {appendAssetToArchive, buildHashedAssetKey, getAnimatedAssetExtension} from '../utils/AssetArchiveHelpers'; @@ -685,10 +686,12 @@ async function createAndUploadArchive(params: ArchiveParams): Promise { describe('React hook order stability', () => { it('keeps guard returns outside hook-bearing component bodies', () => { expect(findHookOrderHazards()).toEqual([]); - }, 10_000); + }, 60_000); }); diff --git a/fluxer_media_proxy/src/config.rs b/fluxer_media_proxy/src/config.rs index 0467ac3a7..86b01cc33 100644 --- a/fluxer_media_proxy/src/config.rs +++ b/fluxer_media_proxy/src/config.rs @@ -17,6 +17,13 @@ pub enum DeploymentMode { Upload, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BucketStyle { + Path, + VirtualHosted, + Rooted, +} + #[derive(Clone, Debug)] pub struct Config { pub node_env: String, @@ -33,6 +40,10 @@ pub struct Config { pub s3_secret_access_key: String, pub s3_session_token: String, pub s3_force_path_style: bool, + pub s3_read_endpoint: Option, + pub s3_read_bucket: String, + pub s3_read_bucket_style: BucketStyle, + pub s3_read_signed: bool, pub bucket_cdn: String, pub bucket_uploads: String, pub bucket_static: String, @@ -97,6 +108,34 @@ impl Config { env.get("FLUXER_MEDIA_PROXY_UPLOAD_RELAY_SECRET_BASE64"), mode, )?; + + let s3_force_path_style = parse_bool( + "FLUXER_S3_FORCE_PATH_STYLE", + env.get("FLUXER_S3_FORCE_PATH_STYLE"), + )? + .unwrap_or(true); + let bucket_cdn = env + .get("FLUXER_S3_BUCKET_CDN") + .map(ToOwned::to_owned) + .unwrap_or_else(|| "cdn".to_owned()); + let s3_read_endpoint = non_empty(env.get("FLUXER_S3_READ_ENDPOINT")); + if let Some(endpoint) = s3_read_endpoint.as_deref() { + validate_read_endpoint(endpoint)?; + } + let s3_read_bucket = + non_empty(env.get("FLUXER_S3_READ_BUCKET")).unwrap_or_else(|| bucket_cdn.clone()); + let s3_read_bucket_style = parse_bucket_style(env.get("FLUXER_S3_READ_BUCKET_STYLE"))? + .unwrap_or(if s3_force_path_style { + BucketStyle::Path + } else { + BucketStyle::VirtualHosted + }); + let s3_read_signed = parse_bool( + "FLUXER_S3_READ_SIGNED", + non_empty(env.get("FLUXER_S3_READ_SIGNED")).as_deref(), + )? + .unwrap_or(false); + Ok(Self { node_env: env.get("NODE_ENV").unwrap_or("development").to_owned(), bind_host: env @@ -138,15 +177,12 @@ impl Config { .map(ToOwned::to_owned) .unwrap_or_default(), s3_session_token: env.get("FLUXER_S3_SESSION_TOKEN").unwrap_or("").to_owned(), - s3_force_path_style: parse_bool( - "FLUXER_S3_FORCE_PATH_STYLE", - env.get("FLUXER_S3_FORCE_PATH_STYLE"), - )? - .unwrap_or(true), - bucket_cdn: env - .get("FLUXER_S3_BUCKET_CDN") - .map(ToOwned::to_owned) - .unwrap_or_else(|| "cdn".to_owned()), + s3_force_path_style, + s3_read_endpoint, + s3_read_bucket, + s3_read_bucket_style, + s3_read_signed, + bucket_cdn, bucket_uploads: env .get("FLUXER_S3_BUCKET_UPLOADS") .map(ToOwned::to_owned) @@ -347,6 +383,48 @@ fn parse_mode_env(raw: Option<&str>) -> anyhow::Result> { }) } +fn non_empty(raw: Option<&str>) -> Option { + raw.map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) +} + +fn parse_bucket_style(raw: Option<&str>) -> anyhow::Result> { + let Some(raw) = non_empty(raw) else { + return Ok(None); + }; + match raw.to_ascii_lowercase().as_str() { + "path" => Ok(Some(BucketStyle::Path)), + "virtual" => Ok(Some(BucketStyle::VirtualHosted)), + "root" => Ok(Some(BucketStyle::Rooted)), + _ => Err(anyhow::anyhow!( + "FLUXER_S3_READ_BUCKET_STYLE must be one of: path, virtual, root" + )), + } +} + +fn validate_read_endpoint(endpoint: &str) -> anyhow::Result<()> { + let parsed = url::Url::parse(endpoint) + .map_err(|err| anyhow::anyhow!("FLUXER_S3_READ_ENDPOINT is not a valid URL: {err}"))?; + anyhow::ensure!( + matches!(parsed.scheme(), "http" | "https"), + "FLUXER_S3_READ_ENDPOINT must be an http or https URL" + ); + anyhow::ensure!( + parsed.host_str().is_some_and(|host| !host.is_empty()), + "FLUXER_S3_READ_ENDPOINT must include a host" + ); + anyhow::ensure!( + parsed.username().is_empty() && parsed.password().is_none(), + "FLUXER_S3_READ_ENDPOINT must not contain credentials" + ); + anyhow::ensure!( + parsed.query().is_none() && parsed.fragment().is_none(), + "FLUXER_S3_READ_ENDPOINT must not contain a query string or fragment" + ); + Ok(()) +} + fn parse_storage_backend(raw: Option<&str>) -> anyhow::Result> { let Some(raw) = raw else { return Ok(None); @@ -631,4 +709,154 @@ mod tests { .contains("FLUXER_MEDIA_PROXY_TRANSFORM_TIMEOUT_MS") ); } + + fn env_with(extra: &[(&'static str, &'static str)]) -> Vec<(&'static str, &'static str)> { + let mut env = base_env(); + env.extend_from_slice(extra); + env + } + + #[test] + fn read_endpoint_defaults_to_disabled() { + let cfg = Config::load_from_iter(base_env()).unwrap(); + assert_eq!(None, cfg.s3_read_endpoint); + assert_eq!("cdn", cfg.s3_read_bucket); + assert_eq!(BucketStyle::Path, cfg.s3_read_bucket_style); + assert!(!cfg.s3_read_signed); + } + + #[test] + fn empty_read_vars_are_treated_as_unset() { + let cfg = Config::load_from_iter(env_with(&[ + ("FLUXER_S3_READ_ENDPOINT", " "), + ("FLUXER_S3_READ_BUCKET", ""), + ("FLUXER_S3_READ_BUCKET_STYLE", ""), + ("FLUXER_S3_READ_SIGNED", ""), + ])) + .unwrap(); + assert_eq!(None, cfg.s3_read_endpoint); + assert_eq!("cdn", cfg.s3_read_bucket); + assert_eq!(BucketStyle::Path, cfg.s3_read_bucket_style); + assert!(!cfg.s3_read_signed); + } + + #[test] + fn every_new_var_tolerates_a_blank_value_individually() { + for var in [ + "FLUXER_S3_READ_ENDPOINT", + "FLUXER_S3_READ_BUCKET", + "FLUXER_S3_READ_BUCKET_STYLE", + "FLUXER_S3_READ_SIGNED", + ] { + for blank in ["", " "] { + let cfg = Config::load_from_iter(env_with(&[(var, blank)])).unwrap_or_else(|err| { + panic!("{var}={blank:?} must be treated as unset: {err}") + }); + assert_eq!(None, cfg.s3_read_endpoint); + assert_eq!("cdn", cfg.s3_read_bucket); + assert_eq!(BucketStyle::Path, cfg.s3_read_bucket_style); + assert!(!cfg.s3_read_signed); + } + } + } + + #[test] + fn read_bucket_defaults_to_cdn_bucket_and_can_be_overridden() { + let cfg = Config::load_from_iter(env_with(&[("FLUXER_S3_BUCKET_CDN", "fluxer")])).unwrap(); + assert_eq!("fluxer", cfg.s3_read_bucket); + + let cfg = Config::load_from_iter(env_with(&[ + ("FLUXER_S3_BUCKET_CDN", "fluxer"), + ("FLUXER_S3_READ_BUCKET", "fluxer-static"), + ])) + .unwrap(); + assert_eq!("fluxer-static", cfg.s3_read_bucket); + } + + #[test] + fn read_bucket_style_inherits_force_path_style() { + let cfg = + Config::load_from_iter(env_with(&[("FLUXER_S3_FORCE_PATH_STYLE", "false")])).unwrap(); + assert_eq!(BucketStyle::VirtualHosted, cfg.s3_read_bucket_style); + + let cfg = Config::load_from_iter(env_with(&[ + ("FLUXER_S3_FORCE_PATH_STYLE", "false"), + ("FLUXER_S3_READ_BUCKET_STYLE", "root"), + ])) + .unwrap(); + assert_eq!(BucketStyle::Rooted, cfg.s3_read_bucket_style); + assert!(!cfg.s3_force_path_style); + } + + #[test] + fn read_bucket_style_parses_all_values_case_insensitively() { + for (raw, expected) in [ + ("path", BucketStyle::Path), + ("PATH", BucketStyle::Path), + ("virtual", BucketStyle::VirtualHosted), + (" Virtual ", BucketStyle::VirtualHosted), + ("root", BucketStyle::Rooted), + ("ROOT", BucketStyle::Rooted), + ] { + let cfg = + Config::load_from_iter(env_with(&[("FLUXER_S3_READ_BUCKET_STYLE", raw)])).unwrap(); + assert_eq!(expected, cfg.s3_read_bucket_style, "raw={raw}"); + } + } + + #[test] + fn read_bucket_style_rejects_unknown_value() { + let err = Config::load_from_iter(env_with(&[("FLUXER_S3_READ_BUCKET_STYLE", "cdn")])) + .unwrap_err(); + assert!(err.to_string().contains("FLUXER_S3_READ_BUCKET_STYLE")); + } + + #[test] + fn read_signed_parses_boolean() { + let cfg = Config::load_from_iter(env_with(&[("FLUXER_S3_READ_SIGNED", "true")])).unwrap(); + assert!(cfg.s3_read_signed); + + let err = + Config::load_from_iter(env_with(&[("FLUXER_S3_READ_SIGNED", "maybe")])).unwrap_err(); + assert!(err.to_string().contains("FLUXER_S3_READ_SIGNED")); + } + + #[test] + fn read_endpoint_is_validated_at_startup() { + let cfg = Config::load_from_iter(env_with(&[( + "FLUXER_S3_READ_ENDPOINT", + "https://cdn.example.net", + )])) + .unwrap(); + assert_eq!( + Some("https://cdn.example.net".to_owned()), + cfg.s3_read_endpoint + ); + + for (bad, expected) in [ + ("cdn.example.net", "not a valid URL"), + ("ftp://cdn.example.net", "must be an http or https URL"), + ("https://", "not a valid URL"), + ( + "https://user:pw@cdn.example.net", + "must not contain credentials", + ), + ( + "https://cdn.example.net/?token=abc", + "must not contain a query string or fragment", + ), + ( + "https://cdn.example.net/#frag", + "must not contain a query string or fragment", + ), + ] { + let err = + Config::load_from_iter(env_with(&[("FLUXER_S3_READ_ENDPOINT", bad)])).unwrap_err(); + let message = err.to_string(); + assert!( + message.contains("FLUXER_S3_READ_ENDPOINT") && message.contains(expected), + "bad={bad} err={message}" + ); + } + } } diff --git a/fluxer_media_proxy/src/server.rs b/fluxer_media_proxy/src/server.rs index 0a79c23db..567887f8b 100644 --- a/fluxer_media_proxy/src/server.rs +++ b/fluxer_media_proxy/src/server.rs @@ -101,6 +101,15 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> { } else { None }; + if let Some(read_endpoint) = state.cfg.s3_read_endpoint.as_deref() { + info!( + endpoint = read_endpoint, + bucket = state.cfg.s3_read_bucket, + style = ?state.cfg.s3_read_bucket_style, + signed = state.cfg.s3_read_signed, + "object body reads served from the S3 read endpoint" + ); + } let mut router = Router::new() .route("/_health", get(health)) .route("/_metrics", get(metrics_handler)) diff --git a/fluxer_media_proxy/src/storage.rs b/fluxer_media_proxy/src/storage.rs index 03c649303..11d57f8a1 100644 --- a/fluxer_media_proxy/src/storage.rs +++ b/fluxer_media_proxy/src/storage.rs @@ -2,7 +2,7 @@ use crate::{ aws_sigv4, - config::{Config, StorageBackend}, + config::{BucketStyle, Config, StorageBackend}, constants, http_client, mime, }; use axum::body::Body; @@ -370,23 +370,51 @@ impl Store { fn s3_url(&self, bucket: &str, key: &str) -> Result { safe_bucket(bucket)?; safe_key(key)?; - if self.cfg.s3_endpoint.is_empty() { - return Err(StorageError::InvalidS3Endpoint); + object_url( + &self.cfg.s3_endpoint, + write_bucket_style(&self.cfg), + bucket, + key, + ) + } + + fn read_endpoint_for(&self, bucket: &str) -> Option<&str> { + self.cfg + .s3_read_endpoint + .as_deref() + .filter(|_| bucket == self.cfg.s3_read_bucket) + } + + fn s3_read_url(&self, bucket: &str, key: &str) -> Result { + let Some(endpoint) = self.read_endpoint_for(bucket) else { + return self.s3_url(bucket, key); + }; + safe_bucket(bucket)?; + safe_key(key)?; + object_url(endpoint, self.cfg.s3_read_bucket_style, bucket, key) + } + + fn read_status_is_miss(&self, bucket: &str, status: reqwest::StatusCode) -> bool { + if status == reqwest::StatusCode::NOT_FOUND { + return true; } - let endpoint = self.cfg.s3_endpoint.trim_end_matches('/'); - let encoded_key = percent_encode(key.as_bytes(), PATH_ENCODE_SET).to_string(); - if self.cfg.s3_force_path_style { - return Ok(format!("{endpoint}/{bucket}/{encoded_key}")); + status == reqwest::StatusCode::FORBIDDEN + && self.read_endpoint_for(bucket).is_some() + && !self.cfg.s3_read_signed + } + + fn read_headers( + &self, + bucket: &str, + method: Method, + url: &str, + extra_signed_headers: &[aws_sigv4::Header<'_>], + ) -> Result { + if self.read_endpoint_for(bucket).is_some() && !self.cfg.s3_read_signed { + return Ok(reqwest::header::HeaderMap::new()); } - validate_virtual_hosted_bucket(bucket)?; - let parsed = url::Url::parse(endpoint).map_err(|_| StorageError::InvalidS3Endpoint)?; - let scheme = parsed.scheme(); - let host = parsed.host_str().ok_or(StorageError::InvalidS3Endpoint)?; - let port = parsed.port().map(|p| format!(":{p}")).unwrap_or_default(); - let base_path = parsed.path().trim_end_matches('/'); - Ok(format!( - "{scheme}://{bucket}.{host}{port}{base_path}/{encoded_key}" - )) + let signed = self.sign(method, url, &[], None, extra_signed_headers)?; + Ok(signed_headers(&signed, &self.cfg)) } fn s3_bucket_url(&self, bucket: &str) -> Result { @@ -408,15 +436,10 @@ impl Store { } async fn read_s3(&self, bucket: &str, key: &str) -> Result { - let url = self.s3_url(bucket, key)?; - let signed = self.sign(Method::GET, &url, &[], None, &[])?; - let response = self - .client - .get(&url) - .headers(signed_headers(&signed, &self.cfg)) - .send() - .await?; - if response.status() == reqwest::StatusCode::NOT_FOUND { + let url = self.s3_read_url(bucket, key)?; + let headers = self.read_headers(bucket, Method::GET, &url, &[])?; + let response = self.client.get(&url).headers(headers).send().await?; + if self.read_status_is_miss(bucket, response.status()) { return Err(StorageError::NotFound); } if !response.status().is_success() { @@ -484,14 +507,13 @@ impl Store { key: &str, range_header: Option<&str>, ) -> Result { - let url = self.s3_url(bucket, key)?; + let url = self.s3_read_url(bucket, key)?; let range_extra = range_header.map(|value| aws_sigv4::Header { name: "Range", value, }); let extra = range_extra.as_slice(); - let signed = self.sign(Method::GET, &url, &[], None, extra)?; - let mut headers = signed_headers(&signed, &self.cfg); + let mut headers = self.read_headers(bucket, Method::GET, &url, extra)?; if let Some(range_value) = range_header { headers.insert( header::RANGE, @@ -501,7 +523,7 @@ impl Store { ); } let response = self.client.get(&url).headers(headers).send().await?; - if response.status() == reqwest::StatusCode::NOT_FOUND { + if self.read_status_is_miss(bucket, response.status()) { return Err(StorageError::NotFound); } if !response.status().is_success() { @@ -750,6 +772,42 @@ impl http_body::Body for ChannelBody { } } +fn write_bucket_style(cfg: &Config) -> BucketStyle { + if cfg.s3_force_path_style { + BucketStyle::Path + } else { + BucketStyle::VirtualHosted + } +} + +fn object_url( + endpoint: &str, + style: BucketStyle, + bucket: &str, + key: &str, +) -> Result { + if endpoint.is_empty() { + return Err(StorageError::InvalidS3Endpoint); + } + let endpoint = endpoint.trim_end_matches('/'); + let encoded_key = percent_encode(key.as_bytes(), PATH_ENCODE_SET).to_string(); + match style { + BucketStyle::Path => Ok(format!("{endpoint}/{bucket}/{encoded_key}")), + BucketStyle::Rooted => Ok(format!("{endpoint}/{encoded_key}")), + BucketStyle::VirtualHosted => { + validate_virtual_hosted_bucket(bucket)?; + let parsed = url::Url::parse(endpoint).map_err(|_| StorageError::InvalidS3Endpoint)?; + let scheme = parsed.scheme(); + let host = parsed.host_str().ok_or(StorageError::InvalidS3Endpoint)?; + let port = parsed.port().map(|p| format!(":{p}")).unwrap_or_default(); + let base_path = parsed.path().trim_end_matches('/'); + Ok(format!( + "{scheme}://{bucket}.{host}{port}{base_path}/{encoded_key}" + )) + } + } +} + fn signed_headers(signed: &aws_sigv4::SignedRequest, cfg: &Config) -> reqwest::header::HeaderMap { let mut headers = reqwest::header::HeaderMap::new(); headers.insert( @@ -858,7 +916,7 @@ fn reject_symlink_chain(path: &Path) -> Result<(), StorageError> { #[cfg(test)] mod tests { use super::*; - use crate::config::{DeploymentMode, StorageBackend}; + use crate::config::{BucketStyle, DeploymentMode, StorageBackend}; fn test_config(root: &Path) -> Config { Config { @@ -876,6 +934,10 @@ mod tests { s3_secret_access_key: String::new(), s3_session_token: String::new(), s3_force_path_style: true, + s3_read_endpoint: None, + s3_read_bucket: "cdn".to_owned(), + s3_read_bucket_style: BucketStyle::Path, + s3_read_signed: false, bucket_cdn: "cdn".to_owned(), bucket_uploads: "uploads".to_owned(), bucket_static: "static".to_owned(), @@ -1060,4 +1122,446 @@ mod tests { .await; assert!(result.is_err()); } + + type CapturedRequests = + std::sync::Arc>>; + + async fn capture_server() -> (String, CapturedRequests) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let captured: CapturedRequests = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())); + let handler = std::sync::Arc::clone(&captured); + let app = axum::Router::new().fallback(axum::routing::any( + move |request: axum::extract::Request| { + let captured = std::sync::Arc::clone(&handler); + async move { + let (parts, _body) = request.into_parts(); + captured + .lock() + .await + .push((parts.method, parts.uri, parts.headers)); + ([(header::CONTENT_TYPE, "image/png")], "payload") + } + }, + )); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{addr}"), captured) + } + + async fn status_server(status: u16) -> (String, CapturedRequests) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let captured: CapturedRequests = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())); + let handler = std::sync::Arc::clone(&captured); + let app = axum::Router::new().fallback(axum::routing::any( + move |request: axum::extract::Request| { + let captured = std::sync::Arc::clone(&handler); + async move { + let (parts, _body) = request.into_parts(); + captured + .lock() + .await + .push((parts.method, parts.uri, parts.headers)); + (StatusCode::from_u16(status).unwrap(), "denied") + } + }, + )); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{addr}"), captured) + } + + fn s3_test_config(root: &Path, s3_endpoint: &str) -> Config { + let mut cfg = test_config(root); + cfg.storage_backend = StorageBackend::S3; + cfg.s3_endpoint = s3_endpoint.to_owned(); + cfg.s3_access_key_id = "AKIAIOSFODNN7EXAMPLE".to_owned(); + cfg.s3_secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_owned(); + cfg + } + + async fn only_request( + captured: &CapturedRequests, + ) -> (http::Method, http::Uri, http::HeaderMap) { + let mut guard = captured.lock().await; + assert_eq!(1, guard.len(), "expected exactly one captured request"); + guard.remove(0) + } + + fn assert_unsigned(headers: &http::HeaderMap) { + assert!(headers.get(header::AUTHORIZATION).is_none()); + assert!(headers.get("x-amz-date").is_none()); + assert!(headers.get("x-amz-content-sha256").is_none()); + } + + fn assert_signed(headers: &http::HeaderMap) { + assert!( + headers + .get(header::AUTHORIZATION) + .unwrap() + .to_str() + .unwrap() + .starts_with("AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/") + ); + assert!(headers.get("x-amz-date").is_some()); + assert!(headers.get("x-amz-content-sha256").is_some()); + } + + #[tokio::test] + async fn read_without_read_endpoint_hits_s3_endpoint_signed() { + let (s3, s3_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let store = Store::new(s3_test_config(tmp.path(), &s3)); + + let object = store + .read_object("cdn", "attachments/1/2/a.png") + .await + .unwrap(); + + assert_eq!("payload", object.data); + let (method, uri, headers) = only_request(&s3_seen).await; + assert_eq!(http::Method::GET, method); + assert_eq!("/cdn/attachments/1/2/a.png", uri.path()); + assert_signed(&headers); + } + + #[tokio::test] + async fn read_endpoint_routes_fronted_bucket_to_cdn_unsigned() { + let (s3, s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + let store = Store::new(cfg); + + let object = store + .read_object("cdn", "attachments/1/2/a.png") + .await + .unwrap(); + + assert_eq!("payload", object.data); + assert!( + s3_seen.lock().await.is_empty(), + "S3 endpoint must not be touched" + ); + let (method, uri, headers) = only_request(&cdn_seen).await; + assert_eq!(http::Method::GET, method); + assert_eq!("/cdn/attachments/1/2/a.png", uri.path()); + assert_unsigned(&headers); + } + + #[tokio::test] + async fn read_endpoint_rooted_style_omits_bucket_segment() { + let (s3, _s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + store + .read_object("cdn", "attachments/1/2/a.png") + .await + .unwrap(); + + let (_, uri, _) = only_request(&cdn_seen).await; + assert_eq!("/attachments/1/2/a.png", uri.path()); + } + + #[test] + fn read_endpoint_virtual_style_uses_bucket_subdomain() { + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), "https://s3.example.test"); + cfg.s3_read_endpoint = Some("https://cdn.example.net".to_owned()); + cfg.s3_read_bucket_style = BucketStyle::VirtualHosted; + let store = Store::new(cfg); + + assert_eq!( + "https://cdn.cdn.example.net/attachments/1/2/a.png", + store.s3_read_url("cdn", "attachments/1/2/a.png").unwrap() + ); + } + + #[tokio::test] + async fn read_endpoint_never_redirects_other_buckets() { + let (s3, s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + store + .read_object("uploads", "fresh-upload-key") + .await + .unwrap(); + + assert!( + cdn_seen.lock().await.is_empty(), + "uploads must not hit the CDN" + ); + let (_, uri, headers) = only_request(&s3_seen).await; + assert_eq!("/uploads/fresh-upload-key", uri.path()); + assert_signed(&headers); + } + + #[tokio::test] + async fn read_endpoint_never_affects_writes() { + let (s3, s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_bucket = "uploads".to_owned(); + let store = Store::new(cfg); + + let (tx, rx) = tokio::sync::mpsc::channel(2); + tokio::spawn(async move { + tx.send(Ok(Bytes::from_static(b"body"))).await.unwrap(); + }); + store + .relay_put_object( + "uploads", + "guild/x.bin", + RelayPutOptions { + body: RelayBody::Streamed(rx), + content_length: 4, + content_type: Some("application/octet-stream".to_owned()), + upload_id: None, + part_number: None, + timeout_ms: 5_000, + }, + ) + .await + .unwrap(); + + assert!( + cdn_seen.lock().await.is_empty(), + "writes must not hit the CDN" + ); + let (method, uri, headers) = only_request(&s3_seen).await; + assert_eq!(http::Method::PUT, method); + assert_eq!("/uploads/guild/x.bin", uri.path()); + assert_signed(&headers); + } + + #[tokio::test] + async fn read_endpoint_signs_when_read_signed_enabled() { + let (s3, _s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let cdn_host = cdn.trim_start_matches("http://").to_owned(); + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_signed = true; + let store = Store::new(cfg); + + store.read_object("cdn", "a.png").await.unwrap(); + + let (_, _, headers) = only_request(&cdn_seen).await; + assert_signed(&headers); + assert_eq!( + cdn_host, + headers.get(header::HOST).unwrap().to_str().unwrap() + ); + } + + #[tokio::test] + async fn unsigned_read_still_sends_range_header() { + let (s3, _s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + let store = Store::new(cfg); + + store + .stream_object("cdn", "video.mp4", Some("bytes=10-19")) + .await + .unwrap(); + + let (_, _, headers) = only_request(&cdn_seen).await; + assert_eq!("bytes=10-19", headers.get(header::RANGE).unwrap()); + assert_unsigned(&headers); + } + + #[tokio::test] + async fn head_object_always_uses_origin_for_authoritative_length() { + let (s3, s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + store.head_object("cdn", "a.png").await.unwrap(); + + assert!( + cdn_seen.lock().await.is_empty(), + "HEAD must not hit the CDN" + ); + let (method, uri, headers) = only_request(&s3_seen).await; + assert_eq!(http::Method::HEAD, method); + assert_eq!("/cdn/a.png", uri.path()); + assert_signed(&headers); + } + + #[tokio::test] + async fn body_reads_use_cdn_while_head_uses_origin() { + let (s3, s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + store.head_object("cdn", "video.mp4").await.unwrap(); + store + .stream_object("cdn", "video.mp4", Some("bytes=0-3")) + .await + .unwrap(); + + let s3_reqs = s3_seen.lock().await.clone(); + let cdn_reqs = cdn_seen.lock().await.clone(); + assert_eq!(1, s3_reqs.len()); + assert_eq!(http::Method::HEAD, s3_reqs[0].0); + assert_eq!(1, cdn_reqs.len()); + assert_eq!(http::Method::GET, cdn_reqs[0].0); + assert_eq!("bytes=0-3", cdn_reqs[0].2.get(header::RANGE).unwrap()); + } + + #[tokio::test] + async fn unsigned_read_works_without_credentials() { + let (s3, _s3_seen) = capture_server().await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_access_key_id = String::new(); + cfg.s3_secret_access_key = String::new(); + let store = Store::new(cfg); + + store.read_object("cdn", "a.png").await.unwrap(); + + let (_, _, headers) = only_request(&cdn_seen).await; + assert_unsigned(&headers); + } + + #[test] + fn read_url_matches_write_url_shape_for_encoding_and_trailing_slash() { + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), "https://s3.example.test/"); + cfg.s3_read_endpoint = Some("https://cdn.example.net/".to_owned()); + let store = Store::new(cfg); + let key = "attachments/1/2/na me+ΓΌ.png"; + + assert_eq!( + "https://s3.example.test/uploads/attachments/1/2/na%20me%2B%C3%BC.png", + store.s3_url("uploads", key).unwrap() + ); + assert_eq!( + "https://cdn.example.net/cdn/attachments/1/2/na%20me%2B%C3%BC.png", + store.s3_read_url("cdn", key).unwrap() + ); + } + + #[test] + fn read_url_rejects_unsafe_keys_and_buckets() { + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), "https://s3.example.test"); + cfg.s3_read_endpoint = Some("https://cdn.example.net".to_owned()); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + assert!(store.s3_read_url("cdn", "../escape").is_err()); + assert!(store.s3_read_url("cdn", "/leading").is_err()); + } + + #[test] + fn read_url_validates_bucket_inside_the_read_endpoint_branch() { + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), "https://s3.example.test"); + cfg.s3_read_endpoint = Some("https://cdn.example.net".to_owned()); + cfg.s3_read_bucket = "..".to_owned(); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + assert!(store.read_endpoint_for("..").is_some()); + assert!(store.s3_read_url("..", "a.png").is_err()); + } + + #[tokio::test] + async fn unsigned_cdn_read_treats_403_as_not_found() { + let (s3, _s3_seen) = capture_server().await; + let (cdn, cdn_seen) = status_server(403).await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_bucket_style = BucketStyle::Rooted; + let store = Store::new(cfg); + + let err = store + .read_object("cdn", "avatars/1/hash") + .await + .unwrap_err(); + assert!(matches!(err, StorageError::NotFound), "got {err:?}"); + assert_eq!(1, cdn_seen.lock().await.len()); + + let stream = store.stream_object("cdn", "avatars/1/hash", None).await; + assert!( + matches!(stream, Err(StorageError::NotFound)), + "stream_object should map 403 to NotFound too" + ); + } + + #[tokio::test] + async fn signed_cdn_read_keeps_403_as_an_error() { + let (s3, _s3_seen) = capture_server().await; + let (cdn, _cdn_seen) = status_server(403).await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + cfg.s3_read_signed = true; + let store = Store::new(cfg); + + let err = store + .read_object("cdn", "avatars/1/hash") + .await + .unwrap_err(); + assert!(!matches!(err, StorageError::NotFound), "got {err:?}"); + } + + #[tokio::test] + async fn origin_read_keeps_403_as_an_error() { + let (s3, _s3_seen) = status_server(403).await; + let tmp = tempfile::tempdir().unwrap(); + let store = Store::new(s3_test_config(tmp.path(), &s3)); + + let err = store + .read_object("cdn", "avatars/1/hash") + .await + .unwrap_err(); + assert!(!matches!(err, StorageError::NotFound), "got {err:?}"); + } + + #[tokio::test] + async fn unfronted_bucket_keeps_403_as_an_error() { + let (s3, _s3_seen) = status_server(403).await; + let (cdn, cdn_seen) = capture_server().await; + let tmp = tempfile::tempdir().unwrap(); + let mut cfg = s3_test_config(tmp.path(), &s3); + cfg.s3_read_endpoint = Some(cdn.clone()); + let store = Store::new(cfg); + + let err = store.read_object("uploads", "fresh").await.unwrap_err(); + assert!(!matches!(err, StorageError::NotFound), "got {err:?}"); + assert!(cdn_seen.lock().await.is_empty()); + } } diff --git a/packages/config/src/ConfigLoader.ts b/packages/config/src/ConfigLoader.ts index 55a212e41..6c7e5df55 100644 --- a/packages/config/src/ConfigLoader.ts +++ b/packages/config/src/ConfigLoader.ts @@ -94,6 +94,7 @@ function defaultConfig(): MasterConfig { ip_ban_exempt_ips: [], presigned_attachment_uploads_enabled: false, presigned_downloads_enabled: false, + presigned_harvest_downloads_enabled: true, unfurl_ignored_hosts: [], embeds: { oembed_html_enabled: false, diff --git a/packages/config/src/MasterConfig.ts b/packages/config/src/MasterConfig.ts index c25778de8..28c719942 100644 --- a/packages/config/src/MasterConfig.ts +++ b/packages/config/src/MasterConfig.ts @@ -93,6 +93,7 @@ export interface MasterConfig { ip_ban_exempt_ips: Array; presigned_attachment_uploads_enabled: boolean; presigned_downloads_enabled: boolean; + presigned_harvest_downloads_enabled: boolean; unfurl_ignored_hosts: Array; embeds: { oembed_html_enabled: boolean; diff --git a/packages/config/src/config_loader/EnvironmentOverrides.ts b/packages/config/src/config_loader/EnvironmentOverrides.ts index f4f477c2e..adb9175ce 100644 --- a/packages/config/src/config_loader/EnvironmentOverrides.ts +++ b/packages/config/src/config_loader/EnvironmentOverrides.ts @@ -85,6 +85,10 @@ const NAMED_FLUXER_ENV_OVERRIDES: Record = { path: ['services', 'api', 'presigned_downloads_enabled'], parse: parseEnvValue, }, + FLUXER_API_PRESIGNED_HARVEST_DOWNLOADS_ENABLED: { + path: ['services', 'api', 'presigned_harvest_downloads_enabled'], + parse: parseEnvValue, + }, FLUXER_API_WORKER_MODE: {path: ['services', 'api', 'worker', 'mode']}, FLUXER_API_WORKER_LANE: {path: ['services', 'api', 'worker', 'lane']}, FLUXER_API_WORKER_TASK: {path: ['services', 'api', 'worker', 'task']}, diff --git a/packages/schema/src/domains/user/UserHarvestSchemas.ts b/packages/schema/src/domains/user/UserHarvestSchemas.ts index d943ffcc3..705cff60e 100644 --- a/packages/schema/src/domains/user/UserHarvestSchemas.ts +++ b/packages/schema/src/domains/user/UserHarvestSchemas.ts @@ -49,7 +49,7 @@ export type HarvestStatusResponse = z.infer; export const HarvestStatusResponseSchemaNullable = HarvestStatusResponseSchema.nullable(); export const HarvestDownloadUrlResponse = z.object({ - download_url: createStringType(1, 2048).describe('The presigned URL to download the harvest archive'), + download_url: createStringType(1, 2048).describe('The temporary URL to download the harvest archive'), expires_at: z.string().describe('ISO 8601 timestamp when the harvest download expires'), });