diff --git a/packages/connector-runtime/package.json b/packages/connector-runtime/package.json index 65e9f23..b2af4d9 100644 --- a/packages/connector-runtime/package.json +++ b/packages/connector-runtime/package.json @@ -19,7 +19,9 @@ "@tactical-media-broadcast-hub/core": "workspace:*", "pg": "^8.23.1", "@tactical-media-broadcast-hub/connector-mastodon": "workspace:*", - "@tactical-media-broadcast-hub/connector-telegram": "workspace:*" + "@tactical-media-broadcast-hub/connector-telegram": "workspace:*", + "@tactical-media-broadcast-hub/connector-instagram": "workspace:*", + "@aws-sdk/client-s3": "^3.1146.0" }, "devDependencies": { "@types/node": "^22.20.5", diff --git a/packages/connector-runtime/src/execution-context.ts b/packages/connector-runtime/src/execution-context.ts index f3a6952..b6dacf1 100644 --- a/packages/connector-runtime/src/execution-context.ts +++ b/packages/connector-runtime/src/execution-context.ts @@ -39,3 +39,10 @@ export function publishingContext(connection: ConnectorConnection): PublishingEx return context; } export function executionSignal(): AbortSignal | undefined { return executionStorage.getStore()?.signal; } +/** The running execution's context; media access has no connection to fence on, only the tenant. */ +export function currentPublishingContext(): PublishingExecutionContext { + const context = executionStorage.getStore(); + if (!context) throw new Error("EXECUTION_CONTEXT_MISSING"); + context.signal?.throwIfAborted(); + return context; +} diff --git a/packages/connector-runtime/src/idempotency.ts b/packages/connector-runtime/src/idempotency.ts index 914cfde..297925e 100644 --- a/packages/connector-runtime/src/idempotency.ts +++ b/packages/connector-runtime/src/idempotency.ts @@ -37,6 +37,22 @@ export interface DispatchPreparation { * drop author-supplied execution keys, inject the attempt's idempotency key * and strip scheduledAt unless provider scheduling is explicitly allowed. */ +/** + * The approved snapshot carries alt texts positionally (altTexts[i] describes + * mediaAssetIds[i]); connectors read them as providerMetadata.altText keyed by + * asset id. Pipeline-owned, so it replaces any author-set map. + */ +function altTextByAsset(post: NormalizedPost): { altText?: Record } { + const altTexts = (post as NormalizedPost & { altTexts?: unknown }).altTexts; + if (!Array.isArray(altTexts) || altTexts.length !== post.mediaAssetIds.length) return {}; + const map: Record = {}; + post.mediaAssetIds.forEach((id, index) => { + const text = altTexts[index]; + if (typeof text === "string" && text.trim() !== "") map[id] = text; + }); + return Object.keys(map).length === 0 ? {} : { altText: map }; +} + export function preparePostForDispatch(post: NormalizedPost, prep: DispatchPreparation): NormalizedPost { for (const key of FORBIDDEN_POST_METADATA_KEYS) { if (Object.prototype.hasOwnProperty.call(post.providerMetadata, key)) throw new ExecutionMetadataError(key); @@ -45,7 +61,7 @@ export function preparePostForDispatch(post: NormalizedPost, prep: DispatchPrepa const prepared: NormalizedPost = { ...stripped, mediaAssetIds: [...post.mediaAssetIds], - providerMetadata: { ...stripped.providerMetadata, idempotencyKey: prep.idempotencyKey }, + providerMetadata: { ...stripped.providerMetadata, ...altTextByAsset(post), idempotencyKey: prep.idempotencyKey }, }; if (prep.allowProviderScheduling !== true) delete prepared.scheduledAt; return prepared; diff --git a/packages/connector-runtime/src/index.ts b/packages/connector-runtime/src/index.ts index 9b2ffd1..e5cf766 100644 --- a/packages/connector-runtime/src/index.ts +++ b/packages/connector-runtime/src/index.ts @@ -7,3 +7,4 @@ export * from "./vault.js"; export * from "./provider-fetch.js"; export * from "./credentials.js"; export * from "./registry.js"; +export * from "./media.js"; diff --git a/packages/connector-runtime/src/media.test.ts b/packages/connector-runtime/src/media.test.ts new file mode 100644 index 0000000..ccf2cb7 --- /dev/null +++ b/packages/connector-runtime/src/media.test.ts @@ -0,0 +1,86 @@ +import { createCipheriv, createHash, randomBytes } from "node:crypto"; +import type { Pool } from "pg"; +import { describe, expect, it } from "vitest"; +import type { MediaAsset, NormalizedPost } from "@tactical-media-broadcast-hub/core"; +import { runWithPublishingContext } from "./execution-context.js"; +import { preparePostForDispatch } from "./idempotency.js"; +import { createPublishingMediaAccess, openSealedFrame } from "./media.js"; + +const KEK = randomBytes(32); +const WS = "0d9a1b40-6f0f-4b1e-8f0e-3f3c9e6b4a10"; +const context = { workspaceId: WS, connectionId: "conn-1", credentialId: "cred-1", credentialGeneration: "1" }; + +/** The API's frame: magic | iv | AES-GCM(kek, cek) | AES-GCM(cek, body). */ +function seal(plaintext: Buffer, kek = KEK): Buffer { + const cek = randomBytes(32); + const iv = randomBytes(12); + const wrap = createCipheriv("aes-256-gcm", kek, iv); + const sealedCek = Buffer.concat([wrap.update(cek), wrap.final(), wrap.getAuthTag()]); + const cipher = createCipheriv("aes-256-gcm", cek, iv); + const body = Buffer.concat([cipher.update(plaintext), cipher.final(), cipher.getAuthTag()]); + return Buffer.concat([Buffer.from("SPENC1"), iv, sealedCek, body]); +} + +const JPEG = Buffer.from([0xff, 0xd8, 0xff, 0xe0, 1, 2, 3, 4]); +const KEY = `sha256/${createHash("sha256").update(JPEG).digest("hex")}.jpg`; +const asset: MediaAsset = { id: "11111111-2222-4333-8444-555555555555", workspaceId: WS, storageKey: KEY, mimeType: "image/jpeg", sizeBytes: JPEG.length, createdAt: new Date() }; + +function fakePool(rows: unknown[]): { pool: Pool; queries: Array<{ sql: string; values?: unknown[] }> } { + const queries: Array<{ sql: string; values?: unknown[] }> = []; + const client = { + query: async (sql: string, values?: unknown[]) => { queries.push({ sql, values }); return { rows: sql.startsWith("SELECT id") ? rows : [] }; }, + release: () => undefined, + }; + return { pool: { connect: async () => client } as unknown as Pool, queries }; +} + +describe("openSealedFrame", () => { + it("opens the API frame and rejects tampering or a wrong key", () => { + const sealed = seal(JPEG); + expect(openSealedFrame(sealed, KEK).equals(JPEG)).toBe(true); + const flipped = Buffer.from(sealed); flipped[flipped.length - 20]! ^= 1; + expect(() => openSealedFrame(flipped, KEK)).toThrow(); + expect(() => openSealedFrame(sealed, randomBytes(32))).toThrow(); + expect(() => openSealedFrame(Buffer.from("plain"), KEK)).toThrow(/MEDIA_FRAME_INVALID/); + }); +}); + +describe("createPublishingMediaAccess", () => { + it("loads the asset inside the execution's tenant", async () => { + const { pool, queries } = fakePool([{ id: asset.id, workspace_id: WS, storage_key: KEY, mime: "image/jpeg", size_bytes: "8", created_at: asset.createdAt }]); + const media = createPublishingMediaAccess(pool, { read: async () => seal(JPEG) }, KEK); + const loaded = await runWithPublishingContext(context, () => media.loadAsset(asset.id)); + expect(loaded).toMatchObject({ id: asset.id, workspaceId: WS, storageKey: KEY, mimeType: "image/jpeg", sizeBytes: 8 }); + expect(queries[1]).toEqual({ sql: "SELECT set_config('app.workspace_id', $1, true)", values: [WS] }); + expect(queries[2]?.sql).toContain("workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid"); + }); + + it("refuses outside a publishing execution and for unknown ids", async () => { + const { pool } = fakePool([]); + const media = createPublishingMediaAccess(pool, { read: async () => seal(JPEG) }, KEK); + await expect(media.loadAsset(asset.id)).rejects.toThrow(/EXECUTION_CONTEXT_MISSING/); + await expect(runWithPublishingContext(context, () => media.loadAsset(asset.id))).rejects.toThrow(/MEDIA_ASSET_NOT_FOUND/); + }); + + it("opens bytes, checks the content address and the tenant", async () => { + const media = createPublishingMediaAccess(fakePool([]).pool, { read: async () => seal(JPEG) }, KEK); + const opened = await runWithPublishingContext(context, () => media.openAsset(asset)); + expect(Buffer.from(opened.bytes).equals(JPEG)).toBe(true); + expect(opened.filename).toBe(`${asset.id}.jpg`); + await expect(runWithPublishingContext(context, () => media.openAsset({ ...asset, workspaceId: "other" }))).rejects.toThrow(/MEDIA_TENANT_FENCE/); + const swapped = createPublishingMediaAccess(fakePool([]).pool, { read: async () => seal(Buffer.from("other bytes")) }, KEK); + await expect(runWithPublishingContext(context, () => swapped.openAsset(asset))).rejects.toThrow(/MEDIA_INTEGRITY_FAILURE/); + }); +}); + +describe("alt text at dispatch", () => { + it("maps positional altTexts onto providerMetadata.altText by asset id", () => { + const post: NormalizedPost & { altTexts?: Array } = { text: "x", mediaAssetIds: ["a", "b", "c"], altTexts: ["first", null, " "], visibility: "public", providerMetadata: { altText: { a: "author-set", b: "author-set" } } }; + const prepared = preparePostForDispatch(post, { idempotencyKey: "k" }); + expect(prepared.providerMetadata.altText).toEqual({ a: "first" }); + expect(prepared.providerMetadata.idempotencyKey).toBe("k"); + const withoutAltTexts: NormalizedPost = { text: "x", mediaAssetIds: ["a"], visibility: "public", providerMetadata: {} }; + const none = preparePostForDispatch(withoutAltTexts, { idempotencyKey: "k" }); + expect(none.providerMetadata.altText).toBeUndefined(); + }); +}); diff --git a/packages/connector-runtime/src/media.ts b/packages/connector-runtime/src/media.ts new file mode 100644 index 0000000..4db964a --- /dev/null +++ b/packages/connector-runtime/src/media.ts @@ -0,0 +1,137 @@ +/** + * 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 }; +} diff --git a/packages/connector-runtime/src/registry.test.ts b/packages/connector-runtime/src/registry.test.ts index ff2609d..c7a9743 100644 --- a/packages/connector-runtime/src/registry.test.ts +++ b/packages/connector-runtime/src/registry.test.ts @@ -6,10 +6,10 @@ import type { ConnectorConnection } from "@tactical-media-broadcast-hub/core"; import { runWithPublishingContext } from "./execution-context.js"; const connection: ConnectorConnection = { id: "connection", workspaceId: "workspace", provider: "mastodon", channelId: "channel", fields: {}, createdAt: new Date(), credentialId: { kind: "credential-ref", id: "connection" } }; describe("production publishing registry", () => { - it("registers real Mastodon and Telegram connectors and rejects duplicates", () => { + it("registers real Mastodon, Telegram and Instagram connectors and rejects duplicates", () => { const pool = {} as Pool; const registry = createPublishingRegistry(pool); - expect(registry.list().map((c) => c.manifest().id).sort()).toEqual(["mastodon", "telegram"]); + expect(registry.list().map((c) => c.manifest().id).sort()).toEqual(["instagram", "mastodon", "telegram"]); expect(registry.get("mastodon").publish).toBeTypeOf("function"); expect(() => registry.register(registry.get("mastodon"))).toThrow(/already registered/); }); diff --git a/packages/connector-runtime/src/registry.ts b/packages/connector-runtime/src/registry.ts index 03a4f4a..b6b9e64 100644 --- a/packages/connector-runtime/src/registry.ts +++ b/packages/connector-runtime/src/registry.ts @@ -1,17 +1,22 @@ import type { Pool } from "pg"; import { ConnectorRegistry } from "@tactical-media-broadcast-hub/connector-sdk"; +import { createInstagramConnector } from "@tactical-media-broadcast-hub/connector-instagram"; import { createMastodonConnector } from "@tactical-media-broadcast-hub/connector-mastodon"; import { createTelegramConnector } from "@tactical-media-broadcast-hub/connector-telegram"; import { createPublishingCredentialAccessor } from "./credentials.js"; import { providerFetch } from "./provider-fetch.js"; import { executionSignal } from "./execution-context.js"; +import { createPublishingMediaAccess, mediaKekFromEnv, s3ObjectReaderFromEnv, type PublishingMediaAccess } from "./media.js"; /** Production composition registers real connectors, once per worker process. */ -export function createPublishingRegistry(pool: Pool): ConnectorRegistry { +export function createPublishingRegistry(pool: Pool, media?: PublishingMediaAccess): 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 assets = media ?? createPublishingMediaAccess(pool, s3ObjectReaderFromEnv(), mediaKekFromEnv()); + const options = { credentials: createPublishingCredentialAccessor(pool), fetchImpl, env: process.env, loadAsset: assets.loadAsset, openAsset: assets.openAsset }; registry.register(createMastodonConnector(options)); registry.register(createTelegramConnector(options)); + // Connects and refuses text-only posts; image publishing needs a public media URL (not provided yet). + registry.register(createInstagramConnector({ credentials: options.credentials, fetchImpl, env: process.env, loadAsset: assets.loadAsset })); return registry; } export { verifiedDedupeOrigin } from "./policy.js"; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b9e0838..eed8f71 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -502,6 +502,12 @@ importers: packages/connector-runtime: dependencies: + '@aws-sdk/client-s3': + specifier: ^3.1146.0 + version: 3.1146.0 + '@tactical-media-broadcast-hub/connector-instagram': + specifier: workspace:* + version: link:../connector-instagram '@tactical-media-broadcast-hub/connector-mastodon': specifier: workspace:* version: link:../connector-mastodon