/** * Publishing-side media access: the `loadAsset` / `openAsset` pair the * connectors need to attach an approved post's media. * * - loadAsset reads the `media_assets` row inside the current execution's * tenant (app.workspace_id set from the publishing context, RLS forced), * so a job can only ever resolve its own workspace's assets. * - openAsset downloads the sealed object and opens the SPENC1 frame with the * media KEK, checking the content address. Read-only: publishing never * writes to storage. * * Bytes travel worker → provider; they never pass through the API process. */ import { createDecipheriv, createHash } from "node:crypto"; import type { Pool } from "pg"; import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3"; import type { MediaAsset } from "@tactical-media-broadcast-hub/core"; import { currentPublishingContext } from "./execution-context.js"; const FRAME_MAGIC = Buffer.from("SPENC1"); /** Same per-asset caps the API enforces at upload (media.service.ts). */ const MAX_ASSET_BYTES = 64 * 1024 * 1024; export interface SealedObjectReader { /** Returns the stored (sealed) object body. */ read(key: string): Promise; } export interface PublishingMediaAccess { loadAsset(assetId: string): Promise; openAsset(media: MediaAsset): Promise<{ bytes: Uint8Array; filename: string }>; } function aesGcmOpen(key: Buffer, iv: Buffer, sealed: Buffer): Buffer { const decipher = createDecipheriv("aes-256-gcm", key, iv); decipher.setAuthTag(sealed.subarray(sealed.length - 16)); return Buffer.concat([decipher.update(sealed.subarray(0, sealed.length - 16)), decipher.final()]); } /** Open [magic | iv(12) | wrapped CEK(48) | body+tag], the API's storage frame. */ export function openSealedFrame(sealed: Buffer, kek: Buffer): Buffer { if (!sealed.subarray(0, FRAME_MAGIC.length).equals(FRAME_MAGIC)) throw new Error("MEDIA_FRAME_INVALID"); const iv = sealed.subarray(6, 18); const cek = aesGcmOpen(kek, iv, sealed.subarray(18, 66)); return aesGcmOpen(cek, iv, sealed.subarray(66)); } export function mediaKekFromEnv(env: Record = process.env): Buffer { const raw = env.S3_KEK; if (raw !== undefined && raw !== "") { const kek = Buffer.from(raw, "base64"); if (kek.length !== 32) throw new Error("S3_KEK must be 32 bytes base64"); return kek; } if (env.NODE_ENV === "production") throw new Error("S3_KEK is required in production (media encryption is mandatory)"); // Dev mirror of the API's fallback so a local scheduler reads local uploads. return createHash("sha256").update(`dev-media-kek:${env.HOSTNAME ?? "local"}`).digest(); } export function s3ObjectReaderFromEnv(env: Record = process.env): SealedObjectReader { const bucket = env.S3_BUCKET ?? "tactical-media-broadcast-hub-media"; const client = new S3Client({ endpoint: env.S3_ENDPOINT, region: env.S3_REGION ?? "us-east-1", credentials: env.S3_ACCESS_KEY !== undefined && env.S3_SECRET_KEY !== undefined ? { accessKeyId: env.S3_ACCESS_KEY, secretAccessKey: env.S3_SECRET_KEY } : undefined, forcePathStyle: env.S3_FORCE_PATH_STYLE !== "false", }); return { async read(key) { const response = await client.send(new GetObjectCommand({ Bucket: bucket, Key: key })); if (response.Body === undefined) throw new Error("MEDIA_OBJECT_EMPTY"); return Buffer.from(await response.Body.transformToByteArray()); }, }; } const EXTENSIONS: Record = { "image/jpeg": "jpg", "image/png": "png", "image/webp": "webp", "video/mp4": "mp4", "video/webm": "webm", }; interface MediaRow { id: string; workspace_id: string; storage_key: string; mime: string; size_bytes: string | number; created_at: Date; } /** Caller owns pool lifecycle (same contract as the credential accessor). */ export function createPublishingMediaAccess(pool: Pool, reader: SealedObjectReader, kek: Buffer): PublishingMediaAccess { async function loadAsset(assetId: string): Promise { const context = currentPublishingContext(); const client = await pool.connect(); try { await client.query("BEGIN"); await client.query("SELECT set_config('app.workspace_id', $1, true)", [context.workspaceId]); const result = await client.query( `SELECT id, workspace_id, storage_key, mime, size_bytes, created_at FROM media_assets WHERE id = $1 AND workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid`, [assetId], ); await client.query("COMMIT"); const row = result.rows[0]; if (row === undefined) throw new Error("MEDIA_ASSET_NOT_FOUND"); return { id: row.id, workspaceId: row.workspace_id, storageKey: row.storage_key, mimeType: row.mime, sizeBytes: Number(row.size_bytes), createdAt: row.created_at, }; } catch (error) { await client.query("ROLLBACK").catch(() => undefined); throw error; } finally { client.release(); } } async function openAsset(media: MediaAsset): Promise<{ bytes: Uint8Array; filename: string }> { const context = currentPublishingContext(); if (media.workspaceId !== context.workspaceId) throw new Error("MEDIA_TENANT_FENCE"); if (media.sizeBytes > MAX_ASSET_BYTES) throw new Error("MEDIA_ASSET_TOO_LARGE"); const plaintext = openSealedFrame(await reader.read(media.storageKey), kek); const expected = /^sha256\/([0-9a-f]{64})\./.exec(media.storageKey)?.[1]; if (expected === undefined || createHash("sha256").update(plaintext).digest("hex") !== expected) { throw new Error("MEDIA_INTEGRITY_FAILURE"); } return { bytes: plaintext, filename: `${media.id}.${EXTENSIONS[media.mimeType] ?? "bin"}` }; } return { loadAsset, openAsset }; }