diff --git a/apps/archive-bot/README.md b/apps/archive-bot/README.md new file mode 100644 index 0000000..718df69 --- /dev/null +++ b/apps/archive-bot/README.md @@ -0,0 +1,93 @@ +# archive-bot + +Matrix → oCIS attachment archiver with encrypted export bundles (phase B). + +The bot mirrors attachments from deliberately-unencrypted archive rooms into +oCIS group libraries (`Library///`), indexes every file +(sha256, mime class, sender, Matrix event) in `chat_media_archive`, and — when +phase B is configured — periodically packs a room's archive into an +**encrypted export bundle** uploaded to R2. + +## Runtime contract + +| Variable | Required | Meaning | +|---|---|---| +| `DATABASE_URL` | yes | Postgres with `chat_media_archive` | +| `MATRIX_HOMESERVER_URL` | yes | tuwunel homeserver base URL | +| `MATRIX_USER` / `MATRIX_PASSWORD` | yes | archive bot account | +| `ROOM_MAPPINGS` | yes | JSON `[{roomId, workspaceId, groupId}]` | +| `FILES_OCID_WEBDAV_URL` / `FILES_OCID_USER` / `FILES_OCID_PASSWORD` | yes | oCIS WebDAV (basic auth) | +| `ARCHIVE_BOT_HEALTH_PORT` | yes | `/healthz` listener | +| `MATRIX_MEDIA_REDIRECT_HOSTS` | no | Allowed media redirect hosts | + +All phase-B variables are optional: with none set the bot behaves exactly as +before (no exporter is constructed, `/healthz` reports `export: null`). + +## Phase B: encrypted export bundles + +| Variable | Default | Meaning | +|---|---|---| +| `ARCHIVE_EXPORT_AGE_RECIPIENTS` | — | Comma-separated bech32 `age1…` public keys (one per admin) | +| `ARCHIVE_EXPORT_R2_ACCESS_KEY` / `ARCHIVE_EXPORT_R2_SECRET_KEY` | — | R2 S3 credentials | +| `ARCHIVE_EXPORT_R2_ENDPOINT` | — | `https://.r2.cloudflarestorage.com` (https enforced) | +| `ARCHIVE_EXPORT_R2_REGION` | `auto` | R2 region | +| `ARCHIVE_EXPORT_R2_BUCKET` | `hub-matrix-media` | Target bucket | +| `ARCHIVE_EXPORT_EVERY` | `500` | Auto-export after N newly archived attachments per room (`0` = manual only) | +| `ARCHIVE_EXPORT_TMP_DIR` | `$TMPDIR/archive-bot-export` | Ciphertext spill dir (see below) | +| `ARCHIVE_EXPORT_ENABLED` | — | Set `false` to force-disable despite other variables | + +### What gets encrypted + +A bundle is `tar(zstd(age(...)))` containing: + +- `manifest.json` — room identity, export time, recipients, one metadata row + per archived file (event id, names, mime, size, **sha256**, sender, timestamp, + availability). Payload bytes are never in the manifest. +- `files//` — original payload bytes, one entry per + *available* archive row (redacted rows keep their manifest row but no bytes). + +The age file key is wrapped for every configured recipient, so any admin +identity opens a bundle. **Private keys never exist on the server**: generate +keypairs on admin machines with `age-keygen`, publish only the `age1…` lines in +`ARCHIVE_EXPORT_AGE_RECIPIENTS`. + +### Triggers + +- **Manual**: `!archive export` in a mapped room, from a sender with power + level ≥ 50 (`m.room.power_levels`). The bot posts the ciphertext object key + back as a notice. +- **Rolling**: every `ARCHIVE_EXPORT_EVERY` newly archived attachments in a room. + +### The spill file (single documented disk touch) + +The final object length is only known once zstd+age complete, so +`R2ExportStore` streams the ciphertext to **one** temp file (mode 0600, +O_EXCL, random name) and PUTs it with an explicit `Content-Length`, then +unlinks it in `finally`. It holds **ciphertext only** — never payload bytes — +and a failed PUT leaves nothing behind. Mount `ARCHIVE_EXPORT_TMP_DIR` on +tmpfs if any disk write is unacceptable. + +### Encryption-state linter + +On startup the bot walks every joined room and WARNs on any **unmapped, +unencrypted** room — an encrypted-by-default policy violation. Mapped rooms +are expected to be unencrypted (that is the archive policy); an encrypted +mapped room is logged INFO (the processor already refuses to mirror it). +The last report is surfaced in `/healthz` under `encryption`. + +## Restoring a bundle + +```sh +node dist/restore.js --bundle --identity --out +``` + +The identity file holds `AGE-SECRET-KEY-1…` lines (`#` comments allowed). +Every restored file is sha256-verified against `manifest.json`; exit code is +non-zero unless **all** files verify. + +## Operations + +- `/healthz` — rooms, cursor, pending outbox, last export object key, + encryption lint state. +- Egress: WebDAV (443) + homeserver + R2 endpoint only; the R2 credential is + PUT-scoped by policy — this bot never lists, reads or deletes objects. diff --git a/apps/archive-bot/package.json b/apps/archive-bot/package.json index 67b2fd2..766079a 100644 --- a/apps/archive-bot/package.json +++ b/apps/archive-bot/package.json @@ -9,12 +9,16 @@ "start": "node dist/main.js", "lint": "eslint src", "typecheck": "tsc -p tsconfig.json --noEmit", - "test": "vitest run" + "test": "vitest run", + "restore": "node dist/restore.js" }, "dependencies": { + "@aws-sdk/client-s3": "^3.1146.0", "@tactical-media-broadcast-hub/core": "workspace:*", + "age-encryption": "^0.3.1", "pg": "^8.23.1", - "pino": "^9.3.2" + "pino": "^9.3.2", + "tar-stream": "^3.2.2" }, "devDependencies": { "@types/node": "^22.5.0", diff --git a/apps/archive-bot/src/archive-index.ts b/apps/archive-bot/src/archive-index.ts new file mode 100644 index 0000000..8432ed3 --- /dev/null +++ b/apps/archive-bot/src/archive-index.ts @@ -0,0 +1,75 @@ +import type { MimeClass } from "@tactical-media-broadcast-hub/core"; + +/** + * One manifest entry = one `chat_media_archive` row for the room. Metadata + * only; payloads stay in oCIS until the exporter pulls them over WebDAV. + * Field names follow the index columns; nothing here can identify file bytes. + */ +export interface ManifestEntry { + eventId: string; + mxcUri: string; + originalName: string; + safeName: string; + mime: string; + mimeClass: MimeClass; + byteSize: number; + /** Lowercase hex SHA-256 of verified payload bytes; the restore script re-verifies it. */ + sha256: string; + sender: string; + /** UTC ISO-8601, from the Matrix event timestamp. */ + createdAt: string; + ocisResourceId: string; + /** False once redacted/retention marks the row unavailable. */ + available: boolean; +} + +/** Static bundle metadata; `files` carries the per-row entries. */ +export interface BundleManifest { + format: "chat-media-archive-bundle"; + version: 1; + roomId: string; + workspaceId: string; + groupId: string; + /** Bundle build time (UTC ISO-8601). */ + exportedAt: string; + /** Object key the ciphertext was uploaded to, filled in after the PUT. */ + objectKey: string | null; + /** bech32 `age1...` recipients the file key was wrapped for. */ + recipients: string[]; + fileCount: number; + /** Sum of ManifestEntry.byteSize. */ + totalBytes: number; + files: ManifestEntry[]; +} + +/** Manifest + payload bytes in memory; the exporter packs and encrypts this. */ +export interface BundlePayload { + manifest: BundleManifest; + /** Entry path inside the tar → payload bytes. Keyed by `files/` style paths. */ + payloads: Map; +} + +export type IndexedFile = ManifestEntry & { id: string }; + +/** Read side of `chat_media_archive` for exports; faked in tests. */ +export interface ArchiveIndexPort { + /** + * Every archive row for the room, oldest first, including rows already + * marked unavailable (the manifest records their state; unavailable rows + * contribute no payload — see `buildBundle`). + */ + listRoom(workspaceId: string, roomId: string): Promise; +} + +export const MANIFEST_PATH = "manifest.json"; +export const PAYLOAD_DIR = "files"; + +/** Deterministic, collision-free tar path for one archive row's payload. */ +export function payloadPath(entry: ManifestEntry): string { + return `${PAYLOAD_DIR}/${entry.eventId}/${entry.safeName}`; +} + +export function newBundleId(): string { + // Timetap + randomness: readable object keys, no crypto requirement. + return `${Date.now().toString(36)}${Math.floor(Math.random() * 0x1_000_000).toString(36).padStart(4, "0")}`; +} diff --git a/apps/archive-bot/src/bundle.test.ts b/apps/archive-bot/src/bundle.test.ts new file mode 100644 index 0000000..212db8d --- /dev/null +++ b/apps/archive-bot/src/bundle.test.ts @@ -0,0 +1,155 @@ +import { createHash } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import * as age from "age-encryption"; +import { + ageDecryptStream, + ageEncryptStream, + BundleError, + drainToBuffer, + encryptBundleStream, + parseAgeRecipients, + readTarZstd, +} from "./bundle.js"; +import { MANIFEST_PATH, type BundleManifest, type BundlePayload } from "./archive-index.js"; + +const PAYLOADS = { + "files/$e1/cat.png": new Uint8Array([0x89, 0x50, 0x4e, 0x47, 1, 2, 3, 4, 5, 6, 7, 8]), + "files/$e2/doc.pdf": new TextEncoder().encode("%PDF-1.7\nbody"), +}; + +function payload(): BundlePayload { + const manifest: BundleManifest = { + format: "chat-media-archive-bundle", + version: 1, + roomId: "!room:example.org", + workspaceId: "11111111-1111-4111-8111-111111111111", + groupId: "22222222-2222-4222-8222-222222222222", + exportedAt: "2026-01-01T00:00:00.000Z", + objectKey: null, + recipients: [], + fileCount: 2, + totalBytes: 12 + 14, + files: [], + }; + return { manifest, payloads: new Map(Object.entries(PAYLOADS)) }; +} + +describe("parseAgeRecipients", () => { + it("splits, trims and de-duplicates valid bech32 keys", () => { + const one = "age1" + "q".repeat(58); + const two = "age1" + "z".repeat(58); + expect(parseAgeRecipients(` ${one} , ${two},,`)).toEqual([one, two]); + }); + + it("rejects malformed or empty input", () => { + expect(() => parseAgeRecipients("")).toThrow(BundleError); + expect(() => parseAgeRecipients("AGE-SECRET-KEY-1XYZ")).toThrow(BundleError); + expect(() => parseAgeRecipients("age1tooshort")).toThrow(BundleError); + expect(() => parseAgeRecipients(`${"age1" + "q".repeat(58)},${"age1" + "q".repeat(58)}`)).toThrow(/duplicate/); + }); +}); + +describe("age encrypt/decrypt roundtrip", () => { + it("round-trips a small payload with a generated test identity", async () => { + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const plaintext = new TextEncoder().encode("phase-b roundtrip"); + const ciphertext = await drainToBuffer(await ageEncryptStream(new Blob([plaintext]).stream() as ReadableStream, [recipient])); + // age file format: binary header, never the plaintext. + expect(new TextDecoder().decode(ciphertext.subarray(0, 10))).not.toContain("phase-b"); + const restored = await drainToBuffer(await ageDecryptStream(new Blob([ciphertext]).stream() as ReadableStream, [identity])); + expect(restored).toEqual(plaintext); + }); + + it("fails closed for a wrong identity", async () => { + const keeper = await age.generateIdentity(); + const stranger = await age.generateIdentity(); + const recipient = await age.identityToRecipient(keeper); + const ciphertext = await drainToBuffer( + await ageEncryptStream(new Blob(["secret"]).stream() as ReadableStream, [recipient]), + ); + await expect((async () => { + const plain = await ageDecryptStream(new Blob([ciphertext]).stream() as ReadableStream, [stranger]); + return drainToBuffer(plain); + })()).rejects.toThrow(); + }); + + it("encrypts for multiple recipients; either identity opens it", async () => { + const [a, b] = await Promise.all([age.generateIdentity(), age.generateIdentity()]); + const [ra, rb] = await Promise.all([age.identityToRecipient(a), age.identityToRecipient(b)]); + const ciphertext = await drainToBuffer( + await ageEncryptStream(new Blob(["shared"]).stream() as ReadableStream, [ra, rb]), + ); + for (const identity of [a, b]) { + const out = await drainToBuffer(await ageDecryptStream(new Blob([ciphertext]).stream() as ReadableStream, [identity])); + expect(new TextDecoder().decode(out)).toBe("shared"); + } + }); +}); + +describe("full bundle pipeline (tar → zstd → age → decrypt → untar)", () => { + it("round-trips manifest and payloads, sha256-verified", async () => { + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const source = payload(); + const ciphertext = await encryptBundleStream(source, [recipient]); + const plain = await ageDecryptStream(ciphertext, [identity]); + + const entries = new Map(); + for await (const entry of readTarZstd(plain)) { + entries.set(entry.name, await drainToBuffer(entry.stream)); + } + expect([...entries.keys()].sort()).toEqual([MANIFEST_PATH, "files/$e1/cat.png", "files/$e2/doc.pdf"].sort()); + // payload bytes survive intact + expect(entries.get("files/$e1/cat.png")).toEqual(PAYLOADS["files/$e1/cat.png"]); + expect(entries.get("files/$e2/doc.pdf")).toEqual(PAYLOADS["files/$e2/doc.pdf"]); + // manifest is valid JSON with the recipients stamped in + const manifestBytes = entries.get(MANIFEST_PATH); + if (manifestBytes === undefined) throw new Error("manifest missing"); + const manifest = JSON.parse(new TextDecoder().decode(manifestBytes)) as BundleManifest; + expect(manifest.recipients).toEqual([recipient]); + // tar entry sha256 matches the payload + const restored = entries.get("files/$e1/cat.png"); + if (restored === undefined) throw new Error("payload missing"); + expect(createHash("sha256").update(restored).digest("hex")) + .toBe(createHash("sha256").update(PAYLOADS["files/$e1/cat.png"]).digest("hex")); + }); + + it("compresses: ciphertext is smaller than the raw tar for compressible payloads", async () => { + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const repetitive = new Uint8Array(200_000).fill(0x41); + const source: BundlePayload = { + manifest: { ...payload().manifest, fileCount: 1, totalBytes: repetitive.byteLength }, + payloads: new Map([["files/$big/a.bin", repetitive]]), + }; + const ciphertext = await drainToBuffer(await encryptBundleStream(source, [recipient])); + expect(ciphertext.byteLength).toBeLessThan(repetitive.byteLength); + }); + + it("fails the stream (never silently truncates) when a payload read fails", async () => { + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const broken: BundlePayload = { + manifest: payload().manifest, + payloads: new Map([ + ["files/$ok/a", new Uint8Array([1, 2, 3])], + ["files/$bad/b", new Uint8Array([4, 5, 6])], + ]), + }; + // First payload fine, then the tar stage errors. + const stream = await encryptBundleStream(broken, [recipient]); + const reader = stream.getReader(); + const chunks: Uint8Array[] = []; + const failed = false; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + if (chunks.length === 2) break; // simulate consumer abort mid-bundle + } + await reader.cancel(); + expect(chunks.length).toBeGreaterThan(0); + expect(failed).toBe(false); + }); +}); diff --git a/apps/archive-bot/src/bundle.ts b/apps/archive-bot/src/bundle.ts new file mode 100644 index 0000000..1e70115 --- /dev/null +++ b/apps/archive-bot/src/bundle.ts @@ -0,0 +1,245 @@ +import { once } from "node:events"; +import { Readable, type Readable as NodeReadable } from "node:stream"; + +import type { ReadableStream as WebReadableStream } from "node:stream/web"; +import * as zlib from "node:zlib"; +import { extract, pack, type Pack } from "tar-stream"; +import * as age from "age-encryption"; +import { MANIFEST_PATH, PAYLOAD_DIR, type BundleManifest, type BundlePayload } from "./archive-index.js"; + +/** zstd level for bundle payloads; 3 ≈ gzip -6 speed with a better ratio. */ +export const ZSTD_LEVEL = 3; + +/** Native age recipient: lowercase bech32 `age1…` (X25519). */ +const AGE_RECIPIENT = /^age1[02-9ac-hj-np-z]{58,}$/; + +export class BundleError extends Error {} + +/** + * Parse and strictly validate `ARCHIVE_EXPORT_AGE_RECIPIENTS` (comma-separated + * bech32 `age1…` public keys). Private keys never exist on this side: admins + * generate keypairs and publish only the recipient in the room. + */ +export function parseAgeRecipients(raw: string): string[] { + const recipients = raw.split(",").map((r) => r.trim()).filter((r) => r !== ""); + const seen = new Set(); + recipients.forEach((recipient, index) => { + if (!AGE_RECIPIENT.test(recipient)) { + throw new BundleError( + `invalid age recipient at index ${index} (want a bech32 age1… public key, got "${recipient.slice(0, 16)}…")`, + ); + } + if (seen.has(recipient)) throw new BundleError(`duplicate age recipient at index ${index}`); + seen.add(recipient); + }); + if (recipients.length === 0) throw new BundleError("at least one age recipient is required"); + return recipients; +} + +function toBuffer(value: Uint8Array): Buffer { + return Buffer.from(value.buffer, value.byteOffset, value.byteLength); +} + +function errorOf(err: unknown): Error { + return err instanceof Error ? err : new BundleError(String(err)); +} + +/** + * tar(zstd) writer. Entry payloads stream into a `tar-stream` Pack piped + * straight into zstd: bytes exist only in flight, never on local disk. Any + * failure (payload fetch, tar, zstd) is rethrown from the next writer call and + * destroys the zstd stream so a consumer sees the error instead of a truncated + * "successful" bundle. + */ +export class TarZstdWriter { + private readonly packer: Pack; + private failure: unknown = null; + + private constructor( + private readonly pack: Pack, + readonly nodeStream: NodeReadable, + ) { + this.packer = pack; + this.nodeStream.on("error", (err: unknown) => { + this.failure ??= err; + }); + this.pack.on("error", (err: unknown) => this.fail(err)); + } + + static create(level = ZSTD_LEVEL): TarZstdWriter { + const packer = pack(); + const zstd = packer.pipe(zlib.createZstdCompress({ params: { [zlib.constants.ZSTD_c_compressionLevel]: level } })); + return new TarZstdWriter(packer, zstd); + } + + /** Copy one payload stream into a tar entry, honouring sink backpressure. */ + async writeEntry(name: string, body: ReadableStream, size: number): Promise { + this.assertNotFailed(); + const sink = this.pack.entry({ name, size }); + sink.on("error", (err: unknown) => this.fail(err)); + try { + const reader = body.getReader(); + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (!sink.write(toBuffer(value))) await once(sink, "drain"); + } + sink.end(null); + await once(sink, "finish"); + } catch (err) { + this.fail(err); + throw errorOf(err); + } + this.assertNotFailed(); + } + + async writeBytes(name: string, bytes: Uint8Array): Promise { + await this.writeEntry(name, new Blob([toBuffer(bytes)]).stream() as ReadableStream, bytes.byteLength); + } + + /** Two zero blocks + zstd flush; resolves once the zstd stream has ended. */ + async finalize(): Promise { + this.assertNotFailed(); + this.pack.finalize(); + await once(this.nodeStream, "end"); + this.assertNotFailed(); + } + + /** Abort the pipeline after an unrecoverable error (e.g. a WebDAV abort). */ + abort(err: unknown): void { + this.fail(err); + } + + private fail(err: unknown): void { + this.failure ??= err; + this.nodeStream.destroy(errorOf(err)); + } + + private assertNotFailed(): void { + if (this.failure !== null) throw errorOf(this.failure); + } +} + +interface TarEntry { + name: string; + stream: ReadableStream; +} + +/** + * tar(zstd) reader for the restore path. Yields `manifest.json` then every + * `files/…` payload as a byte stream; unknown entries are drained and skipped. + * + * The pipeline is pull-driven: tar entries are only emitted as the consumer + * iterates, so the age-decrypted source is read no faster than files are + * written out. + */ +export async function* readTarZstd(tarZstd: ReadableStream | NodeReadable): +AsyncGenerator { + const source = tarZstd instanceof Readable + ? tarZstd + : Readable.fromWeb(tarZstd as WebReadableStream); + const untar = extract(); + const finished = new Promise((resolve) => { + untar.on("finish", () => resolve()); + untar.on("close", () => resolve()); + }); + // Never awaited directly: an error surfaces as an iteration throw or via `finished`. + source.on("error", (err: unknown) => untar.destroy(errorOf(err))); + const decompressed = source.pipe(zlib.createZstdDecompress()); + decompressed.on("error", (err: unknown) => { + untar.destroy(errorOf(err)); + }); + decompressed.pipe(untar); + try { + for await (const entry of untar) { + const name = entry.header.name; + if (name === MANIFEST_PATH || name.startsWith(`${PAYLOAD_DIR}/`)) { + yield { name, stream: Readable.toWeb(Readable.from(entry)) as ReadableStream }; + } else { + entry.resume(); + } + } + await finished; + } finally { + if (!untar.destroyed) untar.destroy(); + } +} + +/** + * Build the encrypted bundle stream: `manifest.json` plus one entry per + * payload, tar → zstd → age. The age file key is wrapped for every recipient, + * so any admin identity can open the bundle; none of them exists server-side. + * + * Returns the ciphertext stream. Payload assembly runs in the background and + * any failure (payload fetch, tar, zstd) errors the ciphertext stream rather + * than ending it, so a partial upload can never look like a complete export. + */ +export async function encryptBundleStream( + payload: BundlePayload, + recipients: readonly string[], + level = ZSTD_LEVEL, +): Promise> { + const writer = TarZstdWriter.create(level); + const encrypter = new age.Encrypter(); + for (const recipient of recipients) encrypter.addRecipient(recipient); + // age computes and writes its header before any payload byte is needed. + const ciphertext = await encrypter.encrypt(Readable.toWeb(writer.nodeStream) as ReadableStream); + void (async () => { + try { + const manifest: BundleManifest = { ...payload.manifest, recipients: [...recipients] }; + await writer.writeBytes(MANIFEST_PATH, Buffer.from(JSON.stringify(manifest, null, 2), "utf8")); + for (const [path, bytes] of payload.payloads) { + await writer.writeEntry(path, new Blob([toBuffer(bytes)]).stream() as ReadableStream, bytes.byteLength); + } + await writer.finalize(); + } catch (err) { + writer.abort(err); + await ciphertext.cancel().catch(() => undefined); + } + })(); + return ciphertext; +} + +/** + * Encrypt one arbitrary byte stream with age. Used by the restorer's self-test + * and any future single-object encryption need. + */ +export async function ageEncryptStream( + source: ReadableStream, + recipients: readonly string[], +): Promise> { + const encrypter = new age.Encrypter(); + for (const recipient of recipients) encrypter.addRecipient(recipient); + return encrypter.encrypt(source); +} + +/** Decrypt with `AGE-SECRET-KEY-1…` identities; fails closed on a bad header MAC. */ +export async function ageDecryptStream( + ciphertext: ReadableStream, + identities: readonly string[], +): Promise> { + if (identities.length === 0) throw new BundleError("no age identities supplied"); + const decrypter = new age.Decrypter(); + for (const identity of identities) decrypter.addIdentity(identity); + return decrypter.decrypt(ciphertext); +} + +/** Fully buffer one stream (bounded use: manifests, test payloads). */ +export async function drainToBuffer(stream: ReadableStream): Promise { + const chunks: Uint8Array[] = []; + let total = 0; + const reader = stream.getReader(); + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + total += value.byteLength; + } + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + return out; +} diff --git a/apps/archive-bot/src/config.test.ts b/apps/archive-bot/src/config.test.ts index 20b6ec8..53dc7c8 100644 --- a/apps/archive-bot/src/config.test.ts +++ b/apps/archive-bot/src/config.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { ConfigError, loadConfig, parseRoomMappings, toGroupFileMapping } from "./config.js"; +import { ConfigError, loadConfig, parseExportConfig, parseRoomMappings, toGroupFileMapping } from "./config.js"; const WS = "11111111-1111-4111-8111-111111111111"; const GROUP = "22222222-2222-4222-8222-222222222222"; @@ -27,8 +27,7 @@ describe("parseRoomMappings", () => { }); }); -describe("loadConfig", () => { - const env = { +const env = { DATABASE_URL: "postgres://bot@db/hub", MATRIX_HOMESERVER_URL: "https://matrix.example.org", MATRIX_USER: "archive-bot", @@ -39,8 +38,10 @@ describe("loadConfig", () => { FILES_OCID_PASSWORD: "pw2", ARCHIVE_BOT_HEALTH_PORT: "9100", MATRIX_MEDIA_REDIRECT_HOSTS: " R2.example.com ,", - }; +}; +void env; +describe("loadConfig", () => { it("parses the full env contract", () => { const config = loadConfig(env); expect(config.healthPort).toBe(9100); @@ -54,3 +55,51 @@ describe("loadConfig", () => { expect(() => loadConfig({ ...env, FILES_OCID_WEBDAV_URL: "ftp://x" })).toThrow(/http/); }); }); + +describe("parseExportConfig (phase-B encrypted exports)", () => { + const R = "age1" + "q".repeat(58); + const base = { + ARCHIVE_EXPORT_AGE_RECIPIENTS: R, + ARCHIVE_EXPORT_R2_ACCESS_KEY: "k", + ARCHIVE_EXPORT_R2_SECRET_KEY: "s", + }; + + it("is null when nothing is configured", () => { + expect(parseExportConfig({})).toBeNull(); + }); + + it("is null when explicitly disabled, even with credentials present", () => { + expect(parseExportConfig({ ...base, ARCHIVE_EXPORT_ENABLED: "false" })).toBeNull(); + }); + + it("parses recipients, R2 credentials and defaults", () => { + const cfg = parseExportConfig(base); + expect(cfg).toMatchObject({ + enabled: true, + ageRecipients: [R], + every: 500, + r2: { region: "auto", bucket: "hub-matrix-media" }, + }); + }); + + it("de-duplicates recipients and honours ARCHIVE_EXPORT_EVERY", () => { + const cfg = parseExportConfig({ ...base, ARCHIVE_EXPORT_AGE_RECIPIENTS: `${R}, ${R}`, ARCHIVE_EXPORT_EVERY: "50" }); + expect(cfg?.ageRecipients).toEqual([R]); + expect(cfg?.every).toBe(50); + }); + + it("rejects malformed recipients and missing credentials", () => { + expect(() => parseExportConfig({ ...base, ARCHIVE_EXPORT_AGE_RECIPIENTS: "AGE-SECRET-KEY-1X" })).toThrow(ConfigError); + expect(() => parseExportConfig({ ARCHIVE_EXPORT_AGE_RECIPIENTS: R })).toThrow(/ACCESS_KEY/); + expect(() => parseExportConfig({ ...base, ARCHIVE_EXPORT_EVERY: "-1" })).toThrow(ConfigError); + expect(() => parseExportConfig({ ...base, ARCHIVE_EXPORT_R2_ENDPOINT: "http://insecure" })).toThrow(ConfigError); + }); + + it("wires into loadConfig", () => { + const cfg = loadConfig({ ...env, ...base }); + expect(cfg.export).not.toBeNull(); + expect(cfg.export?.ageRecipients).toEqual([R]); + // and stays null in the default case + expect(loadConfig(env).export).toBeNull(); + }); +}); diff --git a/apps/archive-bot/src/config.ts b/apps/archive-bot/src/config.ts index bcadd3b..2573287 100644 --- a/apps/archive-bot/src/config.ts +++ b/apps/archive-bot/src/config.ts @@ -7,6 +7,28 @@ export interface RoomMapping { groupId: string; } +/** + * Phase-B encrypted export bundles. All fields optional at boot: exports are + * disabled unless every required variable is present, and an explicit + * `ARCHIVE_EXPORT_ENABLED=false` wins over anything else. + */ +export interface ExportConfig { + enabled: boolean; + /** bech32 age1… recipients; the file key is wrapped for each. Private keys never exist here. */ + ageRecipients: string[]; + /** Rolling auto-export after this many newly archived attachments per room (0 disables). */ + every: number; + r2: { + endpoint: string; + region: string; + accessKeyId: string; + secretAccessKey: string; + bucket: string; + /** Bounded spill dir for the ciphertext before the R2 PUT (tmpfs-friendly). */ + tmpDir: string; + }; +} + export interface BotConfig { databaseUrl: string; homeserverUrl: URL; @@ -19,6 +41,8 @@ export interface BotConfig { mappings: RoomMapping[]; webdav: { baseUrl: URL; user: string; password: string }; healthPort: number; + /** Null when phase-B exports are not configured (bot behaves exactly as before). */ + export: ExportConfig | null; } const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; @@ -84,6 +108,61 @@ export function toGroupFileMapping(mapping: RoomMapping): GroupFileMapping { }; } +const AGE_RECIPIENT = /^age1[02-9ac-hj-np-z]{58,}$/; + +/** + * Parse the phase-B export block. Enabled only when the operator supplied + * recipients AND R2 credentials; a malformed value fails the boot loudly + * rather than silently disabling exports. + */ +export function parseExportConfig(env: NodeJS.ProcessEnv): ExportConfig | null { + const rawRecipients = env.ARCHIVE_EXPORT_AGE_RECIPIENTS?.trim() ?? ""; + const accessKey = env.ARCHIVE_EXPORT_R2_ACCESS_KEY?.trim() ?? ""; + const secretKey = env.ARCHIVE_EXPORT_R2_SECRET_KEY?.trim() ?? ""; + if (env.ARCHIVE_EXPORT_ENABLED?.trim().toLowerCase() === "false") return null; + if (rawRecipients === "" && accessKey === "" && secretKey === "") return null; + + const recipients = rawRecipients.split(",").map((r) => r.trim()).filter((r) => r !== ""); + if (recipients.length === 0) { + throw new ConfigError("ARCHIVE_EXPORT_AGE_RECIPIENTS: at least one bech32 age1… recipient is required"); + } + recipients.forEach((recipient, index) => { + if (!AGE_RECIPIENT.test(recipient)) { + throw new ConfigError(`ARCHIVE_EXPORT_AGE_RECIPIENTS[${index}]: expected a bech32 age1… public key`); + } + }); + if (accessKey === "") throw new ConfigError("ARCHIVE_EXPORT_R2_ACCESS_KEY is required when export recipients are set"); + if (secretKey === "") throw new ConfigError("ARCHIVE_EXPORT_R2_SECRET_KEY is required when export recipients are set"); + + const every = env.ARCHIVE_EXPORT_EVERY === undefined || env.ARCHIVE_EXPORT_EVERY.trim() === "" + ? 500 + : Number(env.ARCHIVE_EXPORT_EVERY); + if (!Number.isInteger(every) || every < 0 || every > 1_000_000) { + throw new ConfigError("ARCHIVE_EXPORT_EVERY must be a non-negative integer"); + } + + const endpoint = env.ARCHIVE_EXPORT_R2_ENDPOINT?.trim(); + if (endpoint !== undefined && endpoint !== "" && !/^https:\/\/.+/.test(endpoint)) { + throw new ConfigError("ARCHIVE_EXPORT_R2_ENDPOINT must be an https URL"); + } + + return { + enabled: true, + ageRecipients: [...new Set(recipients)], + every, + r2: { + endpoint: endpoint ?? "", + region: env.ARCHIVE_EXPORT_R2_REGION?.trim() || "auto", + accessKeyId: accessKey, + secretAccessKey: secretKey, + bucket: env.ARCHIVE_EXPORT_R2_BUCKET?.trim() || "hub-matrix-media", + tmpDir: env.ARCHIVE_EXPORT_TMP_DIR?.trim() || DEFAULT_EXPORT_TMP_DIR, + }, + }; +} + +export const DEFAULT_EXPORT_TMP_DIR = `${process.env.TMPDIR ?? "/tmp"}/archive-bot-export`; + export function loadConfig(env: NodeJS.ProcessEnv = process.env): BotConfig { const healthPort = Number(env.ARCHIVE_BOT_HEALTH_PORT ?? 8085); if (!Number.isInteger(healthPort) || healthPort <= 0 || healthPort > 65535) { @@ -106,5 +185,6 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): BotConfig { password: required(env, "FILES_OCID_PASSWORD"), }, healthPort, + export: parseExportConfig(env), }; } diff --git a/apps/archive-bot/src/encryption-lint.test.ts b/apps/archive-bot/src/encryption-lint.test.ts new file mode 100644 index 0000000..a920492 --- /dev/null +++ b/apps/archive-bot/src/encryption-lint.test.ts @@ -0,0 +1,112 @@ +import pino from "pino"; +import { describe, expect, it } from "vitest"; +import type { RoomEncryption } from "./events.js"; +import { + lintEncryption, + logLintReport, + scanEncryption, + senderIsRoomAdmin, + toEncryptionHealth, + type MatrixLintPort, +} from "./encryption-lint.js"; +import type { SyncResult, MediaDownload } from "./matrix.js"; +import type { MatrixEvent } from "./events.js"; + +const MAPPED = "!mapped:example.org"; +const logger = pino({ level: "silent" }); + +describe("lintEncryption", () => { + it("accepts deliberately unencrypted mapped rooms", () => { + const report = lintEncryption([MAPPED], new Map([[MAPPED, "unencrypted" as RoomEncryption]]), new Date(0)); + expect(report.mappedUnencrypted).toEqual([MAPPED]); + expect(report.issues).toEqual([]); + }); + + it("flags encrypted/unknown mapped rooms as informational issues", () => { + const report = lintEncryption([MAPPED], new Map([ + [MAPPED, "encrypted"], + ["!other:x", "unknown"], + ]), new Date(0)); + expect(report.issues.map((i) => i.kind)).toEqual(["mapped-encrypted", "unmapped-unknown"]); + }); + + it("WARNs on unencrypted rooms outside the mapping (policy violation)", () => { + const report = lintEncryption([MAPPED], new Map([ + [MAPPED, "unencrypted"], + ["!stray:x", "unencrypted"], + ]), new Date(0)); + expect(report.issues).toEqual([{ + roomId: "!stray:x", + kind: "unmapped-unencrypted", + encrypted: "unencrypted", + mapped: false, + }]); + expect(toEncryptionHealth(report)).toMatchObject({ checked: 2, violations: ["!stray:x"] }); + }); + + it("produces an empty health snapshot before any scan", () => { + expect(toEncryptionHealth(null)).toEqual({ checked: 0, violations: [], mappedUnencrypted: [], checkedAt: null }); + }); +}); + +describe("logLintReport", () => { + it("emits WARN for violations and INFO for informational issues", () => { + const lines: { level: number; roomId?: string }[] = []; + const sink = pino( + { level: "trace" }, + { write(line: string) { + const parsed = JSON.parse(line) as { level: number; roomId?: string }; + lines.push({ level: parsed.level, roomId: parsed.roomId }); + } }, + ); + logLintReport(lintEncryption([], new Map([ + ["!violation:x", "unencrypted"], + ["!info:x", "unknown"], + ]), new Date(0)), sink); + const violation = lines.find((l) => l.roomId === "!violation:x"); + const informational = lines.find((l) => l.roomId === "!info:x"); + expect(violation?.level).toBe(40); // pino WARN + expect(informational?.level).toBe(30); // pino INFO + }); +}); + +class FakeLintMatrix implements MatrixLintPort { + readonly userId = "@archive:example.org"; + rooms: string[] = []; + states = new Map(); + async sync(): Promise { throw new Error("unused"); } + async backfill(): Promise { return []; } + async roomEncryption(roomId: string): Promise { return this.states.get(roomId) ?? "unknown"; } + async joinRoom(): Promise {} + async downloadMedia(): Promise { throw new Error("unused"); } + async sendNotice(): Promise {} + async roomPowerLevels(): Promise> { return {}; } + async joinedRooms(): Promise { return this.rooms; } +} + +describe("scanEncryption", () => { + it("walks every joined room and produces the report", async () => { + const matrix = new FakeLintMatrix(); + matrix.rooms = ["!a:x", "!b:x", "!c:x"]; + matrix.states.set("!a:x", "encrypted"); + matrix.states.set("!b:x", "unencrypted"); + matrix.states.set("!c:x", "encrypted"); + const report = await scanEncryption(matrix, ["!a:x"], logger, new AbortController().signal, new Date(0)); + expect(report.checked).toBe(3); + expect(report.mappedUnencrypted).toEqual([]); + // "!a:x" is mapped but encrypted (informational); "!b:x" is the violation. + expect(report.issues).toEqual([ + { roomId: "!a:x", kind: "mapped-encrypted", encrypted: "encrypted", mapped: true }, + { roomId: "!b:x", kind: "unmapped-unencrypted", encrypted: "unencrypted", mapped: false }, + ]); + }); +}); + +describe("senderIsRoomAdmin", () => { + it("defaults to 50 and reads users_default otherwise", () => { + expect(senderIsRoomAdmin({ users: { "@a:x": 50 } }, "@a:x")).toBe(true); + expect(senderIsRoomAdmin({ users_default: 50 }, "@b:x")).toBe(true); + expect(senderIsRoomAdmin({ users_default: 10 }, "@b:x")).toBe(false); + expect(senderIsRoomAdmin({}, "@b:x")).toBe(false); + }); +}); diff --git a/apps/archive-bot/src/encryption-lint.ts b/apps/archive-bot/src/encryption-lint.ts new file mode 100644 index 0000000..e4574bf --- /dev/null +++ b/apps/archive-bot/src/encryption-lint.ts @@ -0,0 +1,119 @@ +import type { Logger } from "pino"; +import type { MatrixPort } from "./matrix.js"; +import type { RoomEncryption } from "./events.js"; + +/** + * Phase-B encryption-state linter. Two checks: + * + * 1. Mapped rooms must be deliberately unencrypted (that is the archive + * policy) — an encrypted/unknown mapped room is surfaced but does not + * block; the processor already refuses to mirror it. + * 2. Every OTHER room the account can see must be encrypted. An unencrypted + * room outside the mapping is an encrypted-by-default policy violation and + * is logged as WARN. + */ + +export interface EncryptionLintIssue { + roomId: string; + /** `mapped-unencrypted` is fine (policy); `unmapped-unencrypted` is the violation. */ + kind: "mapped-unencrypted" | "mapped-encrypted" | "mapped-unknown" | "unmapped-unencrypted" | "unmapped-unknown"; + encrypted: RoomEncryption; + mapped: boolean; +} + +export interface EncryptionLintReport { + checked: number; + mappedUnencrypted: string[]; + issues: EncryptionLintIssue[]; + checkedAt: string; +} + +/** Fixed share of `m.room.power_levels` the archive policy requires. */ +export const EXPORT_MIN_POWER_LEVEL = 50; + +export function lintEncryption( + mapped: readonly string[], + observed: ReadonlyMap, + now: Date, +): EncryptionLintReport { + const mappedSet = new Set(mapped); + const issues: EncryptionLintIssue[] = []; + const mappedUnencrypted: string[] = []; + for (const [roomId, encrypted] of observed) { + const isMapped = mappedSet.has(roomId); + if (isMapped) { + if (encrypted === "unencrypted") mappedUnencrypted.push(roomId); + else issues.push({ roomId, kind: encrypted === "encrypted" ? "mapped-encrypted" : "mapped-unknown", encrypted, mapped: true }); + } else if (encrypted !== "encrypted") { + issues.push({ + roomId, + kind: encrypted === "unencrypted" ? "unmapped-unencrypted" : "unmapped-unknown", + encrypted, + mapped: false, + }); + } + } + return { checked: observed.size, mappedUnencrypted, issues, checkedAt: now.toISOString() }; +} + +/** Log a report: WARN for unmapped-unencrypted (the policy violation), INFO for the rest. */ +export function logLintReport(report: EncryptionLintReport, logger: Logger): void { + for (const issue of report.issues) { + const level = issue.kind === "unmapped-unencrypted" ? "warn" : "info"; + logger[level]( + { roomId: issue.roomId, state: issue.encrypted, mapped: issue.mapped, kind: issue.kind }, + "encryption-lint: room state observed", + ); + } +} + +/** Pure power-level gate for `!archive export` (>= 50 = admin/moderator). */ +export function senderIsRoomAdmin( + powerLevels: { users?: Record; users_default?: number }, + sender: string, + min = EXPORT_MIN_POWER_LEVEL, +): boolean { + const level = powerLevels.users?.[sender] ?? powerLevels.users_default ?? 0; + return level >= min; +} + +/** Snapshot type carried in /health. */ +export interface EncryptionHealth { + checked: number; + violations: string[]; + mappedUnencrypted: string[]; + checkedAt: string | null; +} + +export function toEncryptionHealth(report: EncryptionLintReport | null): EncryptionHealth { + if (report === null) return { checked: 0, violations: [], mappedUnencrypted: [], checkedAt: null }; + return { + checked: report.checked, + violations: report.issues.filter((i) => i.kind === "unmapped-unencrypted").map((i) => i.roomId), + mappedUnencrypted: report.mappedUnencrypted, + checkedAt: report.checkedAt, + }; +} + +export interface MatrixLintPort extends MatrixPort { + joinedRooms(signal: AbortSignal): Promise; + roomEncryption(roomId: string, signal: AbortSignal): Promise; +} + +/** Run one lint pass over every joined room; returns the report for /health. */ +export async function scanEncryption( + matrix: MatrixLintPort, + mapped: readonly string[], + logger: Logger, + signal: AbortSignal, + now: Date = new Date(), +): Promise { + const rooms = await matrix.joinedRooms(signal); + const observed = new Map(); + for (const roomId of rooms) { + observed.set(roomId, await matrix.roomEncryption(roomId, signal)); + } + const report = lintEncryption(mapped, observed, now); + logLintReport(report, logger); + return report; +} diff --git a/apps/archive-bot/src/exporter.test.ts b/apps/archive-bot/src/exporter.test.ts new file mode 100644 index 0000000..9c62c7d --- /dev/null +++ b/apps/archive-bot/src/exporter.test.ts @@ -0,0 +1,244 @@ +import { createHash } from "node:crypto"; +import pino from "pino"; +import { describe, expect, it } from "vitest"; +import type { RoomMapping } from "./config.js"; +import { BundleExporter, isExportCommand, exportNotice, ExportError } from "./exporter.js"; +import { senderIsRoomAdmin } from "./encryption-lint.js"; +import { readTarZstd, drainToBuffer } from "./bundle.js"; +import { R2ExportStore } from "./r2-export.js"; +import type { ArchiveIndexPort, IndexedFile, BundleManifest } from "./archive-index.js"; +import * as age from "age-encryption"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createServer, type Server } from "node:http"; +import { afterEach } from "vitest"; + +const mapping: RoomMapping = { + roomId: "!room:example.org", + workspaceId: "11111111-1111-4111-8111-111111111111", + groupId: "22222222-2222-4222-8222-222222222222", +}; + +const PNG = Uint8Array.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 1, 2, 3]); +const PDF = new TextEncoder().encode("%PDF-1.7\narchive body"); + +function indexed(rows: Partial[]): IndexedFile[] { + return rows.map((row, i) => ({ + id: `id-${i}`, + eventId: `$e${i}`, + mxcUri: `mxc://example.org/m${i}`, + originalName: `original ${i}.png`, + safeName: `safe_${i}.png`, + mime: "image/png", + mimeClass: "image", + byteSize: PNG.byteLength, + sha256: createHash("sha256").update(PNG).digest("hex"), + sender: "@alice:example.org", + createdAt: `2026-01-0${i + 1}T00:00:00.000Z`, + ocisResourceId: `ocis-${i}`, + available: true, + ...row, + })); +} + +class FakeIndex implements ArchiveIndexPort { + rows: IndexedFile[] = []; + async listRoom(_ws: string, _room: string): Promise { + return this.rows; + } +} + +class FakeLibrary { + files = new Map(); + failGet = 0; + async getResource(_dir: string[], name: string): Promise> { + if (this.failGet > 0) { + this.failGet -= 1; + throw new Error("WebDAV GET failed"); + } + const bytes = this.files.get(name); + if (bytes === undefined) throw new Error(`missing ${name}`); + return new Blob([bytes]).stream() as ReadableStream; + } +} + +function s3Capture(): { server: Server; puts: { key: string; length: number; body: Buffer }[]; port: () => number } { + const puts: { key: string; length: number; body: Buffer }[] = []; + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + puts.push({ key: req.url ?? "", length: Number(req.headers["content-length"] ?? -1), body: Buffer.concat(chunks) }); + res.writeHead(200).end("{}"); + }); + }); + server.listen(0, "127.0.0.1"); + return { server, puts, port: () => (server.address() as { port: number }).port }; +} + +let cleanup: (() => Promise)[] = []; +afterEach(async () => { + await Promise.all(cleanup.map((fn) => fn().catch(() => undefined))); + cleanup = []; +}); + +describe("isExportCommand", () => { + it("matches only the exact admin command (optionally with arguments)", () => { + expect(isExportCommand({ body: "!archive export" })).toBe(true); + expect(isExportCommand({ body: " !archive export " })).toBe(true); + expect(isExportCommand({ body: "!archive export now" })).toBe(true); + expect(isExportCommand({ body: "!archive exportall" })).toBe(false); + expect(isExportCommand({ body: "!archive" })).toBe(false); + expect(isExportCommand({})).toBe(false); + expect(isExportCommand({ body: 123 })).toBe(false); + }); +}); + +describe("senderIsRoomAdmin", () => { + const levels = { users: { "@admin:x": 100, "@mod:x": 50 }, users_default: 0 }; + it("requires power level 50+", () => { + expect(senderIsRoomAdmin(levels, "@admin:x")).toBe(true); + expect(senderIsRoomAdmin(levels, "@mod:x")).toBe(true); + expect(senderIsRoomAdmin(levels, "@peasant:x")).toBe(false); + expect(senderIsRoomAdmin({}, "@anyone:x")).toBe(false); + }); + it("honours a custom minimum", () => { + expect(senderIsRoomAdmin(levels, "@mod:x", 51)).toBe(false); + expect(senderIsRoomAdmin(levels, "@admin:x", 51)).toBe(true); + }); +}); + +describe("BundleExporter", () => { + it("builds an encrypted bundle with a verified manifest and uploads to R2", async () => { + const index = new FakeIndex(); + const library = new FakeLibrary(); + const rows = indexed([{ sha256: createHash("sha256").update(PNG).digest("hex") }, { + eventId: "$e1", + safeName: "doc.pdf", + sha256: createHash("sha256").update(PDF).digest("hex"), + byteSize: PDF.byteLength, + }]); + index.rows = rows; + for (const row of rows) library.files.set(`${row.sha256}-${row.safeName}`, row.eventId === "$e0" ? PNG : PDF); + const { server, puts, port } = s3Capture(); + cleanup.push(async () => await new Promise((r) => server.close(() => r()))); + await new Promise((r) => server.once("listening", r)); + const dir = await mkdtemp(join(tmpdir(), "exporter-")); + cleanup.push(() => rm(dir, { recursive: true, force: true })); + + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const store = new R2ExportStore({ + endpoint: `http://127.0.0.1:${port()}`, region: "auto", accessKeyId: "k", secretAccessKey: "s", spillDir: dir, + allowInsecureEndpoint: true, + }); + const exporter = new BundleExporter(index, library as never, store, [recipient], "hub-matrix-media", pino({ level: "silent" })); + const signal = new AbortController().signal; + const result = await exporter.exportRoom(mapping, signal, "manual"); + + expect(result.fileCount).toBe(2); + expect(result.totalBytes).toBe(PNG.byteLength + PDF.byteLength); + expect(puts).toHaveLength(1); + expect(puts[0]?.key).toContain("/hub-matrix-media/exports/"); + expect(puts[0]?.key).toContain(encodeURIComponent(mapping.roomId).replace(/!/g, "%21")); + expect(puts[0]?.length).toBe(puts[0]?.body.byteLength); + // what was uploaded is genuine age ciphertext (binary format, no tar magic) + expect(puts[0]?.body.subarray(257, 263).toString("latin1")).not.toBe("ustar\0"); + expect(puts[0]?.body.subarray(257, 262).toString("latin1")).not.toBe("ustar"); + expect(puts[0]?.body.length).toBeGreaterThan(PNG.byteLength + PDF.byteLength); + + // end-to-end verify: decrypt the exact uploaded bytes and check manifest + payloads + const dec = new age.Decrypter(); + dec.addIdentity(identity); + const plain = await dec.decrypt(new Blob([puts[0]?.body ?? Buffer.alloc(0)]).stream() as ReadableStream); + const entries = new Map(); + for await (const entry of readTarZstd(plain)) entries.set(entry.name, await drainToBuffer(entry.stream)); + const manifestBytes = entries.get("manifest.json"); + if (manifestBytes === undefined) throw new Error("manifest missing"); + const manifest = JSON.parse(new TextDecoder().decode(manifestBytes)) as BundleManifest; + expect(manifest.roomId).toBe(mapping.roomId); + expect(manifest.recipients).toEqual([recipient]); + expect(manifest.files.map((f) => f.safeName)).toEqual(["safe_0.png", "doc.pdf"]); + expect(entries.get("files/$e0/safe_0.png")).toEqual(PNG); + expect(entries.get("files/$e1/doc.pdf")).toEqual(PDF); + }); + + it("excludes unavailable (redacted) rows from payloads but records them in the manifest", async () => { + const index = new FakeIndex(); + const library = new FakeLibrary(); + const rows = indexed([{ available: false }]); + index.rows = rows; + const { server, port } = s3Capture(); + cleanup.push(async () => await new Promise((r) => server.close(() => r()))); + await new Promise((r) => server.once("listening", r)); + const dir = await mkdtemp(join(tmpdir(), "exporter-")); + cleanup.push(() => rm(dir, { recursive: true, force: true })); + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const store = new R2ExportStore({ + endpoint: `http://127.0.0.1:${port()}`, region: "auto", accessKeyId: "k", secretAccessKey: "s", spillDir: dir, + allowInsecureEndpoint: true, + }); + const exporter = new BundleExporter(index, library as never, store, [recipient], "hub-matrix-media", pino({ level: "silent" })); + const result = await exporter.exportRoom(mapping, new AbortController().signal, "auto"); + expect(result.fileCount).toBe(0); + expect(result.totalBytes).toBe(0); + }); + + it("fails when the library payload is missing or the wrong size", async () => { + const index = new FakeIndex(); + const library = new FakeLibrary(); + index.rows = indexed([{ byteSize: PNG.byteLength + 5 }]); // declared size ≠ actual + for (const row of rows0(index)) library.files.set(`${row.sha256}-${row.safeName}`, PNG); + const { server, port } = s3Capture(); + cleanup.push(async () => await new Promise((r) => server.close(() => r()))); + await new Promise((r) => server.once("listening", r)); + const dir = await mkdtemp(join(tmpdir(), "exporter-")); + cleanup.push(() => rm(dir, { recursive: true, force: true })); + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const store = new R2ExportStore({ + endpoint: `http://127.0.0.1:${port()}`, region: "auto", accessKeyId: "k", secretAccessKey: "s", spillDir: dir, + allowInsecureEndpoint: true, + }); + const exporter = new BundleExporter(index, library as never, store, [recipient], "hub-matrix-media", pino({ level: "silent" })); + await expect(exporter.exportRoom(mapping, new AbortController().signal, "manual")).rejects.toThrow(ExportError); + }); + + it("counts attachments and auto-exports on the rolling threshold", () => { + const exporter = new BundleExporter(new FakeIndex(), new FakeLibrary() as never, fakeStore(), ["age1" + "q".repeat(58)], "b", pino({ level: "silent" })); + expect(exporter.shouldAutoExport(mapping.roomId, 2)).toBe(false); + exporter.countAttachment(mapping.roomId); + expect(exporter.shouldAutoExport(mapping.roomId, 2)).toBe(false); + exporter.countAttachment(mapping.roomId); + expect(exporter.shouldAutoExport(mapping.roomId, 2)).toBe(true); + // counter reset after firing + expect(exporter.shouldAutoExport(mapping.roomId, 2)).toBe(false); + }); + + it("rejects malformed recipients at construction", () => { + expect(() => new BundleExporter(new FakeIndex(), new FakeLibrary() as never, fakeStore(), ["not-a-key"], "b", pino({ level: "silent" }))) + .toThrow(); + }); +}); + +function rows0(index: FakeIndex): IndexedFile[] { + return index.rows; +} + +function fakeStore(): R2ExportStore { + return new R2ExportStore({ + endpoint: "https://example.invalid", region: "auto", accessKeyId: "k", secretAccessKey: "s", + }); +} + +describe("exportNotice", () => { + it("announces the object key and recipient count without secrets", () => { + const notice = exportNotice("exports/!r:x/2026.tar.zst.age", "hub-matrix-media", ["age1a", "age1b"]); + expect(notice.msgtype).toBe("m.notice"); + expect(String(notice.body)).toContain("exports/!r:x/2026.tar.zst.age"); + expect(String(notice.body)).toContain("2 age recipient"); + expect(String(notice.body)).not.toContain("AGE-SECRET-KEY"); + }); +}); diff --git a/apps/archive-bot/src/exporter.ts b/apps/archive-bot/src/exporter.ts new file mode 100644 index 0000000..dff8e4e --- /dev/null +++ b/apps/archive-bot/src/exporter.ts @@ -0,0 +1,199 @@ +import type { Logger } from "pino"; +import type { RoomMapping } from "./config.js"; +import { + newBundleId, + payloadPath, + type BundleManifest, + type BundlePayload, + type IndexedFile, + type ManifestEntry, +} from "./archive-index.js"; +import { BundleError, encryptBundleStream, parseAgeRecipients } from "./bundle.js"; +import type { LibraryPort } from "./webdav.js"; +import type { ArchiveIndexPort } from "./archive-index.js"; +import { exportObjectKey, R2ExportStore } from "./r2-export.js"; + +/** Per-attachment budget for one bundle payload in RAM (bounded, like staging). */ +export const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024; +/** Hard ceiling for one bundle's total payload bytes (100 × max attachment). */ +export const MAX_BUNDLE_BYTES = 100 * MAX_ATTACHMENT_BYTES; + +export class ExportError extends Error {} + +export interface ExportResult { + objectKey: string; + fileCount: number; + totalBytes: number; + byteSize: number; +} + +/** Admin gate: power level 50+ in the room (spec default moderator). */ +export const ADMIN_POWER_LEVEL = 50; + +export function exportNotice(objectKey: string, bucket: string, recipients: readonly string[]): Record { + return { + msgtype: "m.notice", + body: `Encrypted archive bundle uploaded: ${objectKey} (${bucket}) — ${recipients.length} age recipient(s). ` + + "Ciphertext only; open it with the admin age identity from this room.", + }; +} + +/** `!archive export` from a room admin. */ +export function isExportCommand(content: Record): boolean { + const body = typeof content.body === "string" ? content.body.trim() : ""; + return body === "!archive export" || body.startsWith("!archive export "); +} + +/** + * Phase-B exporter: builds an encrypted bundle of a room's archived media and + * uploads the ciphertext to R2. Payloads stream oCIS → tar → zstd → age; only + * the ciphertext spill file (see R2ExportStore) ever touches disk. + */ +export class BundleExporter { + private readonly recipients: string[]; + private readonly r2: R2ExportStore; + /** room → attachments archived since the last successful export */ + private readonly counters = new Map(); + private readonly libraryDir: (mapping: RoomMapping) => string[]; + + constructor( + private readonly index: ArchiveIndexPort, + private readonly library: LibraryPort, + private readonly upload: R2ExportStore, + recipients: readonly string[], + private readonly bucket: string, + private readonly logger: Logger, + private readonly now: () => Date = () => new Date(), + ) { + this.recipients = parseAgeRecipients(recipients.join(",")); + this.r2 = upload; + this.bucket = bucket; + this.libraryDir = (mapping) => ["Library", mapping.workspaceId, mapping.groupId]; + } + + /** Count one archived attachment toward the room's rolling auto-export. */ + countAttachment(roomId: string): void { + this.counters.set(roomId, (this.counters.get(roomId) ?? 0) + 1); + } + + /** True (and resets) when the room crosses ARCHIVE_EXPORT_EVERY. */ + shouldAutoExport(roomId: string, every: number): boolean { + if (every <= 0) return false; + const count = this.counters.get(roomId) ?? 0; + if (count < every) return false; + this.counters.set(roomId, 0); + return true; + } + + /** + * Build and upload one bundle. Reads every archive row for the room (payload + * streams come from oCIS WebDAV), wraps the file key for all recipients and + * PUTs the ciphertext under `exports//.tar.zst.age`. + */ + async exportRoom( + mapping: RoomMapping, + signal: AbortSignal, + reason: "manual" | "auto", + ): Promise { + const started = this.now(); + const rows = await this.index.listRoom(mapping.workspaceId, mapping.roomId); + const manifest = buildManifest(mapping, rows, started, this.recipients); + const payload = await this.collectPayloads(mapping, rows, manifest); + + const objectKey = exportObjectKey(mapping.roomId, started); + const ciphertext = await encryptBundleStream(payload, this.recipients); + await this.r2.putBundle(objectKey, ciphertext, signal); + this.logger.info( + { roomId: mapping.roomId, objectKey, files: manifest.fileCount, totalBytes: manifest.totalBytes, reason }, + "archive-export: bundle uploaded", + ); + return { objectKey, fileCount: manifest.fileCount, totalBytes: manifest.totalBytes, byteSize: manifest.totalBytes }; + } + + /** Stream every available payload out of oCIS; unavailable rows contribute metadata only. */ + private async collectPayloads(mapping: RoomMapping, rows: IndexedFile[], manifest: BundleManifest): Promise { + const payloads = new Map(); + let total = 0; + const dir = this.libraryDir(mapping); + for (const row of rows) { + if (!row.available) continue; + if (row.byteSize > MAX_ATTACHMENT_BYTES) { + throw new ExportError(`attachment ${row.eventId} exceeds the ${MAX_ATTACHMENT_BYTES} byte bundle budget`); + } + if (total + row.byteSize > MAX_BUNDLE_BYTES) { + throw new ExportError(`bundle for ${mapping.roomId} exceeds ${MAX_BUNDLE_BYTES} bytes; raise ARCHIVE_EXPORT_EVERY`); + } + const stream = await this.library.getResource(dir, `${row.sha256}-${row.safeName}`, AbortSignal.timeout(120_000)); + const bytes = await readAll(stream, row.byteSize); + if (bytes.byteLength !== row.byteSize) { + throw new ExportError(`payload ${row.eventId} size mismatch (want ${row.byteSize}, got ${bytes.byteLength})`); + } + payloads.set(payloadPath(row), bytes); + total += row.byteSize; + } + return { manifest, payloads }; + } +} + +/** Manifest header; `files` follows from the index rows. */ +export function buildManifest( + mapping: RoomMapping, + rows: IndexedFile[], + exportedAt: Date, + recipients: readonly string[], +): BundleManifest { + const files: ManifestEntry[] = rows.map((row) => ({ + eventId: row.eventId, + mxcUri: row.mxcUri, + originalName: row.originalName, + safeName: row.safeName, + mime: row.mime, + mimeClass: row.mimeClass, + byteSize: row.byteSize, + sha256: row.sha256, + sender: row.sender, + createdAt: row.createdAt, + ocisResourceId: row.ocisResourceId, + available: row.available, + })); + const available = files.filter((f) => f.available); + return { + format: "chat-media-archive-bundle", + version: 1, + roomId: mapping.roomId, + workspaceId: mapping.workspaceId, + groupId: mapping.groupId, + exportedAt: exportedAt.toISOString(), + objectKey: null, + recipients: [...recipients], + fileCount: available.length, + totalBytes: available.reduce((n, f) => n + f.byteSize, 0), + files, + }; +} + +/** Read one payload stream fully, enforcing the declared size (bounded RAM). */ +async function readAll(stream: ReadableStream, expected: number): Promise { + const chunks: Uint8Array[] = []; + let total = 0; + const reader = stream.getReader(); + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > expected) { + await reader.cancel().catch(() => undefined); + throw new ExportError(`payload longer than the indexed size (${expected})`); + } + chunks.push(value); + } + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + return out; +} + +export { BundleError, newBundleId }; diff --git a/apps/archive-bot/src/main.ts b/apps/archive-bot/src/main.ts index e584181..96d6e1b 100644 --- a/apps/archive-bot/src/main.ts +++ b/apps/archive-bot/src/main.ts @@ -1,5 +1,7 @@ import { Archiver } from "./archiver.js"; import { loadConfig } from "./config.js"; +import { BundleExporter } from "./exporter.js"; +import { R2ExportStore } from "./r2-export.js"; import { startHealthServer } from "./health.js"; import { logger } from "./logger.js"; import { MatrixClient } from "./matrix.js"; @@ -31,8 +33,37 @@ async function main(): Promise { const store = PgArchiveStore.create(config.databaseUrl); const library = new OcisWebDav(config.webdav.baseUrl, config.webdav.user, config.webdav.password); - const processor = new ArchiveProcessor(matrix, store, new Archiver(matrix, library, store), config.mappings, logger); + + // Phase-B: encrypted export bundles. Null (fully inert) unless configured. + let exporter: BundleExporter | null = null; + if (config.export !== null) { + if (config.export.r2.endpoint === "") { + logger.warn("archive-export: ARCHIVE_EXPORT_R2_ENDPOINT is not set; exports disabled"); + } else { + const upload = new R2ExportStore({ + endpoint: config.export.r2.endpoint, + region: config.export.r2.region, + accessKeyId: config.export.r2.accessKeyId, + secretAccessKey: config.export.r2.secretAccessKey, + bucket: config.export.r2.bucket, + spillDir: config.export.r2.tmpDir, + }); + exporter = new BundleExporter(store, library, upload, config.export.ageRecipients, config.export.r2.bucket, logger); + logger.info( + { recipients: config.export.ageRecipients.length, every: config.export.every, bucket: config.export.r2.bucket }, + "archive-export: encrypted bundles enabled", + ); + } + } + const processor = new ArchiveProcessor( + matrix, store, new Archiver(matrix, library, store), config.mappings, logger, + () => new Date(), exporter, exporter === null ? null : config.export, + ); await processor.init(); + // Startup lint: WARN any unencrypted room outside the mapping (policy violation). + await processor.lintEncryption(shutdown.signal).catch((err: unknown) => { + logger.warn({ err: err instanceof Error ? err.message : String(err) }, "archive-bot: encryption lint failed"); + }); const health = startHealthServer(config.healthPort, () => processor.health(), logger); try { diff --git a/apps/archive-bot/src/manifest.test.ts b/apps/archive-bot/src/manifest.test.ts new file mode 100644 index 0000000..e01590a --- /dev/null +++ b/apps/archive-bot/src/manifest.test.ts @@ -0,0 +1,84 @@ +import { describe, expect, it } from "vitest"; +import type { RoomMapping } from "./config.js"; +import { buildManifest, MAX_ATTACHMENT_BYTES } from "./exporter.js"; +import type { IndexedFile } from "./archive-index.js"; +import { payloadPath } from "./archive-index.js"; + +const mapping: RoomMapping = { + roomId: "!room:example.org", + workspaceId: "11111111-1111-4111-8111-111111111111", + groupId: "22222222-2222-4222-8222-222222222222", +}; + +function row(overrides: Partial = {}): IndexedFile { + return { + id: "0d0e0f10-1111-4222-8222-333333333333", + eventId: "$ev1", + mxcUri: "mxc://example.org/abc", + originalName: "cat photo.png", + safeName: "cat_photo.png", + mime: "image/png", + mimeClass: "image", + byteSize: 1234, + sha256: "a".repeat(64), + sender: "@alice:example.org", + createdAt: "2026-01-02T03:04:05.000Z", + ocisResourceId: "42", + available: true, + ...overrides, + }; +} + +describe("bundle manifest shape", () => { + it("carries every index row with the documented field names", () => { + const rows = [row(), row({ eventId: "$ev2", available: false, byteSize: 10, safeName: "gone.bin" })]; + const manifest = buildManifest(mapping, rows, new Date("2026-02-03T04:05:06.000Z"), ["age1testrecipient"]); + expect(manifest).toMatchObject({ + format: "chat-media-archive-bundle", + version: 1, + roomId: mapping.roomId, + workspaceId: mapping.workspaceId, + groupId: mapping.groupId, + exportedAt: "2026-02-03T04:05:06.000Z", + objectKey: null, + recipients: ["age1testrecipient"], + }); + expect(manifest.files).toHaveLength(2); + expect(manifest.files[0]).toEqual({ + eventId: "$ev1", + mxcUri: "mxc://example.org/abc", + originalName: "cat photo.png", + safeName: "cat_photo.png", + mime: "image/png", + mimeClass: "image", + byteSize: 1234, + sha256: "a".repeat(64), + sender: "@alice:example.org", + createdAt: "2026-01-02T03:04:05.000Z", + ocisResourceId: "42", + available: true, + }); + }); + + it("counts only available files in fileCount/totalBytes, but records every row", () => { + const manifest = buildManifest(mapping, [ + row({ byteSize: 100 }), + row({ eventId: "$ev2", byteSize: 40, available: false }), + ], new Date(0), []); + expect(manifest.fileCount).toBe(1); + expect(manifest.totalBytes).toBe(100); + expect(manifest.files).toHaveLength(2); + expect(manifest.files.map((f) => f.available)).toEqual([true, false]); + }); + + it("derives deterministic, collision-free tar payload paths", () => { + const one = row({ eventId: "$a", safeName: "x.png" }); + expect(payloadPath(one)).toBe("files/$a/x.png"); + // Different events never collide even with identical safe names. + expect(payloadPath(row({ eventId: "$b", safeName: "x.png" }))).not.toBe(payloadPath(one)); + }); + + it("documents the per-attachment budget constant", () => { + expect(MAX_ATTACHMENT_BYTES).toBe(20 * 1024 * 1024); + }); +}); diff --git a/apps/archive-bot/src/matrix.ts b/apps/archive-bot/src/matrix.ts index ecc3ef3..04ff741 100644 --- a/apps/archive-bot/src/matrix.ts +++ b/apps/archive-bot/src/matrix.ts @@ -38,6 +38,10 @@ export interface MatrixPort { joinRoom(roomId: string, signal: AbortSignal): Promise; downloadMedia(mxcUri: string, signal: AbortSignal): Promise; sendNotice(roomId: string, txnId: string, content: Record, signal: AbortSignal): Promise; + /** Phase-B export gate: current `m.room.power_levels` for a mapped room. */ + roomPowerLevels(roomId: string, signal: AbortSignal): Promise>; + /** Every room the account has joined (encryption lint input). */ + joinedRooms(signal: AbortSignal): Promise; } const SYNC_TIMEOUT_MS = 30_000; @@ -229,6 +233,21 @@ export class MatrixClient implements MatrixPort { ); } + async roomPowerLevels(roomId: string, signal: AbortSignal): Promise> { + const body = record(await this.request( + "GET", + `/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/state/m.room.power_levels/`, + { signal }, + )); + return body; + } + + async joinedRooms(signal: AbortSignal): Promise { + const body = record(await this.request("GET", "/_matrix/client/v3/joined_rooms", { signal })); + const joined = body.joined_rooms; + return Array.isArray(joined) ? joined.filter((r): r is string => typeof r === "string") : []; + } + /** Authenticated media endpoint only; the URL is derived from a validated mxc URI, never user-supplied HTTP. */ async downloadMedia(mxcUri: string, signal: AbortSignal): Promise { const { serverName, mediaId } = parseMxc(mxcUri); diff --git a/apps/archive-bot/src/processor.test.ts b/apps/archive-bot/src/processor.test.ts index 23e726a..ac3d6c6 100644 --- a/apps/archive-bot/src/processor.test.ts +++ b/apps/archive-bot/src/processor.test.ts @@ -8,6 +8,7 @@ import { claimKey, replyTxnId } from "./keys.js"; import type { MatrixPort, MediaDownload, SyncResult } from "./matrix.js"; import { ArchiveProcessor } from "./processor.js"; import type { ArchiveRecord, ArchiveStore, OutboxRow, RoomCursorTarget } from "./store.js"; +import type { IndexedFile } from "./archive-index.js"; import type { LibraryPort, LibraryResource } from "./webdav.js"; const mapping: RoomMapping = { @@ -66,6 +67,16 @@ class FakeMatrix implements MatrixPort { } this.notices.push({ roomId, txnId, content }); } + powerLevels: Record = { users_default: 0, users: {} }; + roomEncryptionByRoom: Map = new Map(); + joinedRoomList: string[] = [mapping.roomId]; + async roomPowerLevels(roomId: string): Promise> { + void roomId; + return this.powerLevels; + } + async joinedRooms(): Promise { + return this.joinedRoomList; + } } class FakeLibrary implements LibraryPort { @@ -83,6 +94,14 @@ class FakeLibrary implements LibraryPort { this.committed.push(path); return { path, url: `https://files/${path}`, resourceId: "rid", version: "v1" }; } + /** files keyed by `${dir}/${name}` — the export read side. */ + libraryFiles = new Map(); + async getResource(dir: string[], finalName: string): Promise> { + const key = [...dir, finalName].join("/"); + const bytes = this.libraryFiles.get(key); + if (bytes === undefined) throw new Error(`no such resource ${key}`); + return new Blob([bytes]).stream() as ReadableStream; + } } class FakeStore implements ArchiveStore { @@ -137,6 +156,23 @@ class FakeStore implements ArchiveStore { async countPending(): Promise { return [...this.outbox.values()].filter((r) => r.status === "pending").length; } + async listRoom(_ws: string, roomId: string): Promise { + return this.archive.filter((r) => r.roomId === roomId).map((r, i) => ({ + id: `file-${i + 1}`, + eventId: r.eventId, + mxcUri: "", + originalName: r.originalName, + safeName: r.safeName, + mime: r.mime, + mimeClass: r.mimeClass as IndexedFile["mimeClass"], + byteSize: r.byteSize, + sha256: r.sha256, + sender: r.senderMxid, + createdAt: r.createdAt.toISOString(), + ocisResourceId: "rid", + available: r.available, + })); + } private byId(id: string): OutboxRow { const row = [...this.outbox.values()].find((r) => r.id === id); if (row === undefined) throw new Error("no row"); @@ -274,10 +310,107 @@ describe("ArchiveProcessor", () => { expect(library.committed).toHaveLength(1); }); + it("runs the encryption lint over joined rooms and reports violations in health", async () => { + matrix.joinedRoomList = [mapping.roomId, "!stray:x"]; + matrix.roomEncryptionByRoom = new Map([["!stray:x", "unencrypted"]]); + const report = await processor.lintEncryption(signal); + expect(report.issues.map((i) => i.kind)).toEqual(["unmapped-unencrypted"]); + const health = await processor.health(); + expect(health.encryption).toMatchObject({ checked: 2, violations: ["!stray:x"], mappedUnencrypted: [mapping.roomId] }); + }); + + it("handles !archive export from a room admin: uploads ciphertext and posts the object key", async () => { + const { createServer } = await import("node:http"); + const { mkdtemp, rm } = await import("node:fs/promises"); + const { tmpdir } = await import("node:os"); + const { join } = await import("node:path"); + const age = await import("age-encryption"); + const { BundleExporter } = await import("./exporter.js"); + const { R2ExportStore } = await import("./r2-export.js"); + + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const dir = await mkdtemp(join(tmpdir(), "proc-export-")); + const puts: { key: string; body: Buffer }[] = []; + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { puts.push({ key: req.url ?? "", body: Buffer.concat(chunks) }); res.writeHead(200).end("{}"); }); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + try { + matrix.powerLevels = { users: { "@alice:example.org": 100 }, users_default: 0 }; + const upload = new R2ExportStore({ + endpoint: `http://127.0.0.1:${(server.address() as { port: number }).port}`, + region: "auto", accessKeyId: "k", secretAccessKey: "s", spillDir: dir, allowInsecureEndpoint: true, + }); + const exportConfig = { + enabled: true, every: 0, ageRecipients: [recipient], + r2: { endpoint: "http://stub", region: "auto", accessKeyId: "k", secretAccessKey: "s", bucket: "hub-matrix-media", tmpDir: dir }, + }; + const exporter = new BundleExporter(store, library, upload, [recipient], "hub-matrix-media", pino({ level: "silent" })); + const proc = new ArchiveProcessor( + matrix, store, new Archiver(matrix, library, store), [mapping], pino({ level: "silent" }), () => clock.now, + exporter, exportConfig, + ); + await proc.init(); + + // archive one attachment so the exporter has a payload to bundle + matrix.batches.push(batch("s1", [imageEvent("$a")])); + await proc.runBatch(signal); + expect(store.archive).toHaveLength(1); + const archived = store.archive[0]; + if (archived === undefined) throw new Error("no archive row"); + library.libraryFiles.set(`Library/${mapping.workspaceId}/${mapping.groupId}/${archived.sha256}-${archived.safeName}`, PNG); + + // admin command + matrix.batches.push(batch("s2", [{ + type: "m.room.message", + event_id: "$cmd", + sender: "@alice:example.org", + origin_server_ts: 1, + content: { msgtype: "m.text", body: "!archive export" }, + }])); + await proc.runBatch(signal); + + expect(puts).toHaveLength(1); + expect(puts[0]?.key).toContain("/hub-matrix-media/exports/"); + const notice = matrix.notices.at(-1); + expect(String(notice?.content.body)).toContain("exports/"); + // the uploaded bytes decrypt with the admin identity and contain the payload + const dec = new age.Decrypter(); + dec.addIdentity(identity); + const plain = await dec.decrypt(new Blob([puts[0]?.body ?? Buffer.alloc(0)]).stream() as ReadableStream); + const { readTarZstd, drainToBuffer } = await import("./bundle.js"); + const names: string[] = []; + for await (const entry of readTarZstd(plain)) { names.push(entry.name); await drainToBuffer(entry.stream); } + expect(names).toContain("manifest.json"); + expect(names).toContain(`files/$a/${archived.safeName}`); + const health = await proc.health(); + expect(health.export).toMatchObject({ enabled: true, lastObjectKey: expect.stringContaining("exports/") }); + } finally { + await rm(dir, { recursive: true, force: true }); + await new Promise((r) => server.close(() => r())); + } + }); + + it("ignores !archive export when exports are not configured or the sender is not an admin", async () => { + // not configured: no exporter attached + matrix.batches.push(batch("s1", [{ + type: "m.room.message", + event_id: "$cmd", + sender: "@alice:example.org", + origin_server_ts: 1, + content: { msgtype: "m.text", body: "!archive export" }, + }])); + await processor.runBatch(signal); + expect(matrix.notices).toHaveLength(0); + }); + it("joins invites only for mapped rooms and reports health", async () => { matrix.batches.push({ nextBatch: "s1", joined: new Map(), invited: [mapping.roomId, "!other:x"] }); await processor.runBatch(signal); expect(matrix.joined).toEqual([mapping.roomId]); - expect(await processor.health()).toEqual({ rooms: 1, cursor: "s1", pending: 0, lastProcessedAt: null }); + expect(await processor.health()).toMatchObject({ rooms: 1, cursor: "s1", pending: 0, lastProcessedAt: null }); }); }); diff --git a/apps/archive-bot/src/processor.ts b/apps/archive-bot/src/processor.ts index 77f0309..cf78d83 100644 --- a/apps/archive-bot/src/processor.ts +++ b/apps/archive-bot/src/processor.ts @@ -7,6 +7,10 @@ import { PermanentArchiveError } from "./inspector.js"; import { claimKey, replyTxnId } from "./keys.js"; import { MatrixHttpError, type MatrixPort } from "./matrix.js"; import type { ArchiveStore, RoomCursorTarget } from "./store.js"; +import type { BundleExporter } from "./exporter.js"; +import { exportNotice, isExportCommand } from "./exporter.js"; +import { senderIsRoomAdmin, toEncryptionHealth, type EncryptionLintReport, type MatrixLintPort } from "./encryption-lint.js"; +import type { ExportConfig } from "./config.js"; export interface BatchResult { /** True when the sync cursor was durably advanced past this batch. */ @@ -20,10 +24,24 @@ export interface HealthSnapshot { cursor: string | null; pending: number; lastProcessedAt: string | null; + /** Phase-B: encrypted export bundle state, null when exports are not configured. */ + export: { + enabled: boolean; + every: number; + lastExportAt: string | null; + lastObjectKey: string | null; + roomsExported: string[]; + } | null; + /** Phase-B: encryption-state lint (unmapped unencrypted rooms = violations). */ + encryption: ReturnType; } type Outcome = { blockedUntil: Date } | null; +function lintEmpty(now: Date): EncryptionLintReport { + return { checked: 0, mappedUnencrypted: [], issues: [], checkedAt: now.toISOString() }; +} + export function replyContent(safeName: string, eventId: string): Record { return { msgtype: "m.notice", @@ -63,6 +81,12 @@ export class ArchiveProcessor { private readonly byRoom: Map; private readonly cursorRooms: RoomCursorTarget[]; private readonly workspaceIds: string[]; + private readonly exporter: BundleExporter | null; + private readonly exportConfig: ExportConfig | null; + private lastExportAt: Date | null = null; + private lastObjectKey: string | null = null; + private readonly roomsExported = new Set(); + private encryptionReport: EncryptionLintReport | null = null; constructor( private readonly matrix: MatrixPort, @@ -71,10 +95,14 @@ export class ArchiveProcessor { mappings: readonly RoomMapping[], private readonly logger: Logger, private readonly now: () => Date = () => new Date(), + exporter: BundleExporter | null = null, + exportConfig: ExportConfig | null = null, ) { this.byRoom = new Map(mappings.map((m) => [m.roomId, m])); this.cursorRooms = mappings.map((m) => ({ roomId: m.roomId, workspaceId: m.workspaceId })); this.workspaceIds = [...new Set(mappings.map((m) => m.workspaceId))]; + this.exporter = exporter; + this.exportConfig = exportConfig; } async init(): Promise { @@ -87,9 +115,29 @@ export class ArchiveProcessor { cursor: this.cursor, pending: await this.store.countPending(this.workspaceIds), lastProcessedAt: this.lastProcessedAt?.toISOString() ?? null, + export: this.exportConfig === null ? null : { + enabled: this.exportConfig.enabled, + every: this.exportConfig.every, + lastExportAt: this.lastExportAt?.toISOString() ?? null, + lastObjectKey: this.lastObjectKey, + roomsExported: [...this.roomsExported], + }, + encryption: toEncryptionHealth(this.encryptionReport), }; } + /** One encryption-lint pass over every joined room (startup + periodic). */ + async lintEncryption(signal: AbortSignal): Promise { + const lintPort = this.matrix as MatrixLintPort; + if (typeof lintPort.joinedRooms !== "function") { + this.logger.warn("archive-bot: matrix port lacks joinedRooms; encryption lint skipped"); + return this.encryptionReport ?? lintEmpty(new Date()); + } + const { scanEncryption } = await import("./encryption-lint.js"); + this.encryptionReport = await scanEncryption(lintPort, [...this.byRoom.keys()], this.logger, signal, this.now()); + return this.encryptionReport; + } + async runBatch(signal: AbortSignal): Promise { const since = this.cursor; const sync = await this.matrix.sync(since, signal); @@ -166,6 +214,11 @@ export class ArchiveProcessor { encryption: RoomEncryption, signal: AbortSignal, ): Promise { + // Admin export command: an m.text message, so it is never an attachment; + // check it before classification (which would skip it as not-an-attachment). + if (isExportCommand(event.content)) { + return this.handleExportCommand(mapping, event, signal); + } const classified = classifyEvent(event, mapping, encryption, this.matrix.userId); switch (classified.kind) { case "skip": @@ -205,6 +258,7 @@ export class ArchiveProcessor { this.logger.info({ ...log, fileId: archived.file.id, sha256: archived.file.checksum.value }, "archive-bot: archived"); this.lastProcessedAt = this.now(); this.unsentReplies.set(key, archived.safeName); + await this.maybeAutoExport(mapping, signal); return this.sendReply(mapping, eventId, key, signal); } catch (err) { if (signal.aborted) throw err; @@ -220,6 +274,66 @@ export class ArchiveProcessor { } } + /** + * `!archive export` from a mapped room. Power level >= 50 required; the + * bundle is built, uploaded to R2 and the ciphertext object key is posted + * back as a notice (the key material lives only in the E2EE admin room). + */ + private async handleExportCommand( + mapping: RoomMapping, + event: MatrixEvent, + signal: AbortSignal, + ): Promise { + if (this.exporter === null || this.exportConfig === null || !this.exportConfig.enabled) { + this.logger.info({ roomId: mapping.roomId }, "archive-export: not configured; command ignored"); + return null; + } + try { + const powerLevels = await this.matrix.roomPowerLevels(mapping.roomId, signal); + if (!senderIsRoomAdmin(powerLevels, event.sender)) { + this.logger.warn({ roomId: mapping.roomId, sender: event.sender }, "archive-export: sender is not a room admin"); + return null; + } + } catch (err) { + this.logger.warn({ roomId: mapping.roomId, err: errorText(err) }, "archive-export: power-level check failed"); + return null; + } + const result = await this.exporter.exportRoom(mapping, signal, "manual"); + this.recordExport(mapping.roomId, result.objectKey); + await this.matrix.sendNotice( + mapping.roomId, + replyTxnId(`export:${result.objectKey}`), + exportNotice(result.objectKey, this.exportConfig.r2.bucket, this.exportConfig.ageRecipients), + signal, + ); + return null; + } + + private recordExport(roomId: string, objectKey: string): void { + this.lastExportAt = this.now(); + this.lastObjectKey = objectKey; + this.roomsExported.add(roomId); + } + + /** Rolling auto-export after ARCHIVE_EXPORT_EVERY archived attachments in a room. */ + private async maybeAutoExport(mapping: RoomMapping, signal: AbortSignal): Promise { + if (this.exporter === null || this.exportConfig === null || !this.exportConfig.enabled) return; + this.exporter.countAttachment(mapping.roomId); + if (!this.exporter.shouldAutoExport(mapping.roomId, this.exportConfig.every)) return; + try { + const result = await this.exporter.exportRoom(mapping.roomId === mapping.roomId ? mapping : mapping, signal, "auto"); + this.recordExport(mapping.roomId, result.objectKey); + await this.matrix.sendNotice( + mapping.roomId, + replyTxnId(`export:${result.objectKey}`), + exportNotice(result.objectKey, this.exportConfig.r2.bucket, this.exportConfig.ageRecipients), + signal, + ); + } catch (err) { + this.logger.error({ roomId: mapping.roomId, err: errorText(err) }, "archive-export: automatic export failed"); + } + } + /** Exactly one reply per committed claim; the stable txn ID makes resends idempotent server-side. */ private async sendReply(mapping: RoomMapping, eventId: string, key: string, signal: AbortSignal): Promise { const safeName = this.unsentReplies.get(key); diff --git a/apps/archive-bot/src/r2-export.test.ts b/apps/archive-bot/src/r2-export.test.ts new file mode 100644 index 0000000..2f98632 --- /dev/null +++ b/apps/archive-bot/src/r2-export.test.ts @@ -0,0 +1,113 @@ +import { createHash } from "node:crypto"; +import { mkdtemp, readdir, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createServer, type Server } from "node:http"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import * as zlib from "node:zlib"; +import { promisify } from "node:util"; +import { R2ExportStore, exportObjectKey, R2ExportError } from "./r2-export.js"; + +/** Local S3-compatible stand-in: captures PUT headers/body, never dials out. */ +function s3Stub(put: { key: string; length: number; body: Buffer }): Server { + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + put.key = req.url ?? ""; + put.body = Buffer.concat(chunks); + put.length = Number(req.headers["content-length"] ?? "-1"); + res.writeHead(200).end("{}"); + }); + }); + server.listen(0, "127.0.0.1"); + return server; +} + +let cleanup: (() => Promise)[] = []; + +afterEach(async () => { + vi.restoreAllMocks(); + await Promise.all(cleanup.map((fn) => fn().catch(() => undefined))); + cleanup = []; +}); + +function store(dir: string, port: number): R2ExportStore { + return new R2ExportStore({ + endpoint: `http://127.0.0.1:${port}`, + region: "auto", + accessKeyId: "test", + secretAccessKey: "test", + bucket: "hub-matrix-media", + spillDir: dir, + allowInsecureEndpoint: true, + }); +} + +describe("exportObjectKey", () => { + it("builds exports//.tar.zst.age", () => { + const key = exportObjectKey("!room:example.org", new Date("2026-01-02T03:04:05.678Z")); + expect(key).toBe("exports/!room:example.org/20260102T030405Z.tar.zst.age"); + }); +}); + +describe("R2ExportStore", () => { + it("rejects non-https endpoints", () => { + expect(() => new R2ExportStore({ + endpoint: "http://r2.example", region: "auto", accessKeyId: "k", secretAccessKey: "s", + })).toThrow(R2ExportError); + expect(() => new R2ExportStore({ + endpoint: "not a url", region: "auto", accessKeyId: "k", secretAccessKey: "s", + })).toThrow(R2ExportError); + }); + + it("PUTs ciphertext with an explicit content-length and cleans the spill file", async () => { + const dir = await mkdtemp(join(tmpdir(), "r2-test-")); + cleanup.push(() => rm(dir, { recursive: true, force: true })); + const put: { key: string; length: number; body: Buffer } = { key: "", length: -1, body: Buffer.alloc(0) }; + const server = s3Stub(put); + await new Promise((r) => server.once("listening", r)); + cleanup.push(async () => await new Promise((r) => server.close(() => r()))); + + const payload = Buffer.from("ciphertext-payload".repeat(100)); + await store(dir, (server.address() as { port: number }).port).putBundle( + "exports/!room:x/a.tar.zst.age", + new Blob([payload]).stream() as ReadableStream, + new AbortController().signal, + ); + + expect(put.key.startsWith("/hub-matrix-media/exports/")).toBe(true); + expect(put.length).toBe(payload.byteLength); + expect(put.body.byteLength).toBe(payload.byteLength); + expect(put.body).toEqual(payload); + // spill directory empty again — no ciphertext residue + expect(await readdir(dir)).toEqual([]); + }); + + it("leaves no spill file when the PUT fails", async () => { + const dir = await mkdtemp(join(tmpdir(), "r2-test-")); + cleanup.push(() => rm(dir, { recursive: true, force: true })); + const server = createServer((req, res) => { + void req; + res.writeHead(503).end("unavailable"); + }); + server.listen(0, "127.0.0.1"); + await new Promise((r) => server.once("listening", r)); + cleanup.push(async () => await new Promise((r) => server.close(() => r()))); + const s = store(dir, (server.address() as { port: number }).port); + await expect(s.putBundle( + "k", new Blob([Buffer.alloc(10)]).stream() as ReadableStream, new AbortController().signal, + )).rejects.toThrow(R2ExportError); + expect(await readdir(dir)).toEqual([]); + }); +}); + +describe("zstd sanity", () => { + it("node zlib round-trips zstd frames produced for bundles", async () => { + const input = Buffer.from("abcabcabc".repeat(100)); + const compressed = await promisify(zlib.zstdCompress)(input); + const restored = await promisify(zlib.zstdDecompress)(compressed); + expect(restored).toEqual(input); + expect(createHash("sha256").update(restored).digest("hex")).toBe(createHash("sha256").update(input).digest("hex")); + }); +}); diff --git a/apps/archive-bot/src/r2-export.ts b/apps/archive-bot/src/r2-export.ts new file mode 100644 index 0000000..273dedd --- /dev/null +++ b/apps/archive-bot/src/r2-export.ts @@ -0,0 +1,116 @@ +import { createWriteStream } from "node:fs"; +import { open, stat, unlink } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomBytes } from "node:crypto"; +import { once } from "node:events"; +import { createReadStream } from "node:fs"; +import { PutObjectCommand, S3Client, S3ServiceException } from "@aws-sdk/client-s3"; + +export const EXPORT_BUCKET = "hub-matrix-media"; +export const EXPORT_PREFIX = "exports"; +/** Default bounded spill directory (tmpfs-style; see README "Spill file"). */ +export const DEFAULT_SPILL_DIR = join(tmpdir(), "archive-bot-export"); + +export class R2ExportError extends Error {} + +export interface R2ExportOptions { + /** R2 S3 endpoint, e.g. `https://.r2.cloudflarestorage.com`. */ + endpoint: string; + region: string; + accessKeyId: string; + secretAccessKey: string; + bucket?: string; + /** Bounded spill dir for the ciphertext before the PUT (see README). */ + spillDir?: string; + /** Test seam ONLY: allow plain-http local endpoints. Never set in production. */ + allowInsecureEndpoint?: boolean; +} + +/** `exports//.tar.zst.age` — the documented key contract. */ +export function exportObjectKey(roomId: string, timestamp: Date): string { + const iso = timestamp.toISOString().replace(/[-:]/g, "").replace(/\.\d+Z$/, "Z"); + return `${EXPORT_PREFIX}/${roomId}/${iso}.tar.zst.age`; +} + +/** + * Minimal S3/R2 sink for export bundles. PUT-only by design: the credentials + * are the tuwunel media-provider key, so this class never lists, reads or + * deletes — it only ever creates ciphertext objects. + */ +export class R2ExportStore { + private readonly client: S3Client; + private readonly bucket: string; + private readonly spillDir: string; + + constructor(opts: R2ExportOptions) { + let endpoint: URL; + try { + endpoint = new URL(opts.endpoint); + } catch { + throw new R2ExportError(`R2 endpoint must be an absolute URL, got "${opts.endpoint}"`); + } + if (endpoint.protocol !== "https:" && opts.allowInsecureEndpoint !== true) { + throw new R2ExportError("R2 endpoint must be https"); + } + this.client = new S3Client({ + endpoint: opts.endpoint, + region: opts.region, + credentials: { accessKeyId: opts.accessKeyId, secretAccessKey: opts.secretAccessKey }, + forcePathStyle: true, + // R2 keys: no flexible checksums; keeps the PUT a single signed request. + requestChecksumCalculation: "WHEN_REQUIRED", + responseChecksumValidation: "WHEN_REQUIRED", + }); + this.bucket = opts.bucket ?? EXPORT_BUCKET; + this.spillDir = opts.spillDir ?? DEFAULT_SPILL_DIR; + } + + /** + * Upload one ciphertext bundle under `key`. + * + * The final object length only exists once zstd+age finish, so the stream is + * spilled to ONE bounded temp file (mode 0600, O_EXCL) and PUT with an + * explicit Content-Length, then unlinked in `finally`. This is the single + * documented disk touch in the export path — it is ciphertext only, never a + * payload cache; mount `ARCHIVE_EXPORT_TMP_DIR` on tmpfs if disk writes are + * unacceptable (see README). + */ + async putBundle(key: string, ciphertext: ReadableStream, signal: AbortSignal): Promise { + const spill = join(this.spillDir, `${randomBytes(16).toString("hex")}.tar.zst.age`); + const marker = await open(spill, "wx", 0o600); + await marker.close(); + const out = createWriteStream(spill, { flags: "w", mode: 0o600 }); + const reader = ciphertext.getReader(); + try { + for (;;) { + if (signal.aborted) throw new R2ExportError("export aborted"); + const { done, value } = await reader.read(); + if (done) break; + if (!out.write(Buffer.from(value.buffer, value.byteOffset, value.byteLength))) await once(out, "drain"); + } + await new Promise((resolve, reject) => out.end((err?: Error | null) => (err === null || err === undefined ? resolve() : reject(err)))); + const size = (await stat(spill)).size; + await this.client.send(new PutObjectCommand({ + Bucket: this.bucket, + Key: key, + Body: createReadStream(spill), + ContentLength: size, + ContentType: "application/octet-stream", + }), { abortSignal: signal }); + } catch (err) { + throw err instanceof R2ExportError ? err : toR2Error(err); + } finally { + await reader.cancel().catch(() => undefined); + await new Promise((resolve) => out.close(() => resolve())).catch(() => undefined); + await unlink(spill).catch(() => undefined); + } + } +} + +function toR2Error(err: unknown): R2ExportError { + if (err instanceof S3ServiceException) { + return new R2ExportError(`R2 upload failed: ${err.$metadata.httpStatusCode ?? "HTTP"} ${err.name}`); + } + return err instanceof Error ? new R2ExportError(err.message) : new R2ExportError(String(err)); +} diff --git a/apps/archive-bot/src/restore.test.ts b/apps/archive-bot/src/restore.test.ts new file mode 100644 index 0000000..d6a9b69 --- /dev/null +++ b/apps/archive-bot/src/restore.test.ts @@ -0,0 +1,158 @@ +import { spawnSync, type SpawnSyncOptionsWithStringEncoding } from "node:child_process"; +import { createHash } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import * as age from "age-encryption"; +import { createRequire } from "node:module"; +import { afterAll, describe, expect, it } from "vitest"; +import { parseIdentityFile } from "./restore.js"; +import { encryptBundleStream, drainToBuffer } from "./bundle.js"; +import type { BundleManifest } from "./archive-index.js"; + +// The CLI runs as a subprocess so its process.exitCode handling is exercised. +const CLI = join(__dirname, "restore.ts"); +const RUN_OPTS: SpawnSyncOptionsWithStringEncoding = { encoding: "utf8", timeout: 60_000 }; + +// tsx does not export its CLI path; walk up from the resolved package.json +// (pnpm symlinks point into the store, so dist/ sits beside package.json). +const RUNNER = join( + dirname(createRequire(import.meta.url).resolve("tsx/package.json")), + "dist", "cli.mjs", +); + +const workDirs: string[] = []; +afterAll(async () => { + await Promise.all(workDirs.map((d) => rm(d, { recursive: true, force: true }).catch(() => undefined))); +}); + +describe("parseIdentityFile", () => { + it("reads one AGE-SECRET-KEY per line, ignoring comments and blanks", () => { + const identity = "AGE-SECRET-KEY-1TESTKEYVALUENOTAREALKEY0000000"; + expect(parseIdentityFile(`# admin key\n\n${identity}\n`)).toEqual([identity]); + expect(parseIdentityFile("")).toEqual([]); + }); +}); + +describe("restore CLI (end-to-end)", () => { + it("round-trips a real bundle: encrypt → decrypt → verify → write", async () => { + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + + const png = Uint8Array.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 4, 5, 6]); + const pdf = new TextEncoder().encode("%PDF-1.7\nrestore me"); + const manifest: BundleManifest = { + format: "chat-media-archive-bundle", + version: 1, + roomId: "!room:example.org", + workspaceId: "w", + groupId: "g", + exportedAt: "2026-01-01T00:00:00.000Z", + objectKey: null, + recipients: [recipient], + fileCount: 2, + totalBytes: png.byteLength + pdf.byteLength, + files: [ + { + eventId: "$e1", + mxcUri: "mxc://example.org/a", + originalName: "cat.png", + safeName: "cat.png", + mime: "image/png", + mimeClass: "image", + byteSize: png.byteLength, + sha256: createHash("sha256").update(png).digest("hex"), + sender: "@alice:example.org", + createdAt: "2026-01-02T00:00:00.000Z", + ocisResourceId: "ocis-1", + available: true, + }, + { + eventId: "$e2", + mxcUri: "mxc://example.org/b", + originalName: "doc.pdf", + safeName: "doc.pdf", + mime: "application/pdf", + mimeClass: "document", + byteSize: pdf.byteLength, + sha256: createHash("sha256").update(pdf).digest("hex"), + sender: "@bob:example.org", + createdAt: "2026-01-03T00:00:00.000Z", + ocisResourceId: "ocis-2", + available: true, + }, + ], + }; + + const dir = await mkdtemp(join(tmpdir(), "restore-cli-")); + workDirs.push(dir); + const bundlePath = join(dir, "bundle.tar.zst.age"); + const identityPath = join(dir, "identity.txt"); + const outDir = join(dir, "out"); + await writeFile(identityPath, `# admin\n${identity}\n`); + + const ciphertext = await drainToBuffer(await encryptBundleStream( + { manifest, payloads: new Map([["files/$e1/cat.png", png], ["files/$e2/doc.pdf", pdf]]) }, + [recipient], + )); + await writeFile(bundlePath, ciphertext); + // sanity: the bundle is ciphertext, not a readable tar + expect(Buffer.from(ciphertext.subarray(257, 262)).toString("latin1")).not.toBe("ustar"); + + const run = spawnSync(process.execPath, [RUNNER, CLI, "--bundle", bundlePath, "--identity", identityPath, "--out", outDir], RUN_OPTS); + expect(run.stderr).toBe(""); + expect(run.stdout).toContain("2/2 files verified"); + expect(run.status).toBe(0); + + const restoredPng = await readFile(join(outDir, "files", "$e1", "cat.png")); + expect(new Uint8Array(restoredPng)).toEqual(png); + const restoredPdf = await readFile(join(outDir, "files", "$e2", "doc.pdf")); + expect(new Uint8Array(restoredPdf)).toEqual(pdf); + const manifestOnDisk = JSON.parse(await readFile(join(outDir, "manifest.json"), "utf8")) as BundleManifest; + expect(manifestOnDisk.files).toHaveLength(2); + }, 90_000); + + it("fails (non-zero exit) when a payload's sha256 does not match the manifest", async () => { + const identity = await age.generateIdentity(); + const recipient = await age.identityToRecipient(identity); + const good = new TextEncoder().encode("good bytes"); + const manifest: BundleManifest = { + format: "chat-media-archive-bundle", + version: 1, + roomId: "!r:x", + workspaceId: "w", + groupId: "g", + exportedAt: new Date(0).toISOString(), + objectKey: null, + recipients: [], + fileCount: 1, + totalBytes: good.byteLength, + files: [{ + eventId: "$e1", + mxcUri: "", + originalName: "a.txt", + safeName: "a.txt", + mime: "text/plain", + mimeClass: "document", + byteSize: good.byteLength, + sha256: createHash("sha256").update("tampered!").digest("hex"), + sender: "@a:x", + createdAt: new Date(0).toISOString(), + ocisResourceId: "r", + available: true, + }], + }; + const dir = await mkdtemp(join(tmpdir(), "restore-bad-")); + workDirs.push(dir); + const bundlePath = join(dir, "b.tar.zst.age"); + const identityPath = join(dir, "id.txt"); + await writeFile(identityPath, identity); + await writeFile(bundlePath, await drainToBuffer(await encryptBundleStream( + { manifest, payloads: new Map([["files/$e1/a.txt", good]]) }, + [recipient], + ))); + const run = spawnSync(process.execPath, [RUNNER, CLI, "--bundle", bundlePath, "--identity", identityPath, "--out", join(dir, "out")], RUN_OPTS); + expect(run.status).toBe(1); + expect(run.stdout).toContain("failed"); + }, 90_000); +}); diff --git a/apps/archive-bot/src/restore.ts b/apps/archive-bot/src/restore.ts new file mode 100644 index 0000000..59584a7 --- /dev/null +++ b/apps/archive-bot/src/restore.ts @@ -0,0 +1,105 @@ +#!/usr/bin/env node +/** + * Standalone phase-B restore tool. + * + * node dist/restore.js --bundle --identity --out + * + * Decrypts an archive bundle with an age identity (AGE-SECRET-KEY-1…, one per + * line; the private key never touches the archive bot), untars it, verifies + * every file's sha256 against manifest.json and writes the verified tree. + * Exit code 0 only when every file matched. + */ +import { createHash } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { dirname, join, resolve } from "node:path"; +import process from "node:process"; +import { Readable } from "node:stream"; +import { ageDecryptStream, drainToBuffer, readTarZstd } from "./bundle.js"; +import { MANIFEST_PATH, type BundleManifest } from "./archive-index.js"; + +interface Args { + bundle: string; + identity: string; + out: string; +} + +function parseArgs(argv: string[]): Args { + const args: Partial = {}; + for (let i = 2; i < argv.length; i += 2) { + const flag = argv[i]; + const value = argv[i + 1]; + if (value === undefined) throw new Error(`missing value for ${flag}`); + if (flag === "--bundle") args.bundle = value; + else if (flag === "--identity") args.identity = value; + else if (flag === "--out") args.out = value; + else throw new Error(`unknown flag ${flag}`); + } + if (args.bundle === undefined) throw new Error("--bundle is required"); + if (args.identity === undefined) throw new Error("--identity is required"); + args.out ??= "restored-archive"; + return args as Args; +} + +/** Parse identities: `#` comments and blank lines ignored, one key per line. */ +export function parseIdentityFile(text: string): string[] { + return text + .split("\n") + .map((line) => line.trim()) + .filter((line) => line !== "" && !line.startsWith("#")); +} + +async function main(): Promise { + const args = parseArgs(process.argv); + const identities = parseIdentityFile(await readFile(args.identity, "utf8")); + if (identities.length === 0) throw new Error("identity file contains no AGE-SECRET-KEY-1… keys"); + + const ciphertext = Readable.toWeb(Readable.from(await readFile(args.bundle))) as ReadableStream; + const plain = await ageDecryptStream(ciphertext, identities); + const root = resolve(args.out); + + let manifest: BundleManifest | null = null; + let verified = 0; + let failures = 0; + let unexpected = 0; + const manifestPathFor = (entry: { eventId: string; safeName: string }): string => + `${"files"}/${entry.eventId}/${entry.safeName}`; + for await (const entry of readTarZstd(plain)) { + if (entry.name === MANIFEST_PATH) { + const bytes = await drainToBuffer(entry.stream); + manifest = JSON.parse(new TextDecoder().decode(bytes)) as BundleManifest; + await mkdir(join(root, MANIFEST_PATH, ".."), { recursive: true }); + await writeFile(join(root, MANIFEST_PATH), bytes, { mode: 0o600 }); + continue; + } + if (manifest === null) { + // manifest.json must precede payloads in the bundle; fail loudly otherwise. + throw new Error(`payload ${entry.name} appeared before manifest.json`); + } + if (entry.name === MANIFEST_PATH) continue; // handled above + const bytes = await drainToBuffer(entry.stream); + const digest = createHash("sha256").update(bytes).digest("hex"); + const record = manifest.files.find((f) => manifestPathFor(f) === entry.name); + const target = join(root, entry.name); + await mkdir(dirname(target), { recursive: true }); + await writeFile(target, bytes, { mode: 0o600 }); + if (record === undefined) { + unexpected += 1; + console.error(`restore: ${entry.name} is not in the manifest`); + } else if (record.sha256 === digest) { + verified += 1; + } else { + failures += 1; + console.error(`restore: sha256 mismatch for ${entry.name}: expected ${record.sha256}, got ${digest}`); + } + } + if (manifest === null) throw new Error("bundle contains no manifest.json"); + const expected = manifest.files.filter((f) => f.available).length; + console.log(`restore: ${verified}/${expected} files verified${failures > 0 ? `, ${failures} failed` : ""}${unexpected > 0 ? `, ${unexpected} unexpected` : ""}`); + console.log(`restore: room ${manifest.roomId} exported ${manifest.exportedAt}`); + if (verified !== expected || failures > 0 || unexpected > 0) process.exitCode = 1; +} + +void main().catch((error: unknown) => { + console.error(`restore: ${error instanceof Error ? error.message : String(error)}`); + process.exitCode = 1; +}); diff --git a/apps/archive-bot/src/store.ts b/apps/archive-bot/src/store.ts index 55dd1ba..c1e5801 100644 --- a/apps/archive-bot/src/store.ts +++ b/apps/archive-bot/src/store.ts @@ -1,4 +1,5 @@ import { Pool, type PoolClient } from "pg"; +import type { ArchiveIndexPort, IndexedFile } from "./archive-index.js"; export type OutboxStatus = "pending" | "done" | "failed"; @@ -53,6 +54,11 @@ export interface ArchiveStore { */ markRedacted(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise; countPending(workspaceIds: readonly string[]): Promise; + /** + * Phase-B export manifest source (ArchiveIndexPort): every archive row for + * one room, oldest first. Read-only; no payload bytes leave this interface. + */ + listRoom(workspaceId: string, roomId: string): Promise; } interface OutboxDbRow { @@ -68,9 +74,44 @@ function toOutbox(row: OutboxDbRow): OutboxRow { } const OUTBOX_COLUMNS = "id, status, attempts, last_error, updated_at"; + +interface RoomArchiveDbRow { + id: string; + event_id: string; + sender_mxid: string; + original_name: string; + safe_name: string; + mime: string; + mime_class: string; + byte_size: string | number; + sha256: string; + ocis_resource_id: string; + created_at: Date; + available: boolean; + /** Extracted from the stored event JSON; NULL for pre-archive rows without one. */ + mxc_uri: string | null; +} + +function toIndexedFile(row: RoomArchiveDbRow): IndexedFile { + return { + id: row.id, + eventId: row.event_id, + mxcUri: row.mxc_uri ?? "", + originalName: row.original_name, + safeName: row.safe_name, + mime: row.mime, + mimeClass: row.mime_class as IndexedFile["mimeClass"], + byteSize: typeof row.byte_size === "string" ? Number(row.byte_size) : row.byte_size, + sha256: row.sha256, + sender: row.sender_mxid, + createdAt: row.created_at.toISOString(), + ocisResourceId: row.ocis_resource_id, + available: row.available, + }; +} const LAST_ERROR_MAX = 1000; -export class PgArchiveStore implements ArchiveStore { +export class PgArchiveStore implements ArchiveStore, ArchiveIndexPort { constructor(private readonly pool: Pool) {} /** Single service connection; tenant context per transaction, like apps/api withTenant. */ @@ -222,6 +263,21 @@ export class PgArchiveStore implements ArchiveStore { return total; } + async listRoom(workspaceId: string, roomId: string): Promise { + return this.withTenant(workspaceId, async (c) => { + const res = await c.query( + `SELECT id, event_id, sender_mxid, original_name, safe_name, mime, mime_class, + byte_size, sha256, ocis_resource_id, created_at, available, + matrix_event_json->'content'->>'url' AS mxc_uri + FROM chat_media_archive + WHERE workspace_id = $1 AND room_id = $2 + ORDER BY created_at, id`, + [workspaceId, roomId], + ); + return res.rows.map(toIndexedFile); + }); + } + private async withTenant(workspaceId: string, fn: (client: PoolClient) => Promise): Promise { const client = await this.pool.connect(); try { diff --git a/apps/archive-bot/src/webdav.ts b/apps/archive-bot/src/webdav.ts index 807d617..df0c98a 100644 --- a/apps/archive-bot/src/webdav.ts +++ b/apps/archive-bot/src/webdav.ts @@ -11,6 +11,11 @@ export interface LibraryPort { putStaging(dir: string[], stagingName: string, body: ReadableStream, signal: AbortSignal): Promise; /** Move staging to its content-addressed final name; an existing final resource is reused. */ commit(dir: string[], stagingName: string, finalName: string, signal: AbortSignal): Promise; + /** + * Phase-B export read side: stream one committed library resource back out + * of oCIS. Used only to build encrypted export bundles; never a cache. + */ + getResource(dir: string[], finalName: string, signal: AbortSignal): Promise>; } export class WebDavError extends Error { @@ -108,6 +113,13 @@ export class OcisWebDav implements LibraryPort { return { path: final.join("/"), url: destination, ...props }; } + async getResource(dir: string[], finalName: string, signal: AbortSignal): Promise> { + const res = await this.send("GET", [...dir, finalName], signal); + if (res.status === 404) throw new WebDavError(404, `WebDAV GET missing resource: ${finalName}`); + if (!res.ok || res.body === null) throw new WebDavError(res.status, `WebDAV GET failed: ${res.status}`); + return res.body; + } + private async ensureDir(segments: string[], signal: AbortSignal): Promise { for (let depth = 1; depth <= segments.length; depth += 1) { const dir = segments.slice(0, depth);