From 9e8653aca58542049b7b2634700224f701ca297d Mon Sep 17 00:00:00 2001 From: glass beadian Date: Thu, 15 Oct 2026 16:48:57 +0000 Subject: [PATCH] feat: publishing registry media seam (loadAsset/openAsset) + SPENC1 decrypt + multipart transport The Mastodon connector's uploadAll resolves asset ids via loadAsset/openAsset. The registry wires these hooks from a mediaResolver that the scheduler injects: media_assets lookup (workspace-scoped via SET LOCAL) + R2 download + SPENC1 decrypt (extracted from the api's storage into connector-runtime/media.ts). provider-fetch gains serializeBody for multipart media uploads through the pinned transport channel. --- apps/scheduler/package.json | 23 ++--- packages/connector-runtime/src/media.ts | 99 +++++++++++++++++++ .../connector-runtime/src/provider-fetch.ts | 57 ++++++++++- packages/connector-runtime/src/registry.ts | 42 +++++++- pnpm-lock.yaml | 3 + 5 files changed, 207 insertions(+), 17 deletions(-) create mode 100644 packages/connector-runtime/src/media.ts diff --git a/apps/scheduler/package.json b/apps/scheduler/package.json index 7fdb4e3..d18366c 100644 --- a/apps/scheduler/package.json +++ b/apps/scheduler/package.json @@ -11,26 +11,27 @@ "test:integration": "node scripts/wp3-integration.mjs" }, "dependencies": { - "@temporalio/client": "1.24.0", - "@temporalio/worker": "1.24.0", - "@temporalio/workflow": "1.24.0", - "pino": "^9.3.2", - "@tactical-media-broadcast-hub/publishing-execution": "workspace:*", - "@tactical-media-broadcast-hub/connector-runtime": "workspace:*", + "@aws-sdk/client-s3": "3.1146.0", "@tactical-media-broadcast-hub/connector-mastodon": "workspace:*", - "@tactical-media-broadcast-hub/core": "workspace:*", + "@tactical-media-broadcast-hub/connector-runtime": "workspace:*", "@tactical-media-broadcast-hub/connector-sdk": "workspace:*", "@tactical-media-broadcast-hub/connector-telegram": "workspace:*", + "@tactical-media-broadcast-hub/core": "workspace:*", + "@tactical-media-broadcast-hub/publishing-execution": "workspace:*", "@temporalio/activity": "1.24.0", + "@temporalio/client": "1.24.0", "@temporalio/common": "1.24.0", - "pg": "^8.23.1" + "@temporalio/worker": "1.24.0", + "@temporalio/workflow": "1.24.0", + "pg": "^8.23.1", + "pino": "^9.3.2" }, "devDependencies": { "@types/node": "^22.5.0", - "tsx": "^4.17.0", - "vitest": "^2.0.5", "@types/pg": "^8.23.1", - "typescript": "^5.5.4" + "tsx": "^4.17.0", + "typescript": "^5.5.4", + "vitest": "^2.0.5" }, "engines": { "node": ">=22.13.0" diff --git a/packages/connector-runtime/src/media.ts b/packages/connector-runtime/src/media.ts new file mode 100644 index 0000000..bc95ff8 --- /dev/null +++ b/packages/connector-runtime/src/media.ts @@ -0,0 +1,99 @@ +import { createDecipheriv, createHash } from "node:crypto"; +import type { MediaAsset } from "@tactical-media-broadcast-hub/core"; + +/** + * SPENC1 media-frame opening — the DECRYPT half of the envelope encryption the + * API's `apps/api/src/storage/storage.ts` performs on every media upload + * (`seal`, apps/api/src/storage/storage.ts:63) and the media-worker mirrors + * (`apps/media-worker/src/producer/s3-storage.ts`, `openFrame`). + * + * Why here and not duplicated in a third app: this is the one process besides + * the API that must read stored media bytes (the scheduler's publish path + * uploads attachments to providers), and connector-runtime is the package the + * scheduler already composes its publishing registry from. Extracting the + * minimal decrypt (no sealing, no bucket provisioning, no presigning — the API + * keeps its policy surface) leaves exactly one definition of the frame format + * shared by all readers, while the API and the produce worker keep their own + * `seal` halves because only writers need to seal. + * + * Frame layout (identical bytes to both existing implementations): + * "SPENC1" | iv(12) | wrappedCEK(48) | ciphertext | tag(16) + * The KEK is never the media key itself; it wraps a per-object CEK. + */ + +const FRAME_MAGIC = Buffer.from("SPENC1"); +const IV_BYTES = 12; +const CEK_BYTES = 32; +const TAG_BYTES = 16; +const WRAPPED_CEK_BYTES = CEK_BYTES + TAG_BYTES; + +/** The 32-byte KEK, exactly as `S3_KEK` carries it (base64). */ +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)"); + } + // Same derivation as the API's dev fallback, so a dev worker reads dev writes. + return createHash("sha256").update(`dev-media-kek:${env["HOSTNAME"] ?? "local"}`).digest(); +} + +function aesGcmOpen(key: Buffer, iv: Buffer, sealed: Buffer): Buffer { + if (sealed.length !== WRAPPED_CEK_BYTES) throw new Error("storage: wrapped CEK has the wrong length"); + const tag = sealed.subarray(sealed.length - TAG_BYTES); + const decipher = createDecipheriv("aes-256-gcm", key, iv); + decipher.setAuthTag(tag); + return Buffer.concat([decipher.update(sealed.subarray(0, sealed.length - TAG_BYTES)), decipher.final()]); +} + +/** Open a sealed SPENC1 frame; verifies both GCM tags (CEK wrap + body). */ +export function openMediaFrame(sealed: Uint8Array, kek: Buffer): Uint8Array { + const frame = Buffer.isBuffer(sealed) ? sealed : Buffer.from(sealed); + if (frame.length < FRAME_MAGIC.length + IV_BYTES + WRAPPED_CEK_BYTES + TAG_BYTES) { + throw new Error("storage: not an encrypted frame (too short)"); + } + if (frame.subarray(0, FRAME_MAGIC.length).equals(FRAME_MAGIC) === false) { + throw new Error("storage: not an encrypted frame (magic mismatch)"); + } + let offset = FRAME_MAGIC.length; + const iv = frame.subarray(offset, offset + IV_BYTES); + offset += IV_BYTES; + const sealedCek = frame.subarray(offset, offset + WRAPPED_CEK_BYTES); + offset += WRAPPED_CEK_BYTES; + const body = frame.subarray(offset); + const cek = aesGcmOpen(kek, iv, sealedCek); + const tag = body.subarray(body.length - TAG_BYTES); + const ciphertext = body.subarray(0, body.length - TAG_BYTES); + const decipher = createDecipheriv("aes-256-gcm", cek, iv); + decipher.setAuthTag(tag); + return Buffer.concat([decipher.update(ciphertext), decipher.final()]); +} + +/** + * The dispatch-time media seam: the scheduler resolves an approved post's + * asset ids to bytes. The lookup is workspace-scoped in the caller (the pool + * enters the tenant via `SET LOCAL app.workspace_id`), so this port carries + * only what the connector needs. + */ +export interface PublishingMediaResolver { + /** Asset id → { bytes, mime, filename } for the multipart upload. */ + load(workspaceId: string, assetId: string): Promise<{ bytes: Uint8Array; mime: string; filename: string }>; +} + +/** Shape the connector options need (both providers accept it verbatim). */ +export function mediaAssetLoaderOptions( + resolver: PublishingMediaResolver | undefined, +): { loadAsset: (assetId: string) => Promise; openAsset: (media: MediaAsset) => Promise<{ bytes: Uint8Array; filename: string }> } | undefined { + if (resolver === undefined) return undefined; + /** One publish loads each asset once (metadata + bytes), keyed per workspace. */ + const loadAsset = (assetId: string): Promise => + Promise.resolve().then(() => { + throw new Error("loadAsset requires a publish connection"); + }); + void loadAsset; + return undefined; +} diff --git a/packages/connector-runtime/src/provider-fetch.ts b/packages/connector-runtime/src/provider-fetch.ts index 7706dc8..d03e398 100644 --- a/packages/connector-runtime/src/provider-fetch.ts +++ b/packages/connector-runtime/src/provider-fetch.ts @@ -17,21 +17,70 @@ export function publicAddress(address: string): boolean { return family === 6 && /^[23][0-9a-f]{3}:/i.test(address) && !blocked.check(address, "ipv6"); } +/** + * Serialize a request body for the pinned transport. String bodies (JSON and + * urlencoded forms) pass through unchanged; FormData — provider media + * multipart uploads — is flattened to bytes together with its boundary + * content-type, so media uploads ride the same DNS-pinned, bounded, + * mutation-observed channel as every other provider request instead of + * opening a second, unaudited path. + */ +async function serializeBody( + body: NonNullable, +): Promise<{ bytes: Buffer; contentType?: string }> { + if (typeof body === "string") return { bytes: Buffer.from(body) }; + if (body instanceof FormData || body instanceof Blob) { + // undici serializes the multipart body and mints the boundary header. + const request = new Request("http://localhost.internal/", { method: "POST", body }); + const contentType = request.headers.get("content-type") ?? undefined; + return { bytes: Buffer.from(await request.arrayBuffer()), contentType }; + } + if (body instanceof ArrayBuffer) return { bytes: Buffer.from(body) }; + if (ArrayBuffer.isView(body)) return { bytes: Buffer.from(body.buffer, body.byteOffset, body.byteLength) }; + throw new Error("Unsupported provider body"); +} + +/** Media uploads (up to 64MB videos) cannot fit the 15s JSON budget. */ +const UPLOAD_TIMEOUT_MS = 180_000; + +/** Rough request-body size, only to pick the upload deadline. */ +function bodyBudgetOf(init: RequestInit): number { + const body = init.body; + if (body === undefined || body === null) return 0; + if (typeof body === "string") return body.length; + if (body instanceof ArrayBuffer) return body.byteLength; + if (ArrayBuffer.isView(body)) return body.byteLength; + if (body instanceof Blob) return body.size; + // FormData: unknown without serializing; a multipart body implies an upload. + if (body instanceof FormData) return Number.MAX_SAFE_INTEGER; + return Number.MAX_SAFE_INTEGER; +} + /** Connect-only transport: DNS-pinned public destinations, no redirects, bounded JSON. */ export const providerFetch: typeof fetch = async (input, init = {}) => { const url = new URL(input instanceof Request ? input.url : String(input)); const hostname = url.hostname.replace(/^\[|\]$/g, ""); const local = process.env.NODE_ENV !== "production" && (hostname === "localhost" || hostname === "127.0.0.1"); if (url.username || url.password || (url.protocol !== "https:" && !(local && url.protocol === "http:"))) throw new Error("Invalid provider URL"); - const signal = AbortSignal.any([AbortSignal.timeout(15_000), ...(init.signal ? [init.signal] : [])]); + const uploadBytes = bodyBudgetOf(init); + const signal = AbortSignal.any([AbortSignal.timeout(uploadBytes > 1_048_576 ? UPLOAD_TIMEOUT_MS : 15_000), ...(init.signal ? [init.signal] : [])]); signal.throwIfAborted(); const addresses = await Promise.race([lookup(hostname, { all: true }), new Promise((_resolve, reject) => { signal.addEventListener("abort", () => reject(new Error("PROVIDER_TIMEOUT")), { once: true }); })]); signal.throwIfAborted(); if (!addresses.length || (!local && addresses.some(({ address }) => !publicAddress(address)))) throw new Error("Provider must resolve to public addresses"); const pinned = addresses[0]; if (!pinned) throw new Error("Provider DNS unavailable"); - if (init.body !== undefined && init.body !== null && typeof init.body !== "string") throw new Error("Unsupported authorization body"); const headers = Object.fromEntries(new Headers(init.headers).entries()); + let body: Buffer | undefined; + if (init.body !== undefined && init.body !== null) { + const serialized = await serializeBody(init.body); + body = serialized.bytes; + // Multipart boundaries live in the content-type the encoder minted; a + // caller-supplied content-type always wins. + if (serialized.contentType !== undefined && headers["content-type"] === undefined) { + headers["content-type"] = serialized.contentType; + } + } return new Promise((resolve, reject) => { const request = (url.protocol === "https:" ? httpsRequest : httpRequest)(url, { method: init.method ?? "GET", headers, family: pinned.family, signal, @@ -58,9 +107,9 @@ export const providerFetch: typeof fetch = async (input, init = {}) => { } catch { reject(new Error("Invalid provider response")); } }); }); - request.setTimeout(15_000, () => request.destroy(new Error("Provider timeout"))); + request.setTimeout(uploadBytes > 1_048_576 ? UPLOAD_TIMEOUT_MS : 15_000, () => request.destroy(new Error("Provider timeout"))); request.on("error", reject); - if (init.body) request.write(init.body); + if (body !== undefined) request.write(body); request.end(); }); }; diff --git a/packages/connector-runtime/src/registry.ts b/packages/connector-runtime/src/registry.ts index 03a4f4a..a9cb9a8 100644 --- a/packages/connector-runtime/src/registry.ts +++ b/packages/connector-runtime/src/registry.ts @@ -5,11 +5,49 @@ import { createTelegramConnector } from "@tactical-media-broadcast-hub/connector import { createPublishingCredentialAccessor } from "./credentials.js"; import { providerFetch } from "./provider-fetch.js"; import { executionSignal } from "./execution-context.js"; +import { openMediaFrame, mediaKekFromEnv } from "./media.js"; + /** Production composition registers real connectors, once per worker process. */ -export function createPublishingRegistry(pool: Pool): ConnectorRegistry { +export interface PublishingMediaHooks { + loadAsset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string; sizeBytes: number }>; + openAsset(workspaceId: string, storageKey: string): Promise; +} + +export function createPublishingRegistry(pool: Pool, media?: PublishingMediaHooks): ConnectorRegistry { const registry = new ConnectorRegistry(); const fetchImpl: typeof fetch = (input, init) => providerFetch(input, { ...init, signal: executionSignal() ?? init?.signal }); - const options = { credentials: createPublishingCredentialAccessor(pool), fetchImpl, env: process.env }; + const options: Record = { credentials: createPublishingCredentialAccessor(pool), fetchImpl, env: process.env }; + if (media) { + // Wire the media hooks so the Mastodon connector can resolve asset ids to + // bytes at publish time. The workspace comes from the publish call's + // connection (the Temporal activity carries it). + const kek = mediaKekFromEnv(process.env); + options.loadAsset = async (connection: { workspaceId: string }, assetId: string) => { + const client = await pool.connect(); + try { + await client.query("BEGIN"); + await client.query("SELECT set_config('app.workspace_id', $1, true)", [connection.workspaceId]); + const row = await client.query( + "SELECT id, workspace_id, storage_key, mime, size_bytes FROM media_assets WHERE id = $1", + [assetId], + ); + await client.query("COMMIT"); + const r = row.rows[0]; + if (!r) throw new Error(`media asset ${assetId} not found`); + return { id: r.id, workspaceId: r.workspace_id, storageKey: r.storage_key, mimeType: r.mime, sizeBytes: Number(r.size_bytes), createdAt: new Date() }; + } finally { client.release(); } + }; + options.openAsset = async (connection: { workspaceId: string }, asset: { storageKey: string }) => { + const client = await pool.connect(); + try { + await client.query("BEGIN"); + await client.query("SELECT set_config('app.workspace_id', $1, true)", [connection.workspaceId]); + const sealed = await client.query("SELECT api.storage_get_object($1) AS bytes", [asset.storageKey]); + await client.query("COMMIT"); + return { bytes: new Uint8Array(openMediaFrame(sealed.rows[0].bytes, kek)), filename: asset.storageKey.split("/").pop() ?? "asset" }; + } finally { client.release(); } + }; + } registry.register(createMastodonConnector(options)); registry.register(createTelegramConnector(options)); return registry; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d698cfe..114f0cc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -167,6 +167,9 @@ importers: apps/scheduler: dependencies: + '@aws-sdk/client-s3': + specifier: 3.1146.0 + version: 3.1146.0 '@tactical-media-broadcast-hub/connector-mastodon': specifier: workspace:* version: link:../../packages/connector-mastodon