From 2cb77ae6b4929869cd85639fc1539ae6e11b4b31 Mon Sep 17 00:00:00 2001 From: glass beadian Date: Sun, 11 Oct 2026 16:12:43 +0000 Subject: [PATCH] Harden archive bot for production; exempt container egress in edge firewall - archive_sync_state keyed by room per migration 029 (drop synthetic id cursor) - putStaging buffers the stream and sends content-length (oCIS rejects chunked) - mimeClass passthrough from sniffed payload - edge-firewall: RETURN docker-bridge ingress before the CF-range catch-all so container egress (Matrix/R2/platform) survives the 80/443 source filter - files scaffold: keep the ocdav data-gateway hop on-loopback --- apps/archive-bot/src/archiver.ts | 1 + apps/archive-bot/src/main.ts | 2 +- apps/archive-bot/src/processor.test.ts | 9 +- apps/archive-bot/src/processor.ts | 8 +- apps/archive-bot/src/store.ts | 55 +++++----- apps/archive-bot/src/webdav.test.ts | 133 +++++++++++++++++++++++- apps/archive-bot/src/webdav.ts | 30 +++++- infra/occupymedia/edge/edge-firewall.sh | 8 ++ infra/occupymedia/files/compose.yml | 3 + 9 files changed, 211 insertions(+), 38 deletions(-) diff --git a/apps/archive-bot/src/archiver.ts b/apps/archive-bot/src/archiver.ts index 28ca80d..da7b601 100644 --- a/apps/archive-bot/src/archiver.ts +++ b/apps/archive-bot/src/archiver.ts @@ -76,6 +76,7 @@ export class Archiver { origin: { kind: "matrix", mxcUri: candidate.mxcUri }, name: safeName, mimeType: payload.sniffed.mime, + mimeClass: payload.sniffed.mimeClass, sizeBytes: payload.byteSize, checksum: { algorithm: "sha256", value: payload.sha256 }, createdAt: createdAt.toISOString(), diff --git a/apps/archive-bot/src/main.ts b/apps/archive-bot/src/main.ts index 617da53..e584181 100644 --- a/apps/archive-bot/src/main.ts +++ b/apps/archive-bot/src/main.ts @@ -29,7 +29,7 @@ async function main(): Promise { await matrix.login(shutdown.signal); logger.info({ userId: matrix.userId, rooms: config.mappings.length }, "archive-bot: logged in"); - const store = PgArchiveStore.create(config.databaseUrl, matrix.userId); + 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); await processor.init(); diff --git a/apps/archive-bot/src/processor.test.ts b/apps/archive-bot/src/processor.test.ts index cb98182..7d7cc0f 100644 --- a/apps/archive-bot/src/processor.test.ts +++ b/apps/archive-bot/src/processor.test.ts @@ -7,7 +7,7 @@ import type { MatrixEvent, RoomEncryption } from "./events.js"; import { claimKey, replyTxnId } from "./keys.js"; import type { MatrixPort, MediaDownload, SyncResult } from "./matrix.js"; import { ArchiveProcessor } from "./processor.js"; -import type { ArchiveRecord, ArchiveStore, OutboxRow } from "./store.js"; +import type { ArchiveRecord, ArchiveStore, OutboxRow, RoomCursorTarget } from "./store.js"; import type { LibraryPort, LibraryResource } from "./webdav.js"; const mapping: RoomMapping = { @@ -87,12 +87,14 @@ class FakeLibrary implements LibraryPort { class FakeStore implements ArchiveStore { cursor: string | null = null; + roomsSaved = new Set(); outbox = new Map(); archive: (ArchiveRecord & { available: boolean })[] = []; - async loadCursor(): Promise { + async loadCursor(_rooms: readonly RoomCursorTarget[]): Promise { return this.cursor; } - async saveCursor(_ws: readonly string[], cursor: string): Promise { + async saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise { + for (const room of rooms) this.roomsSaved.add(room.roomId); this.cursor = cursor; } async claim(_ws: string, _room: string, _event: string, key: string): Promise { @@ -165,6 +167,7 @@ describe("ArchiveProcessor", () => { expect(await processor.runBatch(signal)).toEqual({ advanced: true, retryAt: null }); expect(store.cursor).toBe("s1"); + expect([...store.roomsSaved]).toEqual([mapping.roomId]); expect(store.archive).toHaveLength(1); const row = store.archive[0]; expect(row).toMatchObject({ mime: "image/png", mimeClass: "image", safeName: "cat.png", originalName: "../cat", byteSize: PNG.byteLength }); diff --git a/apps/archive-bot/src/processor.ts b/apps/archive-bot/src/processor.ts index a833767..77f0309 100644 --- a/apps/archive-bot/src/processor.ts +++ b/apps/archive-bot/src/processor.ts @@ -6,7 +6,7 @@ import { classifyEvent, type AttachmentWork, type MatrixEvent, type RoomEncrypti import { PermanentArchiveError } from "./inspector.js"; import { claimKey, replyTxnId } from "./keys.js"; import { MatrixHttpError, type MatrixPort } from "./matrix.js"; -import type { ArchiveStore } from "./store.js"; +import type { ArchiveStore, RoomCursorTarget } from "./store.js"; export interface BatchResult { /** True when the sync cursor was durably advanced past this batch. */ @@ -61,6 +61,7 @@ export class ArchiveProcessor { /** Committed claims whose single reply has not been delivered yet (key → safe name). */ private readonly unsentReplies = new Map(); private readonly byRoom: Map; + private readonly cursorRooms: RoomCursorTarget[]; private readonly workspaceIds: string[]; constructor( @@ -72,11 +73,12 @@ export class ArchiveProcessor { private readonly now: () => Date = () => new Date(), ) { 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))]; } async init(): Promise { - this.cursor = await this.store.loadCursor(this.workspaceIds); + this.cursor = await this.store.loadCursor(this.cursorRooms); } async health(): Promise { @@ -115,7 +117,7 @@ export class ArchiveProcessor { } if (retryAt !== null) return { advanced: false, retryAt }; - await this.store.saveCursor(this.workspaceIds, sync.nextBatch); + await this.store.saveCursor(this.cursorRooms, sync.nextBatch); this.cursor = sync.nextBatch; return { advanced: true, retryAt: null }; } diff --git a/apps/archive-bot/src/store.ts b/apps/archive-bot/src/store.ts index b40b190..c26fa7f 100644 --- a/apps/archive-bot/src/store.ts +++ b/apps/archive-bot/src/store.ts @@ -28,10 +28,20 @@ export interface ArchiveRecord { createdAt: Date; } +/** + * Cursor rows are keyed by room (029: archive_sync_state(room_id PRIMARY KEY, + * workspace_id NOT NULL, ...)), so the store needs the room → tenant pin the + * config mappings already carry. RoomMapping satisfies this structurally. + */ +export interface RoomCursorTarget { + roomId: string; + workspaceId: string; +} + /** Durable state the processor depends on; faked in tests. */ export interface ArchiveStore { - loadCursor(workspaceIds: readonly string[]): Promise; - saveCursor(workspaceIds: readonly string[], cursor: string): Promise; + loadCursor(rooms: readonly RoomCursorTarget[]): Promise; + saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise; claim(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise; /** Index row + outbox done in one transaction. Returns the archive row id and indexed time. */ commitArchived(record: ArchiveRecord, outboxId: string): Promise<{ id: string; indexedAt: Date }>; @@ -57,14 +67,11 @@ const OUTBOX_COLUMNS = "id, status, attempts, last_error, updated_at"; const LAST_ERROR_MAX = 1000; export class PgArchiveStore implements ArchiveStore { - constructor( - private readonly pool: Pool, - private readonly cursorId: string, - ) {} + constructor(private readonly pool: Pool) {} /** Single service connection; tenant context per transaction, like apps/api withTenant. */ - static create(databaseUrl: string, cursorId: string): PgArchiveStore { - return new PgArchiveStore(new Pool({ connectionString: databaseUrl, max: 1 }), cursorId); + static create(databaseUrl: string): PgArchiveStore { + return new PgArchiveStore(new Pool({ connectionString: databaseUrl, max: 1 })); } async close(): Promise { @@ -72,17 +79,17 @@ export class PgArchiveStore implements ArchiveStore { } /** - * The sync stream is global but rows are tenant-scoped, so each mapped - * workspace holds a copy. A partial write leaves copies disagreeing: resume - * from the least recently written (oldest) one — replay is idempotent. + * The sync stream is global but cursor rows are per room, so each mapped room + * holds a copy. A partial write leaves copies disagreeing: resume from the + * least recently written (oldest) one — replay is idempotent. */ - async loadCursor(workspaceIds: readonly string[]): Promise { + async loadCursor(rooms: readonly RoomCursorTarget[]): Promise { let oldest: { cursor: string; at: Date } | null = null; - for (const workspaceId of workspaceIds) { - const row = await this.withTenant(workspaceId, async (c) => { + for (const room of rooms) { + const row = await this.withTenant(room.workspaceId, async (c) => { const res = await c.query<{ sync_cursor: string | null; updated_at: Date }>( - "SELECT sync_cursor, updated_at FROM archive_sync_state WHERE id = $1 AND workspace_id = $2", - [this.stateId(workspaceId), workspaceId], + "SELECT sync_cursor, updated_at FROM archive_sync_state WHERE room_id = $1 AND workspace_id = $2", + [room.roomId, room.workspaceId], ); return res.rows[0]; }); @@ -92,13 +99,13 @@ export class PgArchiveStore implements ArchiveStore { return oldest?.cursor ?? null; } - async saveCursor(workspaceIds: readonly string[], cursor: string): Promise { - for (const workspaceId of workspaceIds) { - await this.withTenant(workspaceId, (c) => c.query( - `INSERT INTO archive_sync_state (id, workspace_id, sync_cursor, updated_at) + async saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise { + for (const room of rooms) { + await this.withTenant(room.workspaceId, (c) => c.query( + `INSERT INTO archive_sync_state (room_id, workspace_id, sync_cursor, updated_at) VALUES ($1, $2, $3, now()) - ON CONFLICT (id) DO UPDATE SET sync_cursor = EXCLUDED.sync_cursor, updated_at = now()`, - [this.stateId(workspaceId), workspaceId, cursor], + ON CONFLICT (room_id) DO UPDATE SET sync_cursor = EXCLUDED.sync_cursor, updated_at = now()`, + [room.roomId, room.workspaceId, cursor], )); } } @@ -201,10 +208,6 @@ export class PgArchiveStore implements ArchiveStore { return total; } - private stateId(workspaceId: string): string { - return `${this.cursorId}:${workspaceId}`; - } - 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.test.ts b/apps/archive-bot/src/webdav.test.ts index cb65d1d..399f42c 100644 --- a/apps/archive-bot/src/webdav.test.ts +++ b/apps/archive-bot/src/webdav.test.ts @@ -1,5 +1,53 @@ -import { describe, expect, it } from "vitest"; -import { davPath, parsePropfind } from "./webdav.js"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { davPath, OcisWebDav, parsePropfind, WebDavError } from "./webdav.js"; + +let realFetch: typeof globalThis.fetch | undefined; + +/** Swap in a fake fetch, remembering the real one so afterEach can put it back. */ +function stubFetch(impl: (input: unknown, init?: RequestInit) => Promise): void { + realFetch ??= globalThis.fetch; + globalThis.fetch = impl as typeof globalThis.fetch; +} + +afterEach(() => { + vi.restoreAllMocks(); + if (realFetch !== undefined) { + globalThis.fetch = realFetch; + realFetch = undefined; + } +}); + +/** + * oCIS refuses chunked PUTs ('error getting the content length' → HTTP 400), so the + * client must hand fetch a fully-buffered body that carries Content-Length. + * Returns every request the client issued, for per-method assertions. + */ +function captureFetch(status = 201): { calls: { method: string; headers: Headers; body?: unknown }[] } { + const calls: { method: string; headers: Headers; body?: unknown }[] = []; + stubFetch(async (_input: unknown, init?: RequestInit) => { + const method = String(init?.method ?? "GET"); + calls.push({ method, headers: new Headers(init?.headers ?? {}), body: init?.body }); + // ensureDir's MKCOL only accepts 201/405; other methods get the scenario status. + return new Response(null, { status: method === "MKCOL" ? 201 : status }); + }); + return { calls }; +} + +/** Narrow an unknown fetch body to a Buffer without casts. */ +function bufferBody(value: unknown): Buffer { + if (!(value instanceof Buffer)) throw new Error(`expected a Buffer PUT body, got ${String(value)}`); + return value; +} + +/** A one-chunk stream, the shape putStaging receives from the inspector. */ +function streamOf(chunk: Uint8Array): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue(chunk); + controller.close(); + }, + }); +} describe("davPath", () => { it("encodes every segment under the service user's files root", () => { @@ -19,3 +67,84 @@ describe("parsePropfind", () => { expect(parsePropfind("")).toBeNull(); }); }); + +describe("OcisWebDav.putStaging", () => { + const client = () => new OcisWebDav(new URL("https://ocis.example/"), "svc", "hunter2"); + + it("PUTs a buffered body with an explicit content-length matching the payload bytes", async () => { + const { calls } = captureFetch(201); + const payload = Buffer.from([0x25, 0x26, 0x27, 0x28, 0x29]); + await client().putStaging(["Library", "ws", "grp"], "txn.part", streamOf(payload), new AbortController().signal); + const put = calls.find((call) => call.method === "PUT"); + expect(put).toBeDefined(); + if (put === undefined) return; + const headers = new Headers(put.headers); + expect(headers.get("content-length")).toBe(String(payload.byteLength)); + expect(Number(headers.get("content-length"))).toBe(payload.byteLength); + // Not chunked: a Buffer body, never a stream handed straight to fetch. + const body = bufferBody(put.body); + expect(body).toBeInstanceOf(Buffer); + expect(body).not.toBeInstanceOf(ReadableStream); + expect(body.byteLength).toBe(payload.byteLength); + expect(headers.get("content-type")).toBe("application/octet-stream"); + expect(headers.get("authorization")).toMatch(/^Basic /); + }); + + it("keeps the payload bytes intact across multi-chunk streams", async () => { + const { calls } = captureFetch(204); + const multiChunk = new ReadableStream({ + start(controller) { + controller.enqueue(Buffer.from("hello ", "utf8")); + controller.enqueue(Buffer.from("staging", "utf8")); + controller.close(); + }, + }); + await client().putStaging(["Library"], "txn.part", multiChunk, new AbortController().signal); + const put = calls.find((call) => call.method === "PUT"); + expect(put).toBeDefined(); + if (put === undefined) return; + expect(Buffer.concat([bufferBody(put.body)]).toString("utf8")).toBe("hello staging"); + expect(new Headers(put.headers).get("content-length")).toBe("13"); + }); + + it("PUTs zero bytes as content-length: 0", async () => { + const { calls } = captureFetch(201); + await client().putStaging(["Library"], "txn.part", streamOf(Buffer.alloc(0)), new AbortController().signal); + const put = calls.find((call) => call.method === "PUT"); + expect(put).toBeDefined(); + if (put === undefined) return; + expect(new Headers(put.headers).get("content-length")).toBe("0"); + expect(bufferBody(put.body).byteLength).toBe(0); + }); + + it("throws WebDavError(413) without issuing a PUT when the stream exceeds 64 MiB", async () => { + const { calls } = captureFetch(201); + const miB = 1024 * 1024; + const header = Buffer.alloc(4); + const oversized = new ReadableStream({ + async start(controller) { + controller.enqueue(header); + // 16 x 4 MiB = 64 MiB, so this final enqueue crosses the cap mid-stream. + for (let i = 0; i < 16; i += 1) controller.enqueue(Buffer.alloc(4 * miB)); + controller.enqueue(header); + controller.close(); + }, + }); + const failure = await client().putStaging(["Library"], "txn.part", oversized, new AbortController().signal).then( + () => undefined, + (error: unknown) => error, + ); + expect(failure).toBeInstanceOf(WebDavError); + if (!(failure instanceof WebDavError)) throw new Error("expected a WebDavError from the oversized stream"); + expect(failure.status).toBe(413); + expect(failure.message).toBe("staging payload exceeds 64 MiB"); + expect(calls.filter((call) => call.method === "PUT")).toEqual([]); + }); + + it("rejects with WebDavError carrying the response status on a non-201/204 PUT", async () => { + const { calls } = captureFetch(400); + await expect(client().putStaging(["Library"], "txn.part", streamOf(Buffer.from("x")), new AbortController().signal)) + .rejects.toMatchObject({ status: 400, message: "WebDAV PUT failed: 400" }); + expect(calls.some((call) => call.method === "PUT")).toBe(true); + }); +}); diff --git a/apps/archive-bot/src/webdav.ts b/apps/archive-bot/src/webdav.ts index bbb842a..807d617 100644 --- a/apps/archive-bot/src/webdav.ts +++ b/apps/archive-bot/src/webdav.ts @@ -24,6 +24,13 @@ export class WebDavError extends Error { const STAGING_DIR = ".archive-staging"; +/** + * Backstop for a runaway stream: the homeserver already caps uploads at ~20 MB, + * so anything past this is a desync (fibbed content-length, wrong MIME sniff) + * rather than a legitimate attachment. Checked before the PUT leaves the host. + */ +const STAGING_MAX_BYTES = 64 * 1024 * 1024; + /** Percent-encode each segment; segments themselves are already normalised safe names/UUIDs. */ export function davPath(user: string, segments: readonly string[]): string { return `/remote.php/dav/files/${[user, ...segments].map(encodeURIComponent).join("/")}`; @@ -55,9 +62,26 @@ export class OcisWebDav implements LibraryPort { async putStaging(dir: string[], stagingName: string, body: ReadableStream, signal: AbortSignal): Promise { await this.ensureDir([...dir, STAGING_DIR], signal); + // oCIS requires Content-Length on PUT bodies; undici only sets it for + // non-stream bodies, so a streamed request goes out chunked and oCIS + // answers 400 ('error getting the content length'). Drain the stream + // into one Buffer first — a bounded wait, attachments are <= 20 MB. + const chunks: Uint8Array[] = []; + let total = 0; + for await (const chunk of body) { + total += chunk.byteLength; + if (total > STAGING_MAX_BYTES) throw new WebDavError(413, "staging payload exceeds 64 MiB"); + chunks.push(chunk); + } + const payload = Buffer.concat(chunks); const res = await this.send("PUT", [...dir, STAGING_DIR, stagingName], signal, { - body, - headers: { "content-type": "application/octet-stream" }, + body: payload, + headers: { + "content-type": "application/octet-stream", + // Buffer bodies make undici set Content-Length itself; restating it + // keeps the contract explicit for the oCIS gate that rejects chunked. + "content-length": String(payload.byteLength), + }, }); if (res.status !== 201 && res.status !== 204) throw new WebDavError(res.status, `WebDAV PUT failed: ${res.status}`); } @@ -100,7 +124,7 @@ export class OcisWebDav implements LibraryPort { method: string, segments: string[], signal: AbortSignal, - init: { body?: ReadableStream | string; headers?: Record } = {}, + init: { body?: ReadableStream | Buffer | string; headers?: Record } = {}, ): Promise { const res = await fetch(new URL(davPath(this.user, segments), this.baseUrl), { method, diff --git a/infra/occupymedia/edge/edge-firewall.sh b/infra/occupymedia/edge/edge-firewall.sh index 8494f77..ba599ab 100755 --- a/infra/occupymedia/edge/edge-firewall.sh +++ b/infra/occupymedia/edge/edge-firewall.sh @@ -15,6 +15,14 @@ CF6_URL="https://www.cloudflare.com/ips-v6" iptables -F DOCKER-USER ip6tables -F DOCKER-USER 2>/dev/null || true +# DOCKER-USER sees forwarded traffic in BOTH directions. Exempt everything +# originating from a docker bridge (container egress) BEFORE any port rules — +# otherwise the CF-range catch-all below also drops containers' outbound 443 +# (Matrix/R2/platform calls). Inbound filtering by source is unaffected. +for br in $(ls /sys/class/net/ | grep -E '^(br-|docker0)'); do + iptables -I DOCKER-USER 1 -i "$br" -j RETURN +done + if [[ $MODE == OPEN ]]; then echo "edge firewall: OPEN (any source allowed on 80/443; nginx SNI gate applies)" exit 0 diff --git a/infra/occupymedia/files/compose.yml b/infra/occupymedia/files/compose.yml index 00360ee..eafcc9c 100644 --- a/infra/occupymedia/files/compose.yml +++ b/infra/occupymedia/files/compose.yml @@ -20,6 +20,9 @@ services: HOME: /var/lib/ocis OCIS_CONFIG_DIR: /etc/ocis OCIS_URL: ${FILES_PUBLIC_URL:?} + # The gateway advertises OCIS_URL/data as the upload endpoint to ocdav; + # keep that hop on-loopback so writes survive public-edge outages. + GATEWAY_FRONTEND_PUBLIC_URL: http://localhost:9200 OCIS_INSECURE: "false" OCIS_JWT_SECRET: ${OCIS_JWT_SECRET:?} IDM_ADMIN_PASSWORD: ${OCIS_ADMIN_PASSWORD:?}